1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
|
"""
Support module for test_cursor[_async].py
"""
from __future__ import annotations
import re
from typing import Any
import pytest
import psycopg
from psycopg.rows import RowMaker
@pytest.fixture(scope="session")
def _execmany(svcconn):
svcconn.execute(
"""
drop table if exists execmany;
create table execmany (id serial primary key, num integer, data text)
"""
)
@pytest.fixture(scope="function")
def execmany(svcconn, _execmany):
svcconn.execute("delete from execmany")
def ph(cur: Any, query: str) -> str:
"""Change placeholders in a query from %s to $n if testing a raw cursor"""
from psycopg.raw_cursor import RawCursorMixin
if not isinstance(cur, RawCursorMixin):
return query
if "%(" in query:
pytest.skip("RawCursor only supports positional placeholders")
n = 1
def s(m: re.Match[str]) -> str:
nonlocal n
rv = f"${n}"
n += 1
return rv
return re.sub(r"(?<!%)(%[bst])", s, query)
def my_row_factory(
cursor: psycopg.Cursor[list[str]] | psycopg.AsyncCursor[list[str]],
) -> RowMaker[list[str]]:
if cursor.description is not None:
titles = [c.name for c in cursor.description]
def mkrow(values):
return [f"{value.upper()}{title}" for title, value in zip(titles, values)]
return mkrow
else:
return psycopg.rows.no_result
|
