AsyncPG#

High-performance async PostgreSQL adapter using asyncpg. Supports native pipelines, Arrow export, and Cloud SQL / AlloyDB connectors.

Configuration#

class sqlspec.adapters.asyncpg.AsyncpgConfig[source]#

Bases: AsyncDatabaseConfig[PoolConnectionProxy, Pool[Record], AsyncpgDriver]

Configuration for AsyncPG database connections using TypedDict.

driver_type#: alias of AsyncpgDriver

__init__(*, connection_config=None, connection_instance=None, migration_config=None, statement_config=None, driver_features=None, bind_key=None, extension_config=None, observability_config=None, **kwargs)[source]#

Initialize AsyncPG configuration.

Parameters:

connection_config¶ -- Connection and pool configuration parameters (TypedDict or dict)
connection_instance¶ -- Existing pool instance to use
migration_config¶ -- Migration configuration
statement_config¶ -- Statement configuration override
driver_features¶ -- Driver features configuration (TypedDict or dict)
bind_key¶ -- Optional unique identifier for this configuration
extension_config¶ -- Extension-specific configuration
observability_config¶ -- Adapter-level observability overrides for lifecycle hooks and observers
**kwargs¶ -- Additional keyword arguments

property pg_textsearch_available: bool#: Return True if the pg_textsearch extension is available.

async create_connection()[source]#

Open a standalone connection owned by the caller.

The connection carries the same connection parameters and init hook the pool applies, consumes no pool slot, and must be closed by the caller.

Return type:: PoolConnectionProxy
Returns:: An AsyncPG connection instance.

provide_session(*_args, statement_config=None, **_kwargs)[source]#

Provide an async driver session context manager.

Parameters:

*_args¶ (Any) -- Additional arguments.
statement_config¶ (StatementConfig | None) -- Optional statement configuration override.
**_kwargs¶ (Any) -- Additional keyword arguments.

Return type:

AsyncpgSessionContext

Returns:

An AsyncPG driver session context manager.

get_signature_namespace()[source]#

Get the signature namespace for AsyncPG types.

This provides all AsyncPG-specific types that Litestar needs to recognize to avoid serialization attempts.

Return type:: dict[str, typing.Any]
Returns:: Dictionary mapping type names to types.

get_event_runtime_hints()[source]#

Return polling defaults for PostgreSQL queue fallback.

Return type:: EventRuntimeHints

derive_direct_config()[source]#

Derive a session-mode configuration using direct_dsn for DDL, migrations, or LISTEN/NOTIFY.

Return type:: Self

Connection Parameters#

class sqlspec.adapters.asyncpg.AsyncpgConnectionConfig[source]#

Bases: TypedDict

TypedDict for AsyncPG connection parameters.

Pool Parameters#

class sqlspec.adapters.asyncpg.AsyncpgPoolConfig[source]#

Bases: AsyncpgConnectionConfig

TypedDict for AsyncPG pool parameters, inheriting connection parameters.

Driver Features#

class sqlspec.adapters.asyncpg.AsyncpgDriverFeatures[source]#

Bases: TypedDict

AsyncPG driver feature flags.

