=============
Cursor's `fetch*` methods return tuples of column values by default. This can
-be changed to adapt the needs of the programmer by using custom row factories.
+be changed to adapt the needs of the programmer by using custom *row
+factories*.
-A row factory is a callable that accepts a cursor object and returns another
-callable accepting a `values` tuple and returning a row in the desired form.
-This can be implemented as a class, for instance:
+A row factory (formally implemented by the `~psycopg3.rows.RowFactory`
+protocol) is a callable that accepts a `Cursor` object and returns another
+callable (formally the `~psycopg3.rows.RowMaker`) accepting a `values` tuple
+and returning a row in the desired form.
+
+.. autoclass:: psycopg3.rows.RowMaker
+
+ .. automethod:: __call__
+
+.. autoclass:: psycopg3.rows.RowFactory
+
+ .. automethod:: __call__
+
+`~RowFactory` objects can be implemented as a class, for instance:
.. code:: python
the `row_factory` specified at `Connection.connect()` can be overridden by
passing another value at `Connection.cursor()`.
-.. note:: By declaring type annotations on the row factory used in
- `Connection` or `Cursor`, rows retrieved by `.fetch*()` calls will have the
- correct return type (i.e. a `dict[str, Any]` in previous example) and your
- code can be type checked with a static analyzer such as mypy.
-
Available row factories
-----------------------
.. currentmodule:: psycopg3.rows
.. autofunction:: tuple_row
+.. autodata:: TupleRow
+
.. autofunction:: dict_row
+.. autodata:: DictRow
+
.. autofunction:: namedtuple_row
Use with a static analyzer
--------------------------
-The `Connection` and `Cursor` classes are parametric types: the parameter
-`!Row` is passed by the ``row_factory`` argument (of the
-`~Connection.connect()` and the `~Connection.cursor()` method) and it controls
-what type of record is returned by the fetch methods of the cursors. The
-default `tuple_row()` returns a generic tuple as return type (`Tuple[Any,
-...]`). This information can be used for type checking using a static
-analyzer such as Mypy_.
+The `~psycopg3.Connection` and `~psycopg3.Cursor` classes are `generic
+types`__: the parameter `!Row` is passed by the ``row_factory`` argument (of
+the `~Connection.connect()` and the `~Connection.cursor()` method) and it
+controls what type of record is returned by the fetch methods of the cursors.
+The default `tuple_row()` returns a generic tuple as return type (`Tuple[Any,
+...]`). This information can be used for type checking using a static analyzer
+such as Mypy_.
.. _Mypy: https://mypy.readthedocs.io/
+.. __: https://mypy.readthedocs.io/en/stable/generics.html
.. code:: python
from . import pq
from . import errors as e
from .oids import INVALID_OID
-from .proto import LoadFunc, AdaptContext, Row, RowMaker
+from .rows import Row, RowMaker
+from .proto import LoadFunc, AdaptContext
from ._enums import Format
if TYPE_CHECKING:
from . import encodings
from .pq import ConnStatus, ExecStatus, TransactionStatus, Format
from .sql import Composable
-from .rows import tuple_row, TupleRow
+from .rows import Row, RowFactory, tuple_row, TupleRow
from .proto import AdaptContext, ConnectionType, Params, PQGen, PQGenConn
-from .proto import Query, Row, RowFactory, RV
+from .proto import Query, RV
from .cursor import Cursor, AsyncCursor
from .conninfo import make_conninfo, ConnectionInfo
from .generators import notifies
from .pq import ExecStatus, Format
from .copy import Copy, AsyncCopy
+from .rows import Row, RowFactory
from .proto import ConnectionType, Query, Params, PQGen
-from .proto import Row, RowFactory
from ._column import Column
from ._queries import PostgresQuery
from ._preparing import Prepare
from ._enums import Format
if TYPE_CHECKING:
- from .connection import BaseConnection
- from .cursor import AnyCursor
+ from .sql import Composable
+ from .rows import Row, RowMaker
from .adapt import Dumper, Loader, AdaptersMap
from .waiting import Wait, Ready
- from .sql import Composable
+ from .connection import BaseConnection
# An object implementing the buffer protocol
Buffer = Union[bytes, bytearray, memoryview]
"""
-# Row factories
-
-Row = TypeVar("Row")
-Row_co = TypeVar("Row_co", covariant=True)
-
-
-class RowMaker(Protocol[Row_co]):
- def __call__(self, __values: Sequence[Any]) -> Row_co:
- ...
-
-
-class RowFactory(Protocol[Row]):
- def __call__(self, __cursor: "AnyCursor[Row]") -> RowMaker[Row]:
- ...
-
-
# Adaptation types
DumpFunc = Callable[[Any], bytes]
...
def load_rows(
- self, row0: int, row1: int, make_row: RowMaker[Row]
- ) -> List[Row]:
+ self, row0: int, row1: int, make_row: "RowMaker[Row]"
+ ) -> List["Row"]:
...
- def load_row(self, row: int, make_row: RowMaker[Row]) -> Optional[Row]:
+ def load_row(self, row: int, make_row: "RowMaker[Row]") -> Optional["Row"]:
...
def load_sequence(
import re
from collections import namedtuple
from typing import Any, Callable, Dict, NamedTuple, Sequence, Tuple, Type
-from typing import TYPE_CHECKING
+from typing import TypeVar, TYPE_CHECKING
+from typing_extensions import Protocol
from . import errors as e
if TYPE_CHECKING:
from .cursor import AnyCursor
+# Row factories
+
+Row = TypeVar("Row")
+Row_co = TypeVar("Row_co", covariant=True)
+
+
+class RowMaker(Protocol[Row_co]):
+ """
+ Callable protocol taking a sequence of value and returning an object.
+
+ The sequence of value is what is returned from a database query, already
+ adapted to the right Python types. Tye return value is the object that your
+ program would like to receive: By defaut (`tuple_row()`) it is a simple
+ tuple, but it may be any type of object.
+
+ Typically, `~RowMaker` functions are returned by `RowFactory`.
+ """
+
+ def __call__(self, __values: Sequence[Any]) -> Row_co:
+ """
+ Convert a sequence of values from the database to a finished object.
+ """
+ ...
+
+
+class RowFactory(Protocol[Row]):
+ """
+ Callable protocol taking a `~psycopg3.Cursor` and returning a `RowMaker`.
+
+ A `!RowFactory` is typically called when a `!Cursor` receives a result.
+ This way it can inspect the cursor state (for instance the
+ `~psycopg3.Cursor.description` attribute) and help a `!RowMaker` to create
+ a complete object.
+
+ For instance the `dict_row()` `!RowFactory` uses the names of the column to
+ define the dictionary key and returns a `!RowMaker` function which would
+ use the values to create a dictionary for each record.
+ """
+
+ def __call__(self, __cursor: "AnyCursor[Row]") -> RowMaker[Row]:
+ """
+ Inspect the result on a cursor and return a `RowMaker` to convert rows.
+ """
+ ...
+
TupleRow = Tuple[Any, ...]
+"""
+An alias for the type returned by `tuple_row()` (i.e. a tuple of any content).
+"""
def tuple_row(
cursor: "AnyCursor[TupleRow]",
) -> Callable[[Sequence[Any]], TupleRow]:
- """Row factory to represent rows as simple tuples.
+ r"""Row factory to represent rows as simple tuples.
This is the default factory.
+
+ :rtype: `RowMaker`\ [`TupleRow`]
"""
# Implementation detail: make sure this is the tuple type itself, not an
# equivalent function, because the C code fast-paths on it.
DictRow = Dict[str, Any]
+"""
+An alias for the type returned by `dict_row()`
+
+A `!DictRow` is a dictionary with keys as string and any value returned by the
+database.
+"""
def dict_row(
cursor: "AnyCursor[DictRow]",
) -> Callable[[Sequence[Any]], DictRow]:
- """Row factory to represent rows as dicts.
+ r"""Row factory to represent rows as dicts.
Note that this is not compatible with the DBAPI, which expects the records
to be sequences.
+
+ :rtype: `RowMaker`\ [`DictRow`]
"""
def make_row(values: Sequence[Any]) -> Dict[str, Any]:
def namedtuple_row(
cursor: "AnyCursor[NamedTuple]",
) -> Callable[[Sequence[Any]], NamedTuple]:
- """Row factory to represent rows as `~collections.namedtuple`."""
+ r"""Row factory to represent rows as `~collections.namedtuple`.
+
+ :rtype: `RowMaker`\ [`NamedTuple`]
+ """
def make_row(values: Sequence[Any]) -> NamedTuple:
desc = cursor.description
from . import pq
from . import sql
from . import errors as e
+from .rows import Row, RowFactory
+from .proto import ConnectionType, Query, Params, PQGen
from .cursor import BaseCursor, execute
-from .proto import ConnectionType, Query, Params, PQGen, Row, RowFactory
if TYPE_CHECKING:
from typing import Any # noqa: F401
from typing import Any, Iterable, List, Optional, Sequence, Tuple
+from psycopg3 import pq
from psycopg3 import proto
+from psycopg3.rows import Row, RowMaker
from psycopg3.adapt import Dumper, Loader, AdaptersMap, Format
-from psycopg3.connection import BaseConnection
-from psycopg3 import pq
from psycopg3.pq.proto import PGconn, PGresult
+from psycopg3.connection import BaseConnection
class Transformer(proto.AdaptContext):
def __init__(self, context: Optional[proto.AdaptContext] = None): ...
) -> Tuple[List[Any], Tuple[int, ...], Sequence[pq.Format]]: ...
def get_dumper(self, obj: Any, format: Format) -> Dumper: ...
def load_rows(
- self, row0: int, row1: int, make_row: proto.RowMaker[proto.Row]
- ) -> List[proto.Row]: ...
- def load_row(
- self, row: int, make_row: proto.RowMaker[proto.Row]
- ) -> Optional[proto.Row]: ...
+ self, row0: int, row1: int, make_row: RowMaker[Row]
+ ) -> List[Row]: ...
+ def load_row(self, row: int, make_row: RowMaker[Row]) -> Optional[Row]: ...
def load_sequence(
self, record: Sequence[Optional[bytes]]
) -> Tuple[Any, ...]: ...
from psycopg3 import errors as e
from psycopg3._enums import Format as Pg3Format
from psycopg3.pq import Format as PqFormat
-from psycopg3.proto import Row, RowMaker
+from psycopg3.rows import Row, RowMaker
# internal structure: you are not supposed to know this. But it's worth some
# 10% of the innermost loop, so I'm willing to ask for forgiveness later...
projects / thirdparty / psycopg.git / commitdiff
