Psycopg

PostgreSQL adapter using psycopg 3 with both sync and async support. Features native pipeline mode for multi-statement batching.

Cloud Connectivity

PsycopgSyncConfig can create sync pools through the Google AlloyDB in-process connector when both psycopg and google-cloud-alloydb-connector are installed:

from sqlspec.adapters.psycopg import PsycopgSyncConfig

config = PsycopgSyncConfig(
    connection_config={"user": "app", "password": "secret", "dbname": "orders"},
    driver_features={
        "enable_alloydb": True,
        "alloydb_instance_uri": "projects/project/locations/us-central1/clusters/cluster/instances/instance",
    },
)

This connector path is sync-only. PsycopgAsyncConfig continues to use regular psycopg async connection configuration.

Sync Configuration

class sqlspec.adapters.psycopg.PsycopgSyncConfig

Bases: SyncDatabaseConfig[Connection, ConnectionPool, PsycopgSyncDriver]

Configuration for Psycopg synchronous database connections with direct field-based configuration.

driver_type

alias of PsycopgSyncDriver

connection_type

alias of Connection

__init__(*, connection_config=None, connection_instance=None, migration_config=None, statement_config=None, driver_features=None, bind_key=None, extension_config=None, **kwargs)

Initialize Psycopg synchronous configuration.

Parameters:

• connection_config (PsycopgPoolParams | dict[str, typing.Any] | None) -- Connection and pool configuration parameters (TypedDict or dict)

• connection_instance (ConnectionPool | None) -- Existing pool instance to use

• migration_config (dict[str, typing.Any] | None) -- Migration configuration

• statement_config (StatementConfig | None) -- Default SQL statement configuration

• driver_features (PsycopgDriverFeatures | dict[str, typing.Any] | None) -- Optional driver feature configuration

• bind_key (str | None) -- Optional unique identifier for this configuration

• extension_config (dict[str, dict[str, Any] | LitestarConfig | FastAPIConfig | StarletteConfig | SanicConfig | FlaskConfig | ADKConfig | EventsConfig | OpenTelemetryConfig | PrometheusConfig] | None) -- Extension-specific configuration

• **kwargs (Any) -- Additional keyword arguments

create_connection()

Create a single connection (not from pool).

Return type:

Connection

Returns:

A psycopg Connection instance.

provide_session(*_args, statement_config=None, **_kwargs)

Provide a driver session context manager.

Parameters:

• *_args (Any) -- Additional arguments.

• statement_config (StatementConfig | None) -- Optional statement configuration override.

• **_kwargs (Any) -- Additional keyword arguments.

Return type:

PsycopgSyncSessionContext

Returns:

A PsycopgSyncDriver session context manager.

provide_pool(*args, **kwargs)

Provide pool instance.

Return type:

ConnectionPool

Returns:

The connection pool.

get_signature_namespace()

Get the signature namespace for Psycopg types.

This provides all Psycopg-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()

Return polling defaults for PostgreSQL queue fallback.

Return type:

EventRuntimeHints

Async Configuration

class sqlspec.adapters.psycopg.PsycopgAsyncConfig

Bases: AsyncDatabaseConfig[AsyncConnection, AsyncConnectionPool, PsycopgAsyncDriver]

Configuration for Psycopg asynchronous database connections with direct field-based configuration.

driver_type

alias of PsycopgAsyncDriver

connection_type

alias of AsyncConnection

__init__(*, connection_config=None, connection_instance=None, migration_config=None, statement_config=None, driver_features=None, bind_key=None, extension_config=None, **kwargs)

Initialize Psycopg asynchronous configuration.

Parameters:

• connection_config (PsycopgPoolParams | dict[str, typing.Any] | None) -- Connection and pool configuration parameters (TypedDict or dict)

• connection_instance (AsyncConnectionPool | None) -- Existing pool instance to use

• migration_config (dict[str, typing.Any] | None) -- Migration configuration

• statement_config (StatementConfig | None) -- Default SQL statement configuration

• driver_features (PsycopgDriverFeatures | dict[str, typing.Any] | None) -- Optional driver feature configuration

• bind_key (str | None) -- Optional unique identifier for this configuration

• extension_config (dict[str, dict[str, Any] | LitestarConfig | FastAPIConfig | StarletteConfig | SanicConfig | FlaskConfig | ADKConfig | EventsConfig | OpenTelemetryConfig | PrometheusConfig] | None) -- Extension-specific configuration

• **kwargs (Any) -- Additional keyword arguments

async create_connection()

Create a single async connection (not from pool).

Return type:

AsyncConnection

Returns:

A psycopg AsyncConnection instance.

get_signature_namespace()

