Psycopg

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

Sync Configuration

class sqlspec.adapters.psycopg.PsycopgSyncConfig[source]

Bases: SyncDatabaseConfig[Connection, ConnectionPool, PsycopgSyncDriver]

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

driver_type

alias of PsycopgSyncDriver

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

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 | FlaskConfig | ADKConfig | EventsConfig | OpenTelemetryConfig | PrometheusConfig] | None) – Extension-specific configuration (e.g., Litestar plugin settings)

  • **kwargs (Any) – Additional keyword arguments

create_connection()[source]

Create a single connection (not from pool).

Return type:

Connection

Returns:

A psycopg Connection instance.

provide_connection(*args, **kwargs)[source]

Provide a connection context manager.

Parameters:

  • *args (Any) – Additional arguments.

  • **kwargs (Any) – Additional keyword arguments.

Return type:

PsycopgSyncConnectionContext

Returns:

A psycopg Connection context manager.

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

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)[source]

Provide pool instance.

Return type:

ConnectionPool

Returns:

The connection pool.

get_signature_namespace()[source]

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()[source]

Return polling defaults for PostgreSQL queue fallback.

Return type:

EventRuntimeHints

Async Configuration

class sqlspec.adapters.psycopg.PsycopgAsyncConfig[source]

Bases: AsyncDatabaseConfig[AsyncConnection, AsyncConnectionPool, PsycopgAsyncDriver]

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

driver_type

alias of PsycopgAsyncDriver

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

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 | FlaskConfig | ADKConfig | EventsConfig | OpenTelemetryConfig | PrometheusConfig] | None) – Extension-specific configuration (e.g., Litestar plugin settings)

  • **kwargs (Any) – Additional keyword arguments

async create_connection()[source]

Create a single async connection (not from pool).

Return type:

AsyncConnection

Returns:

A psycopg AsyncConnection instance.

provide_connection(*args, **kwargs)[source]

Provide an async connection context manager.

Parameters:

  • *args (Any) – Additional arguments.

  • **kwargs (Any) – Additional keyword arguments.

Return type:

PsycopgAsyncConnectionContext

Returns:

A psycopg AsyncConnection context manager.

get_signature_namespace()[source]

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)[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:

PsycopgAsyncSessionContext

Returns:

A PsycopgAsyncDriver session context manager.

async provide_pool(*args, **kwargs)[source]

Provide async pool instance.

Return type:

AsyncConnectionPool

Returns:

The async connection pool.

get_event_runtime_hints()[source]

Return polling defaults for PostgreSQL queue fallback.

Return type:

EventRuntimeHints

Shared Configuration

class sqlspec.adapters.psycopg.config.PsycopgConnectionParams[source]

Bases: TypedDict

Psycopg connection parameters.

class sqlspec.adapters.psycopg.config.PsycopgPoolParams[source]

Bases: PsycopgConnectionParams

Psycopg pool parameters.

Sync Driver

class sqlspec.adapters.psycopg.PsycopgSyncDriver[source]

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)[source]
dispatch_execute(cursor, statement)[source]

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)[source]

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)[source]

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)[source]

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()[source]

Begin a database transaction on the current connection.

Return type:

None

commit()[source]

Commit the current transaction on the current connection.

Return type:

None

rollback()[source]

Rollback the current transaction on the current connection.

Return type:

None

with_cursor(connection)[source]

Create context manager for PostgreSQL cursor.

Return type:

PsycopgSyncCursor

handle_database_exceptions()[source]

Handle database-specific exceptions and wrap them appropriately.

Return type:

PsycopgSyncExceptionHandler

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

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)[source]

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

Return type:

StorageBridgeJob

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

Load Arrow data into PostgreSQL using COPY.

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

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)[source]

Collect psycopg sync rows for the direct execution path.

Return type:

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

resolve_rowcount(cursor)[source]

Resolve rowcount from psycopg cursor for the direct execution path.

Return type:

int

Async Driver

class sqlspec.adapters.psycopg.PsycopgAsyncDriver[source]

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)[source]
async dispatch_execute(cursor, statement)[source]

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)[source]

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)[source]

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)[source]

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()[source]

Begin a database transaction on the current connection.

Return type:

None

async commit()[source]

Commit the current transaction on the current connection.

Return type:

None

async rollback()[source]

Rollback the current transaction on the current connection.

Return type:

None

with_cursor(connection)[source]

Create async context manager for PostgreSQL cursor.

Return type:

PsycopgAsyncCursor

handle_database_exceptions()[source]

Handle database-specific exceptions and wrap them appropriately.

Return type:

PsycopgAsyncExceptionHandler

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

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)[source]

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)[source]

Load Arrow data into PostgreSQL asynchronously via COPY.

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

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)[source]

Collect psycopg async rows for the direct execution path.

Return type:

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

resolve_rowcount(cursor)[source]

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[source]

Bases: SyncDataDictionaryBase

PostgreSQL-specific sync data dictionary.

dialect: ClassVar[str] = 'postgres'

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

__init__()[source]
get_version(driver)[source]

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)[source]

Check if PostgreSQL database supports a specific feature.

Parameters:
Return type:

bool

Returns:

True if feature is supported, False otherwise.

get_optimal_type(driver, type_category)[source]

Get optimal PostgreSQL type for a category.

Parameters:
Return type:

str

Returns:

PostgreSQL-specific type name.

get_tables(driver, schema=None)[source]

Get tables sorted by topological dependency order using Recursive CTE.

Return type:

list[TableMetadata]

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

Get column information for a table or schema.

Return type:

list[ColumnMetadata]

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

Get index metadata for a table or schema.

Return type:

list[IndexMetadata]

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

Get foreign key metadata.

Return type:

list[ForeignKeyMetadata]

class sqlspec.adapters.psycopg.data_dictionary.PsycopgAsyncDataDictionary[source]

Bases: AsyncDataDictionaryBase

PostgreSQL-specific async data dictionary.

dialect: ClassVar[str] = 'postgres'

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

__init__()[source]
async get_version(driver)[source]

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)[source]

Check if PostgreSQL database supports a specific feature.

Parameters:
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:
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]