PsqlPy

Async PostgreSQL adapter using psqlpy, a Rust-backed PostgreSQL driver with native connection pooling.

Configuration

class sqlspec.adapters.psqlpy.PsqlpyConfig

Bases: AsyncDatabaseConfig[Any, ConnectionPool, PsqlpyDriver]

Configuration for Psqlpy asynchronous database connections.

driver_type

alias of PsqlpyDriver

connection_type

alias of Any

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

Initialize Psqlpy configuration.

Extracts the 'on_connection_create' hook from driver_features before storing them. Initializes a set to track initialized connection IDs because psqlpy connections do not support weak references.

Parameters:

• connection_config -- Connection and pool configuration parameters.

• connection_instance -- Existing connection pool instance to use.

• migration_config -- Migration configuration.

• statement_config -- SQL statement configuration.

• driver_features -- Driver feature configuration (TypedDict or dict).

• bind_key -- Optional unique identifier for this configuration.

• extension_config -- Extension-specific configuration.

• **kwargs -- Additional keyword arguments.

async create_connection()

Create a single async connection (not from pool).

Return type:

Any

Returns:

A psqlpy Connection instance.

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:

PsqlpySessionContext

Returns:

A PsqlpyDriver session context manager.

async provide_pool(*args, **kwargs)

Provide async pool instance.

Return type:

ConnectionPool

Returns:

The async connection pool.

get_signature_namespace()

Get the signature namespace for Psqlpy types.

Return type:

dict[str, typing.Any]

Returns:

Dictionary mapping type names to types.

get_event_runtime_hints()

Return LISTEN/NOTIFY defaults for Psqlpy adapters.

Return type:

EventRuntimeHints

class sqlspec.adapters.psqlpy.config.PsqlpyConnectionParams

Bases: TypedDict

Psqlpy connection parameters.

Extension Dialects

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

driver_features={"enable_pgvector": True} enables extension detection and promotes the runtime dialect to pgvector when the PostgreSQL vector extension is installed. It does not register automatic psqlpy vector type handlers; pass psqlpy.extra_types.PgVector values or use explicit SQL casts for vector parameters.

Driver

class sqlspec.adapters.psqlpy.PsqlpyDriver

Bases: AsyncDriverAdapterBase

PostgreSQL driver implementation using psqlpy.

Provides parameter style conversion, type coercion, error handling, and transaction management.

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

Initialize driver adapter with connection and configuration.

Parameters:

• connection (Any) -- 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

async dispatch_execute(cursor, statement)

Execute single SQL statement.

Parameters:

• cursor (Any) -- Psqlpy connection object

• statement (SQL) -- SQL statement to execute

Return type:

ExecutionResult

Returns:

ExecutionResult with execution metadata

async dispatch_execute_many(cursor, statement)

Execute SQL with multiple parameter sets.

Parameters:

• cursor (Any) -- Psqlpy connection object

• statement (SQL) -- SQL statement with multiple parameter sets

Return type:

ExecutionResult

Returns:

ExecutionResult with batch execution metadata

async dispatch_execute_script(cursor, statement)

Execute SQL script with statement splitting.

Parameters:

• cursor (Any) -- Psqlpy connection object

• statement (SQL) -- SQL statement with script content

Return type:

ExecutionResult

Returns:

ExecutionResult with script execution metadata

async begin()

Begin a database transaction.

Return type:

None

async commit()

Commit the current transaction.

Return type:

None

async rollback()

Rollback the current transaction.

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 context manager for psqlpy cursor.

Parameters:

connection (Any) -- Psqlpy connection object

Return type:

PsqlpyCursor

Returns:

PsqlpyCursor context manager

dispatch_select_stream(statement, chunk_size)

Return a native psqlpy row stream backed by a server-side cursor in a transaction.

Return type:

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

handle_database_exceptions()

Handle database-specific exceptions.

Return type:

PsqlpyExceptionHandler

Returns:

Exception handler context manager

async 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 a storage backend.

Return type:

StorageBridgeJob

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

Load Arrow-formatted data into PostgreSQL via psqlpy binary COPY.

Return type:

StorageBridgeJob

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

Load staged artifacts from storage using the storage bridge pipeline.

Return type:

StorageBridgeJob

prepare_driver_parameters(parameters, statement_config, is_many=False, prepared_statement=None)

Prepare parameters with cast-aware type coercion for psqlpy.

Parameters:

• parameters (Any) -- Parameters in any format

• statement_config (StatementConfig) -- Statement configuration

• is_many (bool) -- Whether this is for execute_many operation

• prepared_statement (Any | None) -- Prepared statement containing the original SQL statement

Return type:

Any

Returns:

Parameters with cast-aware type coercion applied

property data_dictionary: PsqlpyDataDictionary

Get the data dictionary for this driver.

Returns:

Data dictionary instance for metadata queries

collect_rows(cursor, fetched)

Collect psqlpy rows for the direct execution path.

The fetched argument may be a psqlpy query result or a plain list.

Return type:

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

resolve_rowcount(cursor)

Resolve rowcount from psqlpy result for the direct execution path.

Return type:

int

Data Dictionary

class sqlspec.adapters.psqlpy.data_dictionary.PsqlpyDataDictionary

Bases: AsyncDataDictionaryBase

PostgreSQL-specific async data dictionary via psqlpy.

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.

Performs an inline cache check first to avoid a cross-module method call that causes mypyc segfaults. If not cached, fetches from the database.

Return type:

VersionInfo | None

async get_feature_flag(driver, feature)

Check if PostgreSQL database supports a specific feature.

Return type:

bool

async get_optimal_type(driver, type_category)

Get optimal PostgreSQL type for a category.

Return type:

str

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]