json_serializer: Custom JSON serializer function for PostgreSQL JSON/JSONB types.: Defaults to sqlspec.utils.serializers.to_json. Use for performance optimization or custom encoding behavior. Applied when enable_json_codecs is True.
json_deserializer: Custom JSON deserializer function for PostgreSQL JSON/JSONB types.: Defaults to sqlspec.utils.serializers.from_json. Use for performance optimization or custom decoding behavior. Applied when enable_json_codecs is True.
enable_json_codecs: Enable automatic JSON/JSONB codec registration on connections.: Defaults to True for seamless Python dict/list to PostgreSQL JSON/JSONB conversion. Set to False to disable automatic codec registration (manual handling required).
enable_pgvector: Enable pgvector extension support for vector similarity search.: Requires pgvector-python package (pip install pgvector) and PostgreSQL with pgvector extension. Defaults to True when pgvector-python is installed. Provides automatic conversion between Python objects and PostgreSQL vector types. Enables vector similarity operations and index support.
enable_paradedb: Enable ParadeDB (pg_search) extension detection.: When enabled and the pg_search extension is detected, the SQL dialect switches to "paradedb" which supports search operators (@@@, &&&, etc.) and inherits all pgvector distance operators. Defaults to True. Independent of enable_pgvector.
enable_pg_textsearch: Enable pg_textsearch extension detection for BM25 search.: When enabled and the pg_textsearch extension is detected, the SQL dialect switches to "pg_textsearch" which supports the BM25 <@> relevance ranking operator. Defaults to True.
enable_cloud_sql: Enable Google Cloud SQL connector integration.: Requires cloud-sql-python-connector package. Defaults to False (explicit opt-in required). Auto-configures IAM authentication, SSL, and IP routing. Mutually exclusive with enable_alloydb.
cloud_sql_instance: Cloud SQL instance connection name.: Format: "project:region:instance" Required when enable_cloud_sql is True.
cloud_sql_enable_iam_auth: Enable IAM database authentication.: Defaults to False for passwordless authentication. When False, requires user/password in connection_config.
cloud_sql_ip_type: IP address type for connection.: Options: "PUBLIC", "PRIVATE", "PSC" Defaults to "PRIVATE".
enable_alloydb: Enable Google AlloyDB connector integration.: Requires cloud-alloydb-python-connector package. Defaults to False (explicit opt-in required). Auto-configures IAM authentication and private networking. Mutually exclusive with enable_cloud_sql.
alloydb_instance_uri: AlloyDB instance URI.: Format: "projects/PROJECT/locations/REGION/clusters/CLUSTER/instances/INSTANCE" Required when enable_alloydb is True.
enable_alloydb_iam_auth: Enable IAM database authentication.: Defaults to False for passwordless authentication.
alloydb_ip_type: IP address type for connection.: Options: "PUBLIC", "PRIVATE", "PSC" Defaults to "PRIVATE".
enable_events: Enable database event channel support.: Defaults to True when extension_config["events"] is configured. Provides pub/sub capabilities via LISTEN/NOTIFY or table-backed fallback. Requires extension_config["events"] for migration setup when using poll_queue or notify_queue.
events_backend: Event channel backend selection.: Options: "notify", "notify_queue", "poll_queue" - "notify": Transient PostgreSQL LISTEN/NOTIFY with no replay or retry - "notify_queue": Durable queue plus a PostgreSQL notification wakeup hint - "poll_queue": Durable queue discovered by polling Defaults to "notify".
on_connection_create: Async callback executed when a connection is initialized.: Receives the raw asyncpg connection after internal codec registration.
pgbouncer: Enable PgBouncer transaction-pooling compatibility mode.: Disables server-side prepared statement caching (statement_cache_size=0).

type_codecs: Optional list of custom type codec specifications to register.

JSON and JSONB Codecs#

AsyncPG connections register binary json and jsonb codecs by default. This lets regular statement execution and load_from_arrow() pass Python dict and list values into PostgreSQL JSON / JSONB columns while preserving asyncpg's binary COPY protocol expectations for jsonb payloads. Set driver_features={"enable_json_codecs": False} when an application needs to manage asyncpg JSON codecs manually.

Driver#

class sqlspec.adapters.asyncpg.AsyncpgDriver[source]#

Bases: AsyncDriverAdapterBase

AsyncPG PostgreSQL driver for async database operations.

Supports COPY operations, numeric parameter style handling, PostgreSQL exception handling, transaction management, SQL statement compilation and caching, and parameter processing with type coercion.

__init__(connection, statement_config=None, driver_features=None)[source]#

Initialize driver adapter with connection and configuration.

Parameters:

connection¶ (PoolConnectionProxy) -- Database connection instance
statement_config¶ (StatementConfig | None) -- Statement configuration for the driver
driver_features¶ (dict[str, typing.Any] | None) -- Driver-specific features like extensions, secrets, and connection callbacks
observability¶ -- Optional runtime handling lifecycle hooks, observers, and spans

