Row factories

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.

A row factory (formally implemented by the RowFactory protocol) is a callable that accepts a Cursor object and returns another callable (formally the RowMaker protocol) accepting a values tuple and returning a row in the desired form.

class psycopg.rows.RowMaker

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. The return value is the object that your program would like to receive: by default (tuple_row()) it is a simple tuple, but it may be any type of object.

Typically, RowMaker functions are returned by RowFactory.

__call__(values: Sequence[Any]) Row

Convert a sequence of values from the database to a finished object.

class psycopg.rows.RowFactory

Callable protocol taking a 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 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.

__call__(cursor: AnyCursor[Row]) RowMaker[Row]

Inspect the result on a cursor and return a RowMaker to convert rows.

AnyCursor may be either a Cursor or an AsyncCursor.

RowFactory objects can be implemented as a class, for instance:

from typing import Any, Sequence
from psycopg import AnyCursor

class DictRowFactory:
    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]:
        return dict(zip(self.fields, values))

or as a plain function:

def dict_row_factory(
    cursor: AnyCursor[dict[str, Any]]
) -> Callable[[Sequence[Any]], dict[str, Any]]:
    fields = [c.name for c in cursor.description]

    def make_row(values: Sequence[Any]) -> dict[str, Any]:
        return dict(zip(fields, values))

    return make_row

These can then be used by specifying a row_factory argument in Connection.connect(), Connection.cursor(), or by writing to Connection.row_factory attribute.

conn = psycopg.connect(row_factory=DictRowFactory)
cur = conn.execute("SELECT first_name, last_name, age FROM persons")
person = cur.fetchone()
print(f"{person['first_name']} {person['last_name']}")

Later usages of row_factory override earlier definitions; for instance, the row_factory specified at Connection.connect() can be overridden by passing another value at Connection.cursor().

Available row factories

The module psycopg.rows provides the implementation for a few row factories:

psycopg.rows.tuple_row(cursor)

Row factory to represent rows as simple tuples.

This is the default factory.

Parameters

cursor (AnyCursor[TupleRow]) – The cursor where the rows are read.

Return type

RowMaker[TupleRow]

psycopg.rows.TupleRow

An alias for the type returned by tuple_row() (i.e. a tuple of any content).

alias of Tuple[Any, …]

psycopg.rows.dict_row(cursor)

Row factory to represent rows as dicts.

Note that this is not compatible with the DBAPI, which expects the records to be sequences.

Parameters

cursor (AnyCursor[DictRow]) – The cursor where the rows are read.

Return type

RowMaker[DictRow]

psycopg.rows.DictRow

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.

alias of Dict[str, Any]

psycopg.rows.namedtuple_row(cursor)

Row factory to represent rows as namedtuple.

Parameters

cursor (AnyCursor[NamedTuple]) – The cursor where the rows are read.

Return type

RowMaker[NamedTuple]

Use with a static analyzer

The Connection and Cursor classes are generic types: the parameter Row is passed by the row_factory argument (of the connect() and the 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.

conn = psycopg.connect()
# conn type is psycopg.Connection[Tuple[Any, ...]]

dconn = psycopg.connect(row_factory=dict_row)
# dconn type is psycopg.Connection[Dict[str, Any]]

cur = conn.cursor()
# cur type is psycopg.Cursor[Tuple[Any, ...]]

dcur = conn.cursor(row_factory=dict_row)
dcur = dconn.cursor()
# dcur type is psycopg.Cursor[Dict[str, Any]] in both cases

rec = cur.fetchone()
# rec type is Optional[Tuple[Any, ...]]

drec = dcur.fetchone()
# drec type is Optional[Dict[str, Any]]

Example: returning records as Pydantic models

Using Pydantic it is possible to enforce static typing at runtime. Using a Pydantic model factory the code can be checked statically using Mypy and querying the database will raise an exception if the resultset is not compatible with the model.

The following example can be checked with mypy --strict without reporting any issue. Pydantic will also raise a runtime error in case the PersonFactory is used with a query that returns incompatible data.

from datetime import date
from typing import Any, Optional, Sequence

import psycopg
from pydantic import BaseModel

class Person(BaseModel):
    id: int
    first_name: str
    last_name: str
    dob: Optional[date]

class PersonFactory:
    def __init__(self, cur: psycopg.AnyCursor[Person]):
        assert cur.description
        self.fields = [c.name for c in cur.description]

    def __call__(self, values: Sequence[Any]) -> Person:
        return Person(**dict(zip(self.fields, values)))

def fetch_person(id: int) -> Person:
    conn = psycopg.connect()
    cur = conn.cursor(row_factory=PersonFactory)
    cur.execute(
        """
        SELECT id, first_name, last_name, dob
        FROM (VALUES
            (1, 'John', 'Doe', '2000-01-01'::date),
            (2, 'Jane', 'White', NULL)
        ) AS data (id, first_name, last_name, dob)
        WHERE id = %(id)s;
        """,
        {"id": id},
    )
    rec = cur.fetchone()
    if not rec:
        raise KeyError(f"person {id} not found")
    return rec

for id in [1, 2]:
    p = fetch_person(id)
    if p.dob:
        print(f"{p.first_name} was born in {p.dob.year}")
    else:
        print(f"Who knows when {p.first_name} was born")