PsqlPy#

Configuration#

class sqlspec.adapters.psqlpy.PsqlpyConfig[source]#

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

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¶ (PsqlpyPoolParams | dict[str, typing.Any] | None) -- Connection and pool configuration parameters.

• connection_instance¶ (Any | None) -- Existing connection pool instance to use.

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

• statement_config¶ (StatementConfig | None) -- SQL statement configuration.

• driver_features¶ (PsqlpyDriverFeatures | dict[str, typing.Any] | None) -- Driver feature configuration (TypedDict or dict).

• 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.

property pg_textsearch_available: bool#

Return True if the pg_textsearch extension is available.

async create_connection()[source]#

Create a single async connection (not from pool).

Return type:

Any

Returns:

A psqlpy 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:

PsqlpySessionContext

Returns:

A PsqlpyDriver session context manager.

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

Provide async pool instance.

Return type:

Any

Returns:

The async connection pool.

get_signature_namespace()[source]#

Get the signature namespace for Psqlpy types.

Return type:

dict[str, typing.Any]

Returns:

Dictionary mapping type names to types.

get_event_runtime_hints()[source]#

Return LISTEN/NOTIFY defaults for Psqlpy adapters.

Return type:

EventRuntimeHints

Connection Parameters#

class sqlspec.adapters.psqlpy.PsqlpyConnectionParams[source]#

Bases: TypedDict

Psqlpy connection parameters.

Pool Parameters#

class sqlspec.adapters.psqlpy.PsqlpyPoolParams[source]#

Bases: PsqlpyConnectionParams

Psqlpy pool parameters.

Driver Features#

class sqlspec.adapters.psqlpy.PsqlpyDriverFeatures[source]#

Bases: TypedDict

Psqlpy driver feature flags.

enable_cast_detection: Enable cast-aware parameter processing.

Defaults to True. When enabled, SQL casts in prepared statements guide psqlpy parameter coercion for JSON, UUID, and timestamp-like values.

enable_pgvector: Enable pgvector extension detection for vector similarity search.

Defaults to True when the pgvector Python package is installed. Detects the PostgreSQL vector extension and promotes the dialect to "pgvector". It does not register psqlpy type handlers; use psqlpy.extra_types.PgVector or explicit SQL casts for vector values.

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.

json_serializer: Custom JSON serializer applied to the statement configuration. json_deserializer: Custom JSON deserializer retained alongside the serializer for parity with asyncpg. on_connection_create: Async callback executed when a connection is acquired from pool.

Receives the raw psqlpy connection for low-level driver configuration. Called exactly once per physical connection using WeakSet tracking.

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".

Extension Dialects#

PsqlPy supports the pgvector, pg_textsearch, 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[source]#

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

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

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

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

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

Begin a database transaction.

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 psqlpy cursor.

Parameters:

connection¶ (Any) -- Psqlpy connection object

Return type:

PsqlpyCursor

Returns:

PsqlpyCursor context manager

dispatch_select_stream(statement, chunk_size)[source]#

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

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

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

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

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

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

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

Resolve rowcount from psqlpy result for the direct execution path.

Return type:

int

Data Dictionary#

class sqlspec.adapters.psqlpy.data_dictionary.PsqlpyDataDictionary[source]#

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

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

Check if PostgreSQL database supports a specific feature.

Return type:

bool

async get_optimal_type(driver, type_category)[source]#

Get optimal PostgreSQL type for a category.

Return type:

str

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.psqlpy.litestar.PsqlpyLitestarConfig[source]#

Bases: LitestarConfig

Psqlpy-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.psqlpy.events.PsqlpyEventsConfig[source]#

Bases: EventsConfig

Psqlpy 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.

class sqlspec.adapters.psqlpy.adk.PsqlpyADKConfig[source]#

Bases: ADKConfig

Psqlpy-specific ADK extension settings.

Use these keys inside extension_config["adk"] with the psqlpy 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.