Get the signature namespace for PsycopgAsyncConfig types.

Return type:

dict[str, typing.Any]

Returns:

Dictionary mapping type names to types.

provide_session(*_args, statement_config=None, **_kwargs)

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:

PsycopgAsyncSessionContext

Returns:

A PsycopgAsyncDriver session context manager.

async provide_pool(*args, **kwargs)

Provide async pool instance.

Return type:

AsyncConnectionPool

Returns:

The async connection pool.

get_event_runtime_hints()

Return polling defaults for PostgreSQL queue fallback.

Return type:

EventRuntimeHints

Shared Configuration

class sqlspec.adapters.psycopg.config.PsycopgConnectionParams

Bases: TypedDict

Psycopg connection parameters.

class sqlspec.adapters.psycopg.config.PsycopgPoolParams

Bases: PsycopgConnectionParams

Psycopg pool parameters.

Sync Driver

class sqlspec.adapters.psycopg.PsycopgSyncDriver

Bases: PsycopgPipelineMixin, SyncDriverAdapterBase

PostgreSQL psycopg synchronous driver.

Provides synchronous database operations for PostgreSQL using psycopg3. Supports SQL statement execution with parameter binding, transaction management, result processing with column metadata, parameter style conversion, PostgreSQL arrays and JSON handling, COPY operations for bulk data transfer, and PostgreSQL-specific error handling.

__init__(connection, statement_config=None, driver_features=None)

dispatch_execute(cursor, statement)

Execute single SQL statement.

Parameters:

• cursor (Any) -- Database cursor

• statement (SQL) -- SQL statement to execute

Return type:

ExecutionResult

Returns:

ExecutionResult with statement execution details

dispatch_execute_many(cursor, statement)

Execute SQL with multiple parameter sets.

Parameters:

• cursor (Any) -- Database cursor

• statement (SQL) -- SQL statement with parameter list

Return type:

ExecutionResult

Returns:

ExecutionResult with batch execution details

dispatch_execute_script(cursor, statement)

Execute SQL script with multiple statements.

Parameters:

• cursor (Any) -- Database cursor

• statement (SQL) -- SQL statement containing multiple commands

Return type:

ExecutionResult

Returns:

ExecutionResult with script execution details

dispatch_special_handling(cursor, statement)

Hook for PostgreSQL-specific special operations.

Parameters:

• cursor (Any) -- Psycopg cursor object

• statement (SQL) -- SQL statement to analyze

Return type:

SQLResult | None

Returns:

SQLResult if special handling was applied, None otherwise

begin()

Begin a database transaction on the current connection.

Return type:

None

commit()

Commit the current transaction on the current connection.

Return type:

None

rollback()

Rollback the current transaction on the current connection.

Return type:

None

set_migration_session_schema(schema)

Set the PostgreSQL search path for migration SQL.

Return type:

None

set_migration_non_transactional_schema(schema)

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

Return type:

None

reset_migration_session_schema()

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

Return type:

None

has_schema(schema)

Return whether a PostgreSQL schema exists.

Return type:

bool

with_cursor(connection)

Create context manager for PostgreSQL cursor.

Return type:

PsycopgSyncCursor

dispatch_select_stream(statement, chunk_size)

Return a native psycopg row stream backed by a server-side named cursor.

Return type:

Optional[SyncRowStream[dict[str, typing.Any]]]

handle_database_exceptions()

Handle database-specific exceptions and wrap them appropriately.

Return type:

PsycopgSyncExceptionHandler

execute_stack(stack, *, continue_on_error=False)

Execute a StatementStack using psycopg pipeline mode when supported.

Return type:

tuple[StackResult, ...]

select_to_storage(statement, destination, /, *parameters, statement_config=None, partitioner=None, format_hint=None, telemetry=None, **kwargs)

Execute a query and stream Arrow results to storage (sync).

Return type:

StorageBridgeJob

load_from_arrow(table, source, *, partitioner=None, overwrite=False, telemetry=None)

Load Arrow data into PostgreSQL using COPY.

Return type:

StorageBridgeJob

load_from_storage(table, source, *, file_format, partitioner=None, overwrite=False)

Load staged artifacts into PostgreSQL via COPY.

Return type:

StorageBridgeJob

property data_dictionary: PsycopgSyncDataDictionary

Get the data dictionary for this driver.

Returns:

Data dictionary instance for metadata queries

collect_rows(cursor, fetched)

Collect psycopg sync rows for the direct execution path.

Return type:

tuple[list[typing.Any], list[str], int]

resolve_rowcount(cursor)

Resolve rowcount from psycopg cursor for the direct execution path.

Return type:

int

Async Driver

class sqlspec.adapters.psycopg.PsycopgAsyncDriver