property data_dictionary: AsyncpgDataDictionary#

Get the data dictionary for this driver.

Returns:: Data dictionary instance for metadata queries

async dispatch_execute(cursor, statement)[source]#

Execute single SQL statement.

Handles both SELECT queries and non-SELECT operations.

Parameters:

cursor¶ (PoolConnectionProxy) -- AsyncPG connection object
statement¶ (SQL) -- SQL statement to execute

Return type:

ExecutionResult

Returns:

ExecutionResult with statement execution details

async dispatch_execute_many(cursor, statement)[source]#

Execute SQL with multiple parameter sets using AsyncPG's executemany.

Parameters:

cursor¶ (PoolConnectionProxy) -- AsyncPG connection object
statement¶ (SQL) -- SQL statement with multiple parameter sets

Return type:

ExecutionResult

Returns:

ExecutionResult with batch execution details

async dispatch_execute_script(cursor, statement)[source]#

Execute SQL script with statement splitting and parameter handling.

Parameters:

cursor¶ (PoolConnectionProxy) -- AsyncPG connection object
statement¶ (SQL) -- SQL statement containing multiple statements

Return type:

ExecutionResult

Returns:

ExecutionResult with script execution details

async dispatch_special_handling(cursor, statement)[source]#

Handle PostgreSQL COPY operations and other special cases.

Parameters:

cursor¶ (PoolConnectionProxy) -- AsyncPG connection object
statement¶ (SQL) -- SQL statement to analyze

Return type:

SQLResult | None

Returns:

SQLResult if special operation was handled, None for standard execution

async begin()[source]#

Begin a database transaction.

Uses asyncpg's own transaction handle so the driver and the connection agree on transaction state. A second call is a no-op rather than a savepoint, matching the other adapters and leaving nested transaction() blocks to the savepoint handling in the driver base.

asyncpg claims the connection for a transaction before it issues BEGIN and does not release the claim when BEGIN fails, which would turn the next attempt into a savepoint whose commit releases rather than commits. The claim is therefore released here.

Return type:: None

async commit()[source]#

Commit the current transaction.

Return type:: None

async rollback()[source]#

Rollback the current transaction.

Return type:: None

async set_migration_session_schema(schema)[source]#

Set the PostgreSQL search path for migration SQL.

Return type:: None

async set_migration_non_transactional_schema(schema)[source]#

Set the PostgreSQL search path for non-transactional migration SQL.

Return type:: None

async reset_migration_session_schema()[source]#

Reset the PostgreSQL search path after non-transactional migration SQL.

Return type:: None

async has_schema(schema)[source]#

Return whether a PostgreSQL schema exists.

Return type:: bool

with_cursor(connection)[source]#

Create context manager for AsyncPG cursor.

Return type:: AsyncpgCursor

dispatch_select_stream(statement, chunk_size)[source]#

Return a native asyncpg row stream backed by a cursor in a stream-owned transaction.

Return type:: Optional[AsyncRowStream[dict[str, typing.Any]]]

handle_database_exceptions()[source]#

Handle database exceptions with PostgreSQL error codes.

Return type:: AsyncpgExceptionHandler

async execute_stack(stack, *, continue_on_error=False)[source]#

Execute a StatementStack sequentially, reusing native prepared statements.

Return type:: tuple[StackResult, ...]

async select_to_storage(statement, destination, /, *parameters, statement_config=None, partitioner=None, format_hint=None, telemetry=None, **kwargs)[source]#

Execute a query and write the results to storage through the storage pipeline.

Results are encoded client-side and uploaded. PostgreSQL offers no native object-store destination for COPY, and asyncpg's COPY helpers use the STDIN/STDOUT stream form.

Return type:: StorageBridgeJob

async load_from_arrow(table, source, *, partitioner=None, overwrite=False, telemetry=None)[source]#

Load Arrow data into a PostgreSQL table via COPY.

