API reference¶
Reference documentation for the public fastpg API, generated from its docstrings.
fastpg
¶
fastpg: async PostgreSQL client for Python, implemented in Rust.
async with await fastpg.connect(dsn) as conn:
person = await conn.select_one(
"SELECT name, age FROM people WHERE id = $1", 42, into=Person
)
# or, on Python 3.14+, as a t-string (interpolations become bound parameters, never SQL text):
person = await conn.select_one(
t"SELECT name, age FROM people WHERE id = {person_id}", into=Person
)
Error = FastPGError
module-attribute
¶
DB-API 2.0 (PEP 249) name for FastPGError, the base class of every fastpg error.
__version__ = '0.1.1'
module-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
IsolationLevel = Literal['read_uncommitted', 'read_committed', 'repeatable_read', 'serializable']
¶
Transaction isolation level for transaction(isolation=...).
None (the default there) keeps the server's default_transaction_isolation.
NamedParams = msgspec.Struct | DataclassInstance | BaseModel | Mapping[str, object]
¶
What params= takes: values are read from fields (Structs, dataclasses, pydantic models,
by Python attribute name) or keys (mappings).
Strategy = Literal['auto', 'direct', 'msgpack', 'json']
¶
How rows become into= objects (msgspec.Struct or pydantic BaseModel).
"direct": Rust decodes each column and builds the object itself: Structs via their__init__(vectorcall, no intermediate containers), pydantic models likemodel_construct(__dict__+ slots, no validation). Field types are checked against column types once per statement. Not available when the type needs validation (constraints, validators, enums, literals, ...)."msgpack"(msgspec only): Rust encodes rows as MessagePack; a cachedmsgspec.msgpack.Decoderbuilds and fully validates them."json": Rust encodes rows as JSON;msgspec.json.Decoderor pydantic-core'svalidate_jsonbuilds and fully validates them."auto"(default):"direct"when possible, otherwise"msgpack"(msgspec) or"json"(pydantic).
into=dict and into= a NamedTuple are always built directly; "msgpack" and
"json" raise UnsupportedTypeError for them.
Connection
dataclass
¶
Bases: _Queries
A single PostgreSQL connection bound to the event loop it was opened on.
Run queries on it directly: await conn.select_one(...). Statements are prepared on
first use and cached per connection (keyed by SQL text); results use PostgreSQL's binary
format. A query is SQL with $1, $2, ... placeholders plus positional parameters, SQL
with :name placeholders plus params= (a Struct, dataclass, pydantic model or
mapping), or a t-string whose interpolations become the parameters
(t"... WHERE id = {pid}"); see fastpg.Identifier for interpolating names.
Concurrent queries from multiple tasks are pipelined over the one connection.
Use as an async context manager, or call close() (terminate() to close at once).
A query whose caller gives up (a cancelled task, asyncio.timeout) is cancelled on the
server too if it is the only one in flight on the connection; the connection's next
queries wait until that cancel is done.
cached_statements
property
¶
Number of server-side prepared statements currently cached for this connection.
close()
async
¶
Close gracefully: in-flight queries complete, then the session is terminated.
The Terminate message is sent and the socket closed without waiting for the server
to answer. If the call is cancelled (e.g. by asyncio.timeout), the connection
is closed at once, as by terminate().
copy_in(table, rows, /, *, columns=None)
async
¶
Load rows with binary COPY ... FROM STDIN; return the number of rows copied.
The fastest way to load many rows. Same row kinds and columns rules as
insert_many; values are encoded in Rust for the table's column types and streamed
in chunks. All-or-nothing, like execute_many.
execute(query, /, *args, params=None)
async
¶
Run a statement (prepared, with parameters) and return the affected row count.
execute_many(query, rows)
async
¶
Run query once per row of positional parameters; return the total row count.
One prepared statement; rows are encoded in Rust and pipelined (no round trip per
row). All-or-nothing: on an error (exc.row_index names the failing row) or
cancellation nothing is written, and an open transaction is left as it was.
Concurrent queries on this connection wait until the call completes.
execute_script(query)
async
¶
Run one or more ;-separated statements without parameters (simple protocol).
A t-string may splice in Identifiers and fragments, but has no value parameters.
insert_many(table, rows, /, *, columns=None)
async
¶
INSERT rows into table (pipelined, like execute_many); return the count.
Rows are msgspec Structs or pydantic models (columns: their encoded field names /
aliases), mappings (columns: the first row's keys) or sequences (columns: the
table's, in order). columns selects/orders the columns explicitly. table is
"name", "schema.name" or a tuple of parts; names are quoted (exact match).
select_and_group(query, /, *args, params=None, into, strategy='auto')
async
¶
Return the rows grouped into into Structs: one per group of rows, in the order
each group's first row arrives.
Rows with equal values in all of into's fields that aren't lists form a group.
Each list[X] field collects one value per row of its group, in row order, from
the column of the same name. A list[S] field, S a Struct, collects one S
per row, from the columns named <field>__<S field> (e.g. items__sku); a row
where all of those are NULL (a LEFT JOIN without a match) adds none.
Columns match fields as for select_many(..., into=...), with the same errors;
a NULL fits a list field only if its item type allows None. Raises TypeError
before running the query if into isn't a Struct with a list field, and when a
group key value is unhashable.
select_many(query, /, *args, params=None, into=None, strategy='auto')
async
¶
select_many(
query: str | Template,
/,
*args: object,
params: NamedParams | None = None,
into: type[dict[Any, Any]],
strategy: Strategy = "auto",
) -> list[dict[str, Any]]
Return every row.
select_one(query, /, *args, params=None, into=None, strategy='auto')
async
¶
select_one(
query: str | Template,
/,
*args: object,
params: NamedParams | None = None,
into: type[dict[Any, Any]],
strategy: Strategy = "auto",
) -> dict[str, Any]
Return exactly one row.
Raises NoRowsError for an empty result and TooManyRowsError for more than one row.
select_one_or_none(query, /, *args, params=None, into=None, strategy='auto')
async
¶
select_one_or_none(
query: str | Template,
/,
*args: object,
params: NamedParams | None = None,
into: type[dict[Any, Any]],
strategy: Strategy = "auto",
) -> dict[str, Any] | None
Return the only row, or None for an empty result (TooManyRowsError for >1 row).
terminate()
¶
Close at once: pending and in-flight queries fail with ConnectionClosedError.
Sends nothing to the server, which ends the session when it sees the socket close.
Safe to call more than once, after close(), or on a connection already lost. Call
it on the connection's event loop.
transaction(*, isolation=None, readonly=False, deferrable=False)
¶
Start a transaction: async with conn.transaction() as tx: await tx.execute(...).
Commits when the block exits normally and rolls back when it raises. Only one
transaction can be active per connection (nest with tx.savepoint()).
The connection is still shared while the block runs: queries other tasks send
through this Connection meanwhile are pipelined into the same session, so they run
inside the transaction. Keep a connection with an open transaction to one task, e.g.
via Pool.transaction() / Pool.acquire().
ConnectionClosedError
¶
DataError
¶
DatabaseError
¶
Bases: FastPGError
An error reported by the PostgreSQL server.
from_server(sqlstate, message, severity, detail, hint, position, schema_name, table_name, column_name, data_type_name, constraint_name)
classmethod
¶
Build the most specific subclass for a server ErrorResponse (called from Rust).
DecodeError
¶
FastPGError
¶
Bases: Exception
Base class for all fastpg errors.
row_index = None
class-attribute
instance-attribute
¶
For bulk writes (execute_many, insert_many, copy_in): the 0-based index of the
input row that failed, when known.
Identifier
¶
A quoted SQL identifier, spliced into a t-string query's text.
Parameters can only stand for values, so names (tables, columns, schemas, ...) that
vary at runtime go through Identifier:
t"SELECT * FROM {Identifier('my schema', 'people')}" gives
SELECT * FROM "my schema"."people". Each part is double-quoted with embedded quotes
doubled, so it's taken verbatim: case-sensitive, never a keyword or SQL syntax.
Put it where a bare name would go, never inside quotes.
t-strings need Python 3.14; on 3.13 Identifier exists but has no use.
quoted
property
¶
The quoted form spliced into the query, e.g. "my schema"."people".
IntegrityError
¶
InterfaceError
¶
InternalError
¶
MissingColumnError
¶
NoRowsError
¶
NotSupportedError
¶
NullValueError
¶
OperationalError
¶
ParameterError
¶
Bases: InterfaceError
A query parameter has no value (named parameters) or can't be encoded for its type.
Pool
dataclass
¶
A pool of connections to one database, bound to the event loop that created it.
Create with await fastpg.create_pool(dsn, ...); close with await pool.close() or
async with await fastpg.create_pool(...) as pool.
The configuration fields (dsn, min_size, ...) are read-only.
Shared statements (select_*, execute, execute_script) must not change session
state (SET, BEGIN, temp tables, ...): the session is shared with other callers
and isn't reset. Use acquire() or transaction() for that.
closed
property
¶
True once close() has been called.
acquire()
async
¶
Check out a connection for exclusive use: async with pool.acquire() as conn.
Waits (FIFO) for an idle connection; busy ones are drained of shared statements if
the pool can't grow. On exit the session is reset (settings, role, advisory locks,
cursors, LISTENs, temp tables; a transaction left open is rolled back; prepared
statements are kept) and the connection rejoins shared rotation. Don't keep using
conn after the block.
close()
async
¶
Close gracefully: new calls and waiters fail with ConnectionClosedError, shared statements in flight complete, and checked-out connections close when released.
If the call is cancelled (e.g. by asyncio.timeout), every connection still open,
checked-out ones included, is closed at once (Connection.terminate()), and the
cancellation is raised.
copy_in(table, rows, /, *, columns=None)
async
¶
Connection.copy_in on a connection checked out exclusively for the call.
Bulk writes hold their connection exclusively, so they never delay shared statements.
execute(query, /, *args, params=None)
async
¶
Shared mode Connection.execute: one autocommit statement; returns rows affected.
execute_many(query, rows)
async
¶
Connection.execute_many on a connection checked out exclusively for the call.
execute_script(query)
async
¶
Shared mode Connection.execute_script: runs as one implicit transaction.
insert_many(table, rows, /, *, columns=None)
async
¶
Connection.insert_many on a connection checked out exclusively for the call.
select_and_group(query, /, *args, params=None, into, strategy='auto')
async
¶
Shared mode Connection.select_and_group: the rows grouped into into Structs.
select_many(query, /, *args, params=None, into=None, strategy='auto')
async
¶
select_many(
query: str | Template,
/,
*args: object,
params: NamedParams | None = None,
into: type[dict[Any, Any]],
strategy: Strategy = "auto",
) -> list[dict[str, Any]]
Shared mode Connection.select_many: every row.
select_one(query, /, *args, params=None, into=None, strategy='auto')
async
¶
select_one(
query: str | Template,
/,
*args: object,
params: NamedParams | None = None,
into: type[dict[Any, Any]],
strategy: Strategy = "auto",
) -> dict[str, Any]
Shared mode Connection.select_one: exactly one row.
select_one_or_none(query, /, *args, params=None, into=None, strategy='auto')
async
¶
select_one_or_none(
query: str | Template,
/,
*args: object,
params: NamedParams | None = None,
into: type[dict[Any, Any]],
strategy: Strategy = "auto",
) -> dict[str, Any] | None
Shared mode Connection.select_one_or_none: the only row, or None.
stats()
¶
Current sizes, in-flight statements and waiters.
transaction(*, isolation=None, readonly=False, deferrable=False)
async
¶
Run a transaction on an exclusive connection: async with pool.transaction() as tx.
Commits on normal exit, rolls back if the block raises. The session reset runs in the COMMIT's round trip (before COMMIT, so a failure means nothing was committed).
PoolStats
dataclass
¶
A snapshot of a pool's state (Pool.stats()).
acquired
instance-attribute
¶
Connections checked out by acquire() / transaction().
idle
instance-attribute
¶
Shared connections with nothing in flight.
in_flight
instance-attribute
¶
Shared statements in flight (including cancelled ones still running server-side).
opening
instance-attribute
¶
Connections being opened.
size
instance-attribute
¶
Open connections, shared and acquired.
waiting
instance-attribute
¶
Shared statements waiting for a free slot.
waiting_acquire
instance-attribute
¶
acquire() / transaction() calls waiting for an idle connection.
ProgrammingError
¶
ResultError
¶
TooManyRowsError
¶
Transaction
¶
Bases: _Queries
A transaction on one connection, with the connection's query methods run inside it.
Use as async with conn.transaction() as tx: (or Pool.transaction()): BEGIN on
entry, COMMIT when the block exits normally, ROLLBACK when it raises (cancellation
included). async with tx.savepoint(): nests a savepoint that is rolled back, and the
exception re-raised, if its block raises.
COMMIT is preceded, in the same round trip, by a statement that fails when the
transaction is already aborted (an error was caught inside the block and the block went
on). PostgreSQL would turn that COMMIT into a silent ROLLBACK; here the block raises the
server's DatabaseError (SQLSTATE 25P02) instead, after rolling back.
Its query methods raise InterfaceError outside the block (before BEGIN, after COMMIT
or ROLLBACK): they would run outside the transaction, and once Pool.transaction()
has released the connection, maybe inside another caller's.
active
property
¶
True between BEGIN and COMMIT/ROLLBACK.
__init__(conn, *, isolation=None, readonly=False, deferrable=False, pre_commit='SELECT 1')
¶
Prefer Connection.transaction(); pre_commit runs right before COMMIT.
pre_commit must fail inside an aborted transaction (any statement does). The pool
passes its session-reset statements here, saving a round trip.
copy_in(table, rows, /, *, columns=None)
async
¶
Load rows with binary COPY ... FROM STDIN; return the number of rows copied.
The fastest way to load many rows. Same row kinds and columns rules as
insert_many; values are encoded in Rust for the table's column types and streamed
in chunks. All-or-nothing, like execute_many.
execute(query, /, *args, params=None)
async
¶
Run a statement (prepared, with parameters) and return the affected row count.
execute_many(query, rows)
async
¶
Run query once per row of positional parameters; return the total row count.
One prepared statement; rows are encoded in Rust and pipelined (no round trip per
row). All-or-nothing: on an error (exc.row_index names the failing row) or
cancellation nothing is written, and an open transaction is left as it was.
Concurrent queries on this connection wait until the call completes.
execute_script(query)
async
¶
Run one or more ;-separated statements without parameters (simple protocol).
A t-string may splice in Identifiers and fragments, but has no value parameters.
insert_many(table, rows, /, *, columns=None)
async
¶
INSERT rows into table (pipelined, like execute_many); return the count.
Rows are msgspec Structs or pydantic models (columns: their encoded field names /
aliases), mappings (columns: the first row's keys) or sequences (columns: the
table's, in order). columns selects/orders the columns explicitly. table is
"name", "schema.name" or a tuple of parts; names are quoted (exact match).
savepoint()
async
¶
Nest a savepoint: released on normal exit, rolled back to if the block raises.
select_and_group(query, /, *args, params=None, into, strategy='auto')
async
¶
Return the rows grouped into into Structs: one per group of rows, in the order
each group's first row arrives.
Rows with equal values in all of into's fields that aren't lists form a group.
Each list[X] field collects one value per row of its group, in row order, from
the column of the same name. A list[S] field, S a Struct, collects one S
per row, from the columns named <field>__<S field> (e.g. items__sku); a row
where all of those are NULL (a LEFT JOIN without a match) adds none.
Columns match fields as for select_many(..., into=...), with the same errors;
a NULL fits a list field only if its item type allows None. Raises TypeError
before running the query if into isn't a Struct with a list field, and when a
group key value is unhashable.
select_many(query, /, *args, params=None, into=None, strategy='auto')
async
¶
select_many(
query: str | Template,
/,
*args: object,
params: NamedParams | None = None,
into: type[dict[Any, Any]],
strategy: Strategy = "auto",
) -> list[dict[str, Any]]
Return every row.
select_one(query, /, *args, params=None, into=None, strategy='auto')
async
¶
select_one(
query: str | Template,
/,
*args: object,
params: NamedParams | None = None,
into: type[dict[Any, Any]],
strategy: Strategy = "auto",
) -> dict[str, Any]
Return exactly one row.
Raises NoRowsError for an empty result and TooManyRowsError for more than one row.
select_one_or_none(query, /, *args, params=None, into=None, strategy='auto')
async
¶
select_one_or_none(
query: str | Template,
/,
*args: object,
params: NamedParams | None = None,
into: type[dict[Any, Any]],
strategy: Strategy = "auto",
) -> dict[str, Any] | None
Return the only row, or None for an empty result (TooManyRowsError for >1 row).
TypeMismatchError
¶
Bases: DecodeError, TypeError
A column's PostgreSQL type is incompatible with the target field's annotation.
UniqueViolationError
¶
UnknownColumnError
¶
Bases: DecodeError
The result has a column the Struct does not declare (forbid_unknown_fields=True).
UnsupportedTypeError
¶
Bases: DecodeError
The PostgreSQL type (or field annotation) is not supported by the chosen strategy.
connect(dsn, *, statement_cache_size=256)
async
¶
Open a connection, e.g. await connect("postgresql://user:pw@host:5432/db").
TLS takes libpq's DSN keys: sslmode (disable, prefer (the default),
require, verify-ca, verify-full), sslrootcert (a CA file, or system;
verify-* without it uses the system's trusted roots), sslnegotiation=direct,
sslcert/sslkey, sslsni and channel_binding (prefer (the default),
require, disable). Unix sockets never use TLS. See the TLS guide in the docs.
connect_timeout (seconds, as in libpq: 0 or absent waits indefinitely, 1 means 2)
bounds each address tried, from its TCP connect to the end of the startup; when it
expires, InterfaceError says so.
TCP keepalives are on, as in libpq (keepalives=0 turns them off), and libpq's
keepalives_idle, keepalives_interval, keepalives_count (seconds, probes) and
tcp_user_timeout (milliseconds, Linux only) tune them: absent or 0 keeps the OS
default. Unix sockets ignore them. A dead peer then fails the connection with
ConnectionClosedError.
The connection is bound to the running event loop, whose thread runs its protocol and waits on the database socket directly: no extra threads, and a query costs no thread hand-offs.
create_pool(dsn, *, min_size=1, max_size=16, max_inflight_per_conn=8, statement_cache_size=256, max_lifetime=3600.0, max_idle=300.0)
async
¶
Open a pool with min_size connections (it grows on demand up to max_size).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
dsn
|
str
|
e.g. |
required |
min_size
|
int
|
Connections kept open even when idle. |
1
|
max_size
|
int
|
Upper bound on connections. Connections only open when every existing one is busy, so a higher ceiling costs nothing at low load, and leaves room for the occasional slow query not to hold up fast ones. |
16
|
max_inflight_per_conn
|
int
|
Shared statements pipelined on one connection at most;
|
8
|
statement_cache_size
|
int
|
Per-connection prepared statement cache (see |
256
|
max_lifetime
|
float | None
|
Seconds after which a connection is replaced once idle (None: never).
If closing it would leave fewer than |
3600.0
|
max_idle
|
float | None
|
Seconds after which an idle connection above |
300.0
|