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

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

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.

classmethod from_features(driver_features)[source]

Build retry config from driver feature mappings.

Return type:

CockroachPsycopgRetryConfig

__init__(max_retries=10, base_delay_ms=50.0, max_delay_ms=5000.0, enable_logging=True)

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_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_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]