CockroachDB + Psycopg#

CockroachDB adapter using psycopg with automatic transaction retry logic. Provides both sync and async support.

Sync Configuration#

class sqlspec.adapters.cockroach_psycopg.CockroachPsycopgSyncConfig[source]#

Bases: SyncDatabaseConfig[CrdbConnection, ConnectionPool, CockroachPsycopgSyncDriver]

Configuration for CockroachDB synchronous connections using psycopg.

driver_type#

alias of CockroachPsycopgSyncDriver

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

Create a database connection.

Return type:

CrdbConnection

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

Provide a database session context manager.

Return type:

CockroachPsycopgSyncSessionContext

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

Provide pool instance.

Return type:

ConnectionPool

get_signature_namespace()[source]#

Get the signature namespace for this database configuration.

Returns a dictionary of type names to objects (classes, functions, or other callables) that should be registered with Litestar's signature namespace to prevent serialization attempts on database-specific structures.

Return type:

dict[str, typing.Any]

Returns:

Dictionary mapping type names to objects.

get_event_runtime_hints()[source]#

Return default event runtime hints for this configuration.

Return type:

EventRuntimeHints

Async Configuration#

class sqlspec.adapters.cockroach_psycopg.CockroachPsycopgAsyncConfig[source]#

Bases: AsyncDatabaseConfig[AsyncCrdbConnection, AsyncConnectionPool, CockroachPsycopgAsyncDriver]

Configuration for CockroachDB async connections using psycopg.

driver_type#

alias of CockroachPsycopgAsyncDriver

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

Create a database connection.

Return type:

AsyncCrdbConnection

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

Provide a database session context manager.

Return type:

CockroachPsycopgAsyncSessionContext

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

Provide pool instance.

Return type:

AsyncConnectionPool

get_signature_namespace()[source]#

Get the signature namespace for this database configuration.

Returns a dictionary of type names to objects (classes, functions, or other callables) that should be registered with Litestar's signature namespace to prevent serialization attempts on database-specific structures.

Return type:

dict[str, typing.Any]

Returns:

Dictionary mapping type names to objects.

get_event_runtime_hints()[source]#

Return default event runtime hints for this configuration.

Return type:

EventRuntimeHints

Sync Driver#

class sqlspec.adapters.cockroach_psycopg.CockroachPsycopgSyncDriver[source]#

Bases: PsycopgSyncDriver

CockroachDB sync driver using psycopg.crdb.

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

Execute a full CockroachDB transaction callback with serialization retries.

Return type:

TypeVar(_T)

dispatch_execute(cursor, statement)[source]#

Execute single SQL statement.

Parameters:

  • cursor (Cursor) -- 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 (Cursor) -- 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 (Cursor) -- Database cursor

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

Return type:

ExecutionResult

Returns:

ExecutionResult with script execution details

handle_database_exceptions()[source]#

Handle database-specific exceptions and wrap them appropriately.

Return type:

CockroachPsycopgSyncExceptionHandler

property data_dictionary: CockroachPsycopgSyncDataDictionary#

Get the data dictionary for this driver.

Returns:

Data dictionary instance for metadata queries

Async Driver#

class sqlspec.adapters.cockroach_psycopg.CockroachPsycopgAsyncDriver[source]#

Bases: PsycopgAsyncDriver

CockroachDB async driver using psycopg.crdb.

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

Execute a full CockroachDB transaction callback with serialization retries.

Return type:

TypeVar(_T)

async dispatch_execute(cursor, statement)[source]#

Execute single SQL statement (async).

Parameters:

  • cursor (AsyncCursor) -- 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 (AsyncCursor) -- 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 (AsyncCursor) -- Database cursor

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

Return type:

ExecutionResult

Returns:

ExecutionResult with script execution details

handle_database_exceptions()[source]#

Handle database-specific exceptions and wrap them appropriately.

Return type:

CockroachPsycopgAsyncExceptionHandler

property data_dictionary: CockroachPsycopgAsyncDataDictionary#

Get the data dictionary for this driver.

Returns:

Data dictionary instance for metadata queries

Retry Configuration#

class sqlspec.adapters.cockroach_psycopg.CockroachPsycopgRetryConfig[source]#

Bases: object

CockroachDB psycopg transaction retry configuration.

__init__(max_retries=10, base_delay_ms=50.0, max_delay_ms=5000.0, enable_logging=True)[source]#
classmethod from_features(driver_features)[source]#

Build retry config from driver feature mappings.

Return type:

CockroachPsycopgRetryConfig

Sync Data Dictionary#

class sqlspec.adapters.cockroach_psycopg.data_dictionary.CockroachPsycopgSyncDataDictionary[source]#

Bases: SyncDataDictionaryBase

CockroachDB sync data dictionary.

dialect: ClassVar[str] = 'cockroachdb'#

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

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

Get CockroachDB data-dictionary capability profile.

Return type:

MetadataCapabilityProfile

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

Get CockroachDB opt-in system metadata capability disclosures.

Return type:

tuple[SystemMetadataCapability, ...]

get_schemas(driver)[source]#

Get schema metadata.

Return type:

MetadataResult

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

Get database object metadata.

Return type:

MetadataResult

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

Get rich table metadata.

Return type:

MetadataResult

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

Get constraint metadata.

Return type:

MetadataResult

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

Get view metadata.

Return type:

MetadataResult

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

Get routine metadata.

Return type:

MetadataResult

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

Get privilege metadata.

Return type:

MetadataResult

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

Get stable dependency metadata.

Return type:

MetadataResult

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

Get lossy CockroachDB DDL status without parameterized identifiers.

Return type:

DDLResult

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

Get opt-in CockroachDB system metadata.

Return type:

SystemMetadataResult

get_version(driver)[source]#

Get CockroachDB version information.

Return type:

VersionInfo | None

get_feature_flag(driver, feature)[source]#

Check if CockroachDB supports a specific feature.

Return type:

bool

get_optimal_type(driver, type_category)[source]#

Get optimal CockroachDB type for a category.

Return type:

str

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

Get tables sorted by dependency order.

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]

Async Data Dictionary#

class sqlspec.adapters.cockroach_psycopg.data_dictionary.CockroachPsycopgAsyncDataDictionary[source]#

Bases: AsyncDataDictionaryBase

CockroachDB async data dictionary.

dialect: ClassVar[str] = 'cockroachdb'#

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

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

Get CockroachDB data-dictionary capability profile.

Return type:

MetadataCapabilityProfile

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

Get CockroachDB 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 stable 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 lossy CockroachDB DDL status without parameterized identifiers.

Return type:

DDLResult

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

Get opt-in CockroachDB system metadata.

Return type:

SystemMetadataResult

async get_version(driver)[source]#

Get CockroachDB version information.

Return type:

VersionInfo | None

async get_feature_flag(driver, feature)[source]#

Check if CockroachDB supports a specific feature.

Return type:

bool

async get_optimal_type(driver, type_category)[source]#

Get optimal CockroachDB type for a category.

Return type:

str

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

Get tables sorted by dependency order.

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]