CockroachDB + AsyncPG

CockroachDB adapter using asyncpg with automatic transaction retry logic.

Configuration

class sqlspec.adapters.cockroach_asyncpg.CockroachAsyncpgConfig

Bases: AsyncDatabaseConfig[PoolConnectionProxy, Pool, CockroachAsyncpgDriver]

Configuration for CockroachDB using AsyncPG.

driver_type

alias of CockroachAsyncpgDriver

connection_type

alias of PoolConnectionProxy

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

async create_connection()

Create a database connection.

Return type:

PoolConnectionProxy

provide_session(*_args, statement_config=None, follower_reads=None, staleness=None, **_kwargs)

Provide a database session context manager.

Return type:

CockroachAsyncpgSessionContext

async provide_pool(*args, **kwargs)

Provide pool instance.

Return type:

Pool

get_signature_namespace()

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

Return default event runtime hints for this configuration.

Return type:

EventRuntimeHints

class sqlspec.adapters.cockroach_asyncpg.CockroachAsyncpgPoolConfig

Bases: CockroachAsyncpgConnectionConfig

AsyncPG pool parameters for CockroachDB.

Driver

class sqlspec.adapters.cockroach_asyncpg.CockroachAsyncpgDriver

Bases: AsyncpgDriver

CockroachDB AsyncPG driver with retry support.

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

Initialize driver adapter with connection and configuration.

Parameters:

• connection (PoolConnectionProxy) -- 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 run_transaction_with_retry(operation)

Execute a full CockroachDB transaction callback with serialization retries.

Return type:

TypeVar(_T)

async dispatch_execute(cursor, statement)

Execute single SQL statement.

Handles both SELECT queries and non-SELECT operations.

Parameters:

• cursor (Any) -- AsyncPG connection object

• statement (SQL) -- SQL statement to execute

Return type:

ExecutionResult

Returns:

ExecutionResult with statement execution details

async dispatch_execute_many(cursor, statement)

Execute SQL with multiple parameter sets using AsyncPG's executemany.

Parameters:

• cursor (Any) -- AsyncPG connection object

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

Return type:

ExecutionResult

Returns:

ExecutionResult with batch execution details

async dispatch_execute_script(cursor, statement)

Execute SQL script with statement splitting and parameter handling.

Parameters:

• cursor (Any) -- AsyncPG connection object

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

Return type:

ExecutionResult

Returns:

ExecutionResult with script execution details

handle_database_exceptions()

Handle database exceptions with PostgreSQL error codes.

Return type:

CockroachAsyncpgExceptionHandler

property data_dictionary: CockroachAsyncpgDataDictionary

Get the data dictionary for this driver.

Returns:

Data dictionary instance for metadata queries

Retry Configuration

class sqlspec.adapters.cockroach_asyncpg.CockroachAsyncpgRetryConfig

Bases: object

CockroachDB asyncpg transaction retry configuration.

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

classmethod from_features(driver_features)

Build retry config from driver feature mappings.

Return type:

CockroachAsyncpgRetryConfig

Data Dictionary

class sqlspec.adapters.cockroach_asyncpg.data_dictionary.CockroachAsyncpgDataDictionary

Bases: AsyncDataDictionaryBase

CockroachDB async data dictionary (AsyncPG).

dialect: ClassVar[str] = 'cockroachdb'

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

async get_metadata_capabilities(driver, domains=None)

Get CockroachDB data-dictionary capability profile.

Return type:

MetadataCapabilityProfile

async get_system_metadata_capabilities(driver, domains=None)

Get CockroachDB 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 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)

Get lossy CockroachDB DDL status without parameterized identifiers.

Return type:

DDLResult

async get_system_metadata(driver, request=None, **kwargs)

Get opt-in CockroachDB system metadata.

Return type:

SystemMetadataResult

async get_version(driver)

Get database version information.

Parameters:

driver (CockroachAsyncpgDriver) -- Async database driver instance

Return type:

VersionInfo | None

Returns:

Version information or None if detection fails

async get_feature_flag(driver, feature)

Check if database supports a specific feature.

Parameters:

• driver (CockroachAsyncpgDriver) -- Async database driver instance

• feature (str) -- Feature name to check

Return type:

bool

Returns:

True if feature is supported, False otherwise

async get_optimal_type(driver, type_category)

Get optimal database type for a category.

Parameters:

• driver (CockroachAsyncpgDriver) -- Async database driver instance

• type_category (str) -- Type category

Return type:

str

Returns:

Database-specific type name

async get_tables(driver, schema=None)

Get list of tables in schema.

Parameters:

• driver (CockroachAsyncpgDriver) -- Async database driver instance

• schema (str | None) -- Schema name (None for default)

Return type:

list[TableMetadata]

Returns:

List of table metadata dictionaries

async get_columns(driver, table=None, schema=None)

Get column information for a table or schema.

Parameters:

• driver (CockroachAsyncpgDriver) -- Async database driver instance

• table (str | None) -- Table name (None to fetch columns for all tables in schema)

• schema (str | None) -- Schema name (None for default)

Return type:

list[ColumnMetadata]

Returns:

List of column metadata dictionaries

async get_indexes(driver, table=None, schema=None)

Get index information for a table or schema.

Parameters:

• driver (CockroachAsyncpgDriver) -- Async database driver instance

• table (str | None) -- Table name (None to fetch indexes for all tables in schema)

• schema (str | None) -- Schema name (None for default)

Return type:

list[IndexMetadata]

Returns:

List of index metadata dictionaries

async get_foreign_keys(driver, table=None, schema=None)

Get foreign key metadata.

Parameters:

• driver (CockroachAsyncpgDriver) -- Async database driver instance

• table (str | None) -- Optional table name filter

• schema (str | None) -- Optional schema name filter

Return type:

list[ForeignKeyMetadata]

Returns:

List of foreign key metadata