Bases: PsycopgPipelineMixin, AsyncDriverAdapterBase

PostgreSQL psycopg asynchronous driver.

Provides asynchronous database operations for PostgreSQL using psycopg3. Supports async SQL statement execution with parameter binding, async transaction management, async result processing with column metadata, parameter style conversion, PostgreSQL arrays and JSON handling, COPY operations for bulk data transfer, PostgreSQL-specific error handling, and async pub/sub support.

__init__(connection, statement_config=None, driver_features=None)

async dispatch_execute(cursor, statement)

Execute single SQL statement (async).

Parameters:

• cursor (Any) -- Database cursor

• statement (SQL) -- SQL statement to execute

Return type:

ExecutionResult

Returns:

ExecutionResult with statement execution details

async dispatch_execute_many(cursor, statement)

Execute SQL with multiple parameter sets (async).

Parameters:

• cursor (Any) -- Database cursor

• statement (SQL) -- SQL statement with parameter list

Return type:

ExecutionResult

Returns:

ExecutionResult with batch execution details

async dispatch_execute_script(cursor, statement)

Execute SQL script with multiple statements (async).

Parameters:

• cursor (Any) -- Database cursor

• statement (SQL) -- SQL statement containing multiple commands

Return type:

ExecutionResult

Returns:

ExecutionResult with script execution details

async dispatch_special_handling(cursor, statement)

Hook for PostgreSQL-specific special operations.

Parameters:

• cursor (Any) -- Psycopg async cursor object

• statement (SQL) -- SQL statement to analyze

Return type:

SQLResult | None

Returns:

SQLResult if special handling was applied, None otherwise

async begin()

Begin a database transaction on the current connection.

Return type:

None

async commit()

Commit the current transaction on the current connection.

Return type:

None

async rollback()

Rollback the current transaction on the current connection.

Return type:

None

async set_migration_session_schema(schema)

Set the PostgreSQL search path for migration SQL.

Return type:

None

async set_migration_non_transactional_schema(schema)

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

Return type:

None

async reset_migration_session_schema()

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

Return type:

None

async has_schema(schema)

Return whether a PostgreSQL schema exists.

Return type:

bool

with_cursor(connection)

Create async context manager for PostgreSQL cursor.

Return type:

PsycopgAsyncCursor

dispatch_select_stream(statement, chunk_size)

Return a native psycopg row stream backed by a server-side named cursor.

Return type:

Optional[AsyncRowStream[dict[str, typing.Any]]]

handle_database_exceptions()

Handle database-specific exceptions and wrap them appropriately.

Return type:

PsycopgAsyncExceptionHandler

async execute_stack(stack, *, continue_on_error=False)

Execute a StatementStack using psycopg async pipeline when supported.

Return type:

tuple[StackResult, ...]

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

Execute a query and stream Arrow data to storage asynchronously.

Return type:

StorageBridgeJob

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

Load Arrow data into PostgreSQL asynchronously via COPY.

Return type:

StorageBridgeJob

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

Load staged artifacts asynchronously.

Return type:

StorageBridgeJob

property data_dictionary: PsycopgAsyncDataDictionary

Get the data dictionary for this driver.

Returns:

Data dictionary instance for metadata queries

collect_rows(cursor, fetched)

Collect psycopg async rows for the direct execution path.

Return type:

tuple[list[typing.Any], list[str], int]

resolve_rowcount(cursor)

Resolve rowcount from psycopg cursor for the direct execution path.

Return type:

int

Extension Dialects

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

Data Dictionary

class sqlspec.adapters.psycopg.data_dictionary.PsycopgSyncDataDictionary

Bases: SyncDataDictionaryBase

PostgreSQL-specific sync data dictionary.

dialect: ClassVar[str] = 'postgres'

Dialect identifier. Must be defined by subclasses as a class attribute.

__init__()

get_metadata_capabilities(driver, domains=None)

Get PostgreSQL replacement data-dictionary capability profile.

Return type:

MetadataCapabilityProfile

get_system_metadata_capabilities(driver, domains=None)

Get PostgreSQL opt-in system metadata capability disclosures.

Return type:

tuple[SystemMetadataCapability, ...]

get_schemas(driver)

Get schema metadata.

Return type:

MetadataResult

get_objects(driver, schema=None)

Get database object metadata.

Return type:

MetadataResult

get_table_details(driver, table, schema=None)

Get rich table metadata.

Return type:

MetadataResult

get_constraints(driver, table=None, schema=None)

Get constraint metadata.

Return type:

MetadataResult

get_views(driver, schema=None)

Get view metadata.

Return type:

MetadataResult

get_routines(driver, schema=None)

Get routine metadata.

