.. code:: python
from typing import Any, Sequence
- from psycopg3 import BaseCursor
+ from psycopg3 import AnyCursor
class DictRowFactory:
- def __init__(self, cursor: BaseCursor[Any, dict[str, Any]]):
+ def __init__(self, cursor: AnyCursor[dict[str, Any]]):
self.fields = [c.name for c in cursor.description]
def __call__(self, values: Sequence[Any]) -> dict[str, Any]:
.. code:: python
def dict_row_factory(
- cursor: BaseCursor[Any, dict[str, Any]]
+ cursor: AnyCursor[dict[str, Any]]
) -> Callable[[Sequence[Any]], dict[str, Any]]:
fields = [c.name for c in cursor.description]
from . import types
from .copy import Copy, AsyncCopy
from .adapt import global_adapters
-from .cursor import AsyncCursor, Cursor, BaseCursor
+from .cursor import AnyCursor, 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",
"AsyncServerCursor",
"AsyncTransaction",
"BaseConnection",
- "BaseCursor",
"Column",
"Connection",
"Copy",
from . import errors as e
if TYPE_CHECKING:
- from .cursor import BaseCursor
+ from .cursor import AnyCursor
class ColumnData(NamedTuple):
__module__ = "psycopg3"
- def __init__(self, cursor: "BaseCursor[Any, Any]", index: int):
+ def __init__(self, cursor: "AnyCursor[Any]", index: int):
res = cursor.pgresult
assert res
self._pgq = pgq
+AnyCursor = BaseCursor[Any, Row]
+
+
class Cursor(BaseCursor["Connection[Any]", Row]):
__module__ = "psycopg3"
__slots__ = ()
if TYPE_CHECKING:
from .connection import BaseConnection
- from .cursor import BaseCursor
+ from .cursor import AnyCursor
from .adapt import Dumper, Loader, AdaptersMap
from .waiting import Wait, Ready
from .sql import Composable
class RowFactory(Protocol[Row]):
- def __call__(self, __cursor: "BaseCursor[Any, Row]") -> RowMaker[Row]:
+ def __call__(self, __cursor: "AnyCursor[Row]") -> RowMaker[Row]:
...
from . import errors as e
if TYPE_CHECKING:
- from .cursor import BaseCursor
+ from .cursor import AnyCursor
TupleRow = Tuple[Any, ...]
def tuple_row(
- cursor: "BaseCursor[Any, TupleRow]",
+ cursor: "AnyCursor[TupleRow]",
) -> Callable[[Sequence[Any]], TupleRow]:
"""Row factory to represent rows as simple tuples.
def dict_row(
- cursor: "BaseCursor[Any, DictRow]",
+ cursor: "AnyCursor[DictRow]",
) -> Callable[[Sequence[Any]], DictRow]:
"""Row factory to represent rows as dicts.
def namedtuple_row(
- cursor: "BaseCursor[Any, NamedTuple]",
+ cursor: "AnyCursor[NamedTuple]",
) -> Callable[[Sequence[Any]], NamedTuple]:
"""Row factory to represent rows as `~collections.namedtuple`."""
self.kwargs = kwargs
def thing_row(
- cur: psycopg3.BaseCursor[Any, Thing],
+ cur: psycopg3.AnyCursor[Thing],
) -> Callable[[Sequence[Any]], Thing]:
assert cur.description
names = [d.name for d in cur.description]
from dataclasses import dataclass
from typing import Any, Callable, Optional, Sequence, Tuple
-from psycopg3 import BaseCursor, Connection, Cursor, ServerCursor, connect
+from psycopg3 import AnyCursor, Connection, Cursor, ServerCursor, connect
-def int_row_factory(
- cursor: BaseCursor[Any, int]
-) -> Callable[[Sequence[int]], int]:
+def int_row_factory(cursor: AnyCursor[int]) -> Callable[[Sequence[int]], int]:
return lambda values: values[0] if values else 42
@classmethod
def row_factory(
- cls, cursor: BaseCursor[Any, Person]
+ cls, cursor: AnyCursor[Person]
) -> Callable[[Sequence[str]], Person]:
def mkrow(values: Sequence[str]) -> Person:
name, address = values