Return type:: StorageBridgeJob

async load_from_records(table, records, *, columns=None, overwrite=False)[source]#

Load mapping or positional records directly with binary COPY.

Return type:: StorageBridgeJob

async load_from_storage(table, source, *, file_format, partitioner=None, overwrite=False)[source]#

Read an artifact from storage and ingest it via COPY.

Return type:: StorageBridgeJob

Extension Dialects#

AsyncPG supports the pgvector, pg_textsearch, and ParadeDB dialects for vector similarity search and full-text search operators. See the Dialects reference for operator details.

Data Dictionary#

class sqlspec.adapters.asyncpg.data_dictionary.AsyncpgDataDictionary[source]#

Bases: AsyncDataDictionaryBase

PostgreSQL-specific async data dictionary.

dialect: ClassVar[str] = 'postgres'#: Dialect identifier. Must be defined by subclasses as a class attribute.

async get_metadata_capabilities(driver, domains=None)[source]#

Get PostgreSQL data-dictionary capability profile.

Return type:: MetadataCapabilityProfile

async get_system_metadata_capabilities(driver, domains=None)[source]#

Get PostgreSQL opt-in system metadata capability disclosures.

Return type:: tuple[SystemMetadataCapability, ...]

async get_schemas(driver)[source]#

Get schema metadata.

Return type:: MetadataResult

async get_objects(driver, schema=None)[source]#

Get database object metadata.

Return type:: MetadataResult

async get_table_details(driver, table, schema=None)[source]#

Get rich table metadata.

Return type:: MetadataResult

async get_constraints(driver, table=None, schema=None)[source]#

Get constraint metadata.

Return type:: MetadataResult

async get_views(driver, schema=None)[source]#

Get view metadata.

Return type:: MetadataResult

async get_routines(driver, schema=None)[source]#

Get routine metadata.

Return type:: MetadataResult

async get_privileges(driver, object_name=None, schema=None)[source]#

Get privilege metadata.

Return type:: MetadataResult

async get_dependencies(driver, object_name=None, schema=None)[source]#

Get dependency metadata.

Return type:: MetadataResult

async get_ddl(driver, object_name, schema=None, *, object_type='table', include_dependencies=True, prefer_native=True, redact=True)[source]#

Get object DDL where PostgreSQL exposes native definition helpers.

Return type:: DDLResult

async get_system_metadata(driver, request=None, **kwargs)[source]#

Get opt-in PostgreSQL system metadata.

Return type:: SystemMetadataResult

async get_version(driver)[source]#

Get PostgreSQL database version information.

Performs an inline cache check to avoid a cross-module method call that can trigger a mypyc segmentation fault.

Parameters:: driver¶ (AsyncpgDriver) -- Async database driver instance.
Return type:: VersionInfo | None
Returns:: PostgreSQL version information or None if detection fails.

async get_feature_flag(driver, feature)[source]#

Check if PostgreSQL database supports a specific feature.

Parameters:

driver¶ (AsyncpgDriver) -- Async database driver instance.
feature¶ (str) -- Feature name to check.

Return type:

bool

Returns:

True if feature is supported, False otherwise.

async get_optimal_type(driver, type_category)[source]#

Get optimal PostgreSQL type for a category.

Parameters:

driver¶ (AsyncpgDriver) -- Async database driver instance.
type_category¶ (str) -- Type category.

Return type:

str

Returns:

PostgreSQL-specific type name.

async get_tables(driver, schema=None)[source]#

Get tables sorted by topological dependency order using Recursive CTE.

Return type:: list[TableMetadata]

async get_columns(driver, table=None, schema=None)[source]#

Get column information for a table or schema.

Return type:: list[ColumnMetadata]

async get_indexes(driver, table=None, schema=None)[source]#

Get index metadata for a table or schema.

Return type:: list[IndexMetadata]

async get_foreign_keys(driver, table=None, schema=None)[source]#

Get foreign key metadata.

Return type:: list[ForeignKeyMetadata]

Extension Settings#