Return type:

MetadataResult

get_privileges(driver, object_name=None, schema=None)

Get privilege metadata.

Return type:

MetadataResult

get_dependencies(driver, object_name=None, schema=None)

Get dependency metadata.

Return type:

MetadataResult

get_ddl(driver, object_name, schema=None, *, object_type='table', include_dependencies=True, prefer_native=True, redact=True)

Get object DDL where PostgreSQL exposes native definition helpers.

Return type:

DDLResult

get_system_metadata(driver, request=None, **kwargs)

Get opt-in PostgreSQL system metadata.

Return type:

SystemMetadataResult

get_version(driver)

Get PostgreSQL database version information.

Parameters:

driver (PsycopgSyncDriver) -- Sync database driver instance.

Return type:

VersionInfo | None

Returns:

PostgreSQL version information or None if detection fails.

get_feature_flag(driver, feature)

Check if PostgreSQL database supports a specific feature.

Parameters:

• driver (PsycopgSyncDriver) -- Sync database driver instance.

• feature (str) -- Feature name to check.

Return type:

bool

Returns:

True if feature is supported, False otherwise.

get_optimal_type(driver, type_category)

Get optimal PostgreSQL type for a category.

Parameters:

• driver (PsycopgSyncDriver) -- Sync database driver instance.

• type_category (str) -- Type category.

Return type:

str

Returns:

PostgreSQL-specific type name.

get_tables(driver, schema=None)

Get tables sorted by topological dependency order using Recursive CTE.

Return type:

list[TableMetadata]

get_columns(driver, table=None, schema=None)

Get column information for a table or schema.

Return type:

list[ColumnMetadata]

get_indexes(driver, table=None, schema=None)

Get index metadata for a table or schema.

Return type:

list[IndexMetadata]

get_foreign_keys(driver, table=None, schema=None)

Get foreign key metadata.

Return type:

list[ForeignKeyMetadata]

class sqlspec.adapters.psycopg.data_dictionary.PsycopgAsyncDataDictionary

Bases: AsyncDataDictionaryBase

PostgreSQL-specific async data dictionary.

dialect: ClassVar[str] = 'postgres'

Dialect identifier. Must be defined by subclasses as a class attribute.

__init__()

async get_metadata_capabilities(driver, domains=None)

Get PostgreSQL replacement data-dictionary capability profile.

Return type:

MetadataCapabilityProfile

async get_system_metadata_capabilities(driver, domains=None)

Get PostgreSQL opt-in system metadata capability disclosures.

Return type:

tuple[SystemMetadataCapability, ...]

async get_schemas(driver)

Get schema metadata.

Return type:

MetadataResult

async get_objects(driver, schema=None)

Get database object metadata.

Return type:

MetadataResult

async get_table_details(driver, table, schema=None)

Get rich table metadata.

Return type:

MetadataResult

async get_constraints(driver, table=None, schema=None)

Get constraint metadata.

Return type:

MetadataResult

async get_views(driver, schema=None)

Get view metadata.

Return type:

MetadataResult

async get_routines(driver, schema=None)

Get routine metadata.

Return type:

MetadataResult

async get_privileges(driver, object_name=None, schema=None)

Get privilege metadata.

Return type:

MetadataResult

async get_dependencies(driver, object_name=None, schema=None)

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)

Get object DDL where PostgreSQL exposes native definition helpers.

Return type:

DDLResult

async get_system_metadata(driver, request=None, **kwargs)

Get opt-in PostgreSQL system metadata.

Return type:

SystemMetadataResult

async get_version(driver)

Get PostgreSQL database version information.

Parameters:

driver (PsycopgAsyncDriver) -- Async database driver instance.

Return type:

VersionInfo | None

Returns:

PostgreSQL version information or None if detection fails.

async get_feature_flag(driver, feature)

Check if PostgreSQL database supports a specific feature.

Parameters:

• driver (PsycopgAsyncDriver) -- 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)

Get optimal PostgreSQL type for a category.

Parameters:

• driver (PsycopgAsyncDriver) -- Async database driver instance.

• type_category (str) -- Type category.

Return type:

str

Returns:

PostgreSQL-specific type name.

async get_tables(driver, schema=None)

Get tables sorted by topological dependency order using Recursive CTE.

Return type:

list[TableMetadata]

async get_columns(driver, table=None, schema=None)

Get column information for a table or schema.

Return type:

list[ColumnMetadata]

async get_indexes(driver, table=None, schema=None)

Get index metadata for a table or schema.

Return type:

list[IndexMetadata]

async get_foreign_keys(driver, table=None, schema=None)

Get foreign key metadata.

Return type:

list[ForeignKeyMetadata]