Note that it's easy to implement an object implementing both `!RowFactory` and
`!AsyncRowFactory`: usually, everything you need to implement a row factory is
to access `~Cursor.description`, which is provided by both the cursor flavours.
-The `psycopg` module also exposes a class `AnyCursor` which you may use if you
-want to use the same row factory for both sync and async cursors.
`~RowFactory` objects can be implemented as a class, for instance:
.. currentmodule:: psycopg.rows
-.. autofunction:: tuple_row(cursor: AnyCursor[TupleRow])
+.. autofunction:: tuple_row
.. autodata:: TupleRow
-.. autofunction:: dict_row(cursor: AnyCursor[DictRow])
+.. autofunction:: dict_row
.. autodata:: DictRow
-.. autofunction:: namedtuple_row(cursor: AnyCursor[NamedTuple])
+.. autofunction:: namedtuple_row
Use with a static analyzer
from . import postgres
from .copy import Copy, AsyncCopy
from ._enums import IsolationLevel
-from .cursor import AnyCursor, AsyncCursor, Cursor
+from .cursor import AsyncCursor, Cursor
from .errors import Warning, Error, InterfaceError, DatabaseError
from .errors import DataError, OperationalError, IntegrityError
from .errors import InternalError, ProgrammingError, NotSupportedError
# so that function signatures are consistent with the documentation.
__all__ = [
"__version__",
- "AnyCursor",
"AsyncConnection",
"AsyncCopy",
"AsyncCursor",
from . import errors as e
if TYPE_CHECKING:
- from .cursor import AnyCursor
+ from .cursor import BaseCursor
class ColumnData(NamedTuple):
__module__ = "psycopg"
- def __init__(self, cursor: "AnyCursor[Any]", index: int):
+ def __init__(self, cursor: "BaseCursor[Any, Any]", index: int):
res = cursor.pgresult
assert res
if TYPE_CHECKING:
from .abc import Transformer
from .pq.abc import PGconn, PGresult
- from .connection import BaseConnection # noqa: F401
- from .connection import Connection, AsyncConnection # noqa: F401
+ from .connection import Connection, AsyncConnection
execute: Callable[["PGconn"], PQGen[List["PGresult"]]]
self._closed = True
-AnyCursor = BaseCursor[Any, Row]
-
-
-C = TypeVar("C", bound="BaseCursor[Any, Any]")
+AnyCursor = TypeVar("AnyCursor", bound="BaseCursor[Any, Any]")
class Cursor(BaseCursor["Connection[Any]", Row]):
super().__init__(connection)
self._row_factory = row_factory
- def __enter__(self: C) -> C:
+ def __enter__(self: AnyCursor) -> AnyCursor:
return self
def __exit__(
return self._row_factory(self)
def execute(
- self: C,
+ self: AnyCursor,
query: Query,
params: Optional[Params] = None,
*,
prepare: Optional[bool] = None,
- ) -> C:
+ ) -> AnyCursor:
"""
Execute a query or command to the database.
"""
super().__init__(connection)
self._row_factory = row_factory
- async def __aenter__(self: C) -> C:
+ async def __aenter__(self: AnyCursor) -> AnyCursor:
return self
async def __aexit__(
return self._row_factory(self)
async def execute(
- self: C,
+ self: AnyCursor,
query: Query,
params: Optional[Params] = None,
*,
prepare: Optional[bool] = None,
- ) -> C:
+ ) -> AnyCursor:
try:
async with self._conn.lock:
await self._conn.wait(
import functools
import re
from collections import namedtuple
-from typing import Any, Callable, Dict, NamedTuple, Sequence, Tuple, Type
-from typing import TypeVar, TYPE_CHECKING
+from typing import Any, Dict, NamedTuple, Sequence, Tuple, Type, TypeVar
+from typing import TYPE_CHECKING
from . import errors as e
from .compat import Protocol
if TYPE_CHECKING:
- from .cursor import AnyCursor, Cursor, AsyncCursor
+ from .cursor import BaseCursor, Cursor, AsyncCursor
# Row factories
def tuple_row(
- cursor: "AnyCursor[TupleRow]",
-) -> Callable[[Sequence[Any]], TupleRow]:
+ cursor: "BaseCursor[Any, TupleRow]",
+) -> RowMaker[TupleRow]:
r"""Row factory to represent rows as simple tuples.
This is the default factory.
- :param cursor: The cursor where the rows are read.
- :rtype: `RowMaker`\ [`TupleRow`]
+ :param cursor: The cursor where to read from.
"""
# Implementation detail: make sure this is the tuple type itself, not an
# equivalent function, because the C code fast-paths on it.
def dict_row(
- cursor: "AnyCursor[DictRow]",
-) -> Callable[[Sequence[Any]], DictRow]:
+ cursor: "BaseCursor[Any, DictRow]",
+) -> RowMaker[DictRow]:
r"""Row factory to represent rows as dicts.
Note that this is not compatible with the DBAPI, which expects the records
to be sequences.
- :param cursor: The cursor where the rows are read.
- :rtype: `RowMaker`\ [`DictRow`]
+ :param cursor: The cursor where to read from.
"""
def make_row(values: Sequence[Any]) -> Dict[str, Any]:
def namedtuple_row(
- cursor: "AnyCursor[NamedTuple]",
-) -> Callable[[Sequence[Any]], NamedTuple]:
+ cursor: "BaseCursor[Any, NamedTuple]",
+) -> RowMaker[NamedTuple]:
r"""Row factory to represent rows as `~collections.namedtuple`.
- :param cursor: The cursor where the rows are read.
- :rtype: `RowMaker`\ [`NamedTuple`]
+ :param cursor: The cursor where to read from.
"""
def make_row(values: Sequence[Any]) -> NamedTuple:
from . import errors as e
from .abc import ConnectionType, Query, Params, PQGen
from .rows import Row, RowFactory, AsyncRowFactory
-from .cursor import C, BaseCursor, Cursor, AsyncCursor, execute
+from .cursor import AnyCursor, BaseCursor, Cursor, AsyncCursor, execute
if TYPE_CHECKING:
- from .connection import BaseConnection # noqa: F401
- from .connection import Connection, AsyncConnection # noqa: F401
+ from .connection import Connection, AsyncConnection
DEFAULT_ITERSIZE = 100
super().close()
def execute(
- self: C,
+ self: AnyCursor,
query: Query,
params: Optional[Params] = None,
**kwargs: Any,
- ) -> C:
+ ) -> AnyCursor:
"""
Open a cursor to execute a query to the database.
"""
await super().close()
async def execute(
- self: C,
+ self: AnyCursor,
query: Query,
params: Optional[Params] = None,
**kwargs: Any,
- ) -> C:
+ ) -> AnyCursor:
if kwargs:
raise TypeError(f"keyword not supported: {list(kwargs)[0]}")
helper = cast(AsyncServerCursor[Row], self)._helper
stmts = "\n".join(f" {line}" for line in stmts.splitlines())
src = f"""\
-from typing import Any, Callable, Dict, List, NamedTuple, Optional, Sequence, Tuple
+from typing import Any, Callable, Dict, List, NamedTuple, Optional, Sequence
+from typing import Tuple, Union
import psycopg
from psycopg import rows
self.kwargs = kwargs
def thing_row(
- cur: psycopg.AnyCursor[Thing],
+ cur: Union[psycopg.Cursor[Thing], psycopg.AsyncCursor[Thing]],
) -> Callable[[Sequence[Any]], Thing]:
assert cur.description
names = [d.name for d in cur.description]
projects / thirdparty / psycopg.git / commitdiff
