CockroachDBEngine API

Connection and transaction management.

langchain_cockroachdb.engine.CockroachDBEngine

Manages an async SQLAlchemy engine for CockroachDB with retry support.

Example

engine = CockroachDBEngine.from_connection_string(
    "cockroachdb://root@localhost:26257/defaultdb?sslmode=disable",
    retry_policy=RetryPolicy(max_attempts=8),
)

from_connection_string(connection_string, *, pool_size=10, max_overflow=20, pool_pre_ping=True, pool_recycle=3600, pool_timeout=30.0, connect_timeout=10.0, retry_policy=None, retry_max_attempts=None, retry_initial_backoff=None, retry_max_backoff=None, retry_backoff_multiplier=None, retry_jitter=None, **kwargs) classmethod

Create an engine from a connection string.

Any common spelling works (cockroachdb://, postgresql://, postgres://, with or without +psycopg or +psycopg2); all of them use psycopg 3.

Parameters:

Name	Type	Description	Default
connection_string	str	CockroachDB connection URL	required
pool_size	int	Connection pool size (default: 10)	10
max_overflow	int	Connections allowed beyond pool_size (default: 20)	20
pool_pre_ping	bool	Check connections before use (default: True)	True
pool_recycle	int	Recycle connections after N seconds (default: 3600)	3600
pool_timeout	float	Seconds to wait for a pooled connection (default: 30.0)	30.0
connect_timeout	float | None	Seconds to wait when opening a connection, or None. Ignored when the URL or connect_args already set one (default: 10.0)	10.0
retry_policy	RetryPolicy | None	How operations retry (default: RetryPolicy())	None
retry_max_attempts	int | None	Shorthand for RetryPolicy.max_attempts	None
retry_initial_backoff	float | None	Shorthand for RetryPolicy.initial_backoff	None
retry_max_backoff	float | None	Shorthand for RetryPolicy.max_backoff	None
retry_backoff_multiplier	float | None	Shorthand for RetryPolicy.backoff_multiplier	None
retry_jitter	bool | None	Shorthand for RetryPolicy.jitter (True means 0.5)	None
**kwargs	Any	Passed to create_async_engine	{}

Returns:

Type	Description
CockroachDBEngine	CockroachDBEngine instance

from_engine(engine, *, loop=None, retry_policy=None, retry_max_attempts=None, retry_initial_backoff=None, retry_max_backoff=None, retry_backoff_multiplier=None, retry_jitter=None) classmethod

Wrap an existing AsyncEngine.

If you also use engine directly from your own event loop while calling sync methods here, pass that loop as loop so all work stays on it. Keep that loop running in its thread while the engine is in use. A call raises RuntimeError instead of waiting forever if the loop is closed or stops, or in a forked child process. In a child, build the engine with a loop started in the child, or without loop.

Parameters:

Name	Type	Description	Default
engine	AsyncEngine	SQLAlchemy AsyncEngine	required
loop	AbstractEventLoop | None	Event loop that owns engine (default: none)	None
retry_policy	RetryPolicy | None	How operations retry (default: RetryPolicy())	None
retry_max_attempts	int | None	Shorthand for RetryPolicy.max_attempts	None
retry_initial_backoff	float | None	Shorthand for RetryPolicy.initial_backoff	None
retry_max_backoff	float | None	Shorthand for RetryPolicy.max_backoff	None
retry_backoff_multiplier	float | None	Shorthand for RetryPolicy.backoff_multiplier	None
retry_jitter	bool | None	Shorthand for RetryPolicy.jitter (True means 0.5)	None

Returns:

Type	Description
CockroachDBEngine	CockroachDBEngine instance

Raises:

Type	Description
RuntimeError	If loop was running in the parent process when this process was forked. Its copy here can never run.

ainit_vectorstore_table(table_name, vector_dimension, *, schema='public', id_type='UUID', content_column='content', embedding_column='embedding', metadata_column='metadata', namespace_column=None, create_tsvector=False, fts_language='english', drop_if_exists=False) async

Create vector store table with optional full-text search and namespace.

Retries transient errors under the engine's retry policy.

Parameters:

Name	Type	Description	Default
table_name	str	Name of the table to create	required
vector_dimension	int	Dimension of vector embeddings	required
schema	str	Database schema (default: public)	'public'
id_type	str	Type for ID column (default: UUID)	'UUID'
content_column	str	Name of content column	'content'
embedding_column	str	Name of embedding column	'embedding'
metadata_column	str	Name of metadata column	'metadata'
namespace_column	str | None	Name of namespace column for multi-tenancy (default: None, disabled)	None
create_tsvector	bool	Create TSVECTOR column for FTS	False
fts_language	str	Text search config for the generated tsvector column (default: english). Must match HybridSearchConfig.fts_query_language when hybrid search is used.	'english'
drop_if_exists	bool	Drop table if it exists	False

init_vectorstore_table(table_name, vector_dimension, **kwargs)

Sync version of ainit_vectorstore_table.

aclose() async

Dispose the connection pool.

close()

Sync version of aclose.

Overview

The CockroachDBEngine manages database connections, connection pooling, and table initialization.

Class Methods

from_connection_string

Create engine from connection string.

engine = CockroachDBEngine.from_connection_string(
    "cockroachdb://user:pass@host:26257/db",
    pool_size=10,
    max_overflow=20,
)

from_engine

Create from existing SQLAlchemy engine.

from sqlalchemy.ext.asyncio import create_async_engine

async_engine = create_async_engine("cockroachdb://...")
engine = CockroachDBEngine.from_engine(async_engine)

Instance Methods

ainit_vectorstore_table

Create vector store table (async).

await engine.ainit_vectorstore_table(
    table_name="documents",
    vector_dimension=1536,
    drop_if_exists=False,
)

init_vectorstore_table

Create vector store table (sync).

engine.init_vectorstore_table(
    table_name="documents",
    vector_dimension=1536,
)

aclose / close

Close engine and connections.

await engine.aclose()  # Async
engine.close()          # Sync

Configuration Parameters

Parameter	Type	Default	Description
pool_size	int	10	Base connection pool size
max_overflow	int	20	Additional connections allowed
pool_pre_ping	bool	True	Health check before use
pool_recycle	int	3600	Recycle connections after (seconds)
pool_timeout	float	30.0	Wait timeout for connection
connect_timeout	int	10	Connection timeout in seconds
retry_policy	RetryPolicy	None	Retry policy for transient errors
retry_max_attempts	int	5	Maximum retry attempts (legacy, use retry_policy)
retry_initial_backoff	float	0.1	Initial retry delay (legacy, use retry_policy)
retry_max_backoff	float	10.0	Maximum retry delay (legacy, use retry_policy)
retry_backoff_multiplier	float	2.0	Backoff multiplier (legacy, use retry_policy)
retry_jitter	bool	True	Add randomization to backoff (legacy, use retry_policy)

Examples

See Configuration Guide for detailed examples.