Use the configuration types below in their corresponding extension_config namespace: "litestar", "events", or "adk" as supported by this adapter.

class sqlspec.adapters.asyncpg.litestar.AsyncpgLitestarConfig[source]#

Bases: LitestarConfig

Asyncpg-specific Litestar settings.

Use inside extension_config["litestar"] with this adapter's session store.

fillfactor: NotRequired[int]#

Type:: Table fillfactor. Default

autovacuum_vacuum_scale_factor: NotRequired[float]#: Table autovacuum vacuum scale factor.

autovacuum_analyze_scale_factor: NotRequired[float]#: Table autovacuum analyze scale factor.

class sqlspec.adapters.asyncpg.events.AsyncpgEventsConfig[source]#

Bases: EventsConfig

Asyncpg events settings for queue storage and supported native transports.

fillfactor: NotRequired[int]#: PostgreSQL queue-table fillfactor; omitted to use the server default.

autovacuum_vacuum_scale_factor: NotRequired[float]#: PostgreSQL queue vacuum threshold fraction; omitted to use the server default.

autovacuum_analyze_scale_factor: NotRequired[float]#: PostgreSQL queue analyze threshold fraction; omitted to use the server default.

listener_queue_capacity: NotRequired[int]#: Maximum buffered notifications per native listener; omitted for an unbounded queue.

class sqlspec.adapters.asyncpg.adk.AsyncpgADKConfig[source]#

Bases: ADKConfig

Asyncpg-specific ADK extension settings.

Use these keys inside extension_config["adk"] with the asyncpg ADK store.

enable_event_generated_columns: NotRequired[bool]#: Create PostgreSQL generated columns and indexes for common ADK event JSON paths.

enable_covering_indexes: NotRequired[bool]#: Add PostgreSQL INCLUDE columns to ADK event replay indexes.

fillfactor: NotRequired[int]#: Table fillfactor. Defaults to 80.

autovacuum_vacuum_scale_factor: NotRequired[float]#: Optional event-table autovacuum vacuum scale factor.

autovacuum_analyze_scale_factor: NotRequired[float]#: Optional event-table autovacuum analyze scale factor.

vector_index_type: NotRequired[Literal['hnsw', 'ivfflat', 'scann']]#

'hnsw'.

Type:: Vector index algorithm for memory embeddings ('hnsw', 'ivfflat', 'scann'). Default

vector_dimensions: NotRequired[int]#

Type:: Dimensionality of embedding vectors (e.g. 768 for gemini-embedding-001 with MRL). Default

enable_bm25: NotRequired[bool]#

False.

Type:: Enable native BM25 full-text indexing. Requires the pg_textsearch extension. Default

scann_num_leaves: NotRequired[int]#

Type:: Number of partition leaves (clusters) for ScaNN tree quantization. Default

scann_quantizer: NotRequired[str]#

'SQ8'.

Type:: Quantization method for ScaNN index ('SQ8', 'FP32'). Default

JSONB existence operator#

All PostgreSQL adapters support data ? 'key', data ? $1, data ? :key, ?|, and ?& without treating the operator as a placeholder. Write data ?? other_col for identifier or function right-hand operands. Write parameterized intervals as ? * interval '1 day' (or use $1).

Native execution options#

Set execution_args={"timeout": seconds} on StatementConfig to forward an asyncpg timeout to queries, batches, script statements, stack operations, and stream fetches. command_timeout is accepted as an alias; timeout takes precedence. Explicit 0 and None values are preserved.

driver_features={"pgbouncer": True} disables asyncpg's statement cache and SQLSpec's explicit prepared statements in stacks, while preserving transaction cleanup. The option is also accepted in connection_config. Use this mode when the proxy configuration does not support prepared statements; PgBouncer can support them when configured to track protocol-level prepared statements.

driver_features["type_codecs"] accepts a list of native codec specifications. Each entry requires typename, encoder, and decoder; schema defaults to public and format to text. Codecs register after SQLSpec's built-in JSON and vector setup and before on_connection_create.