Resources¶

Physical connections, and the databases built on top of them. A database is one of two kinds, discriminated by a required kind field: a plain generic database, or a CDM database with its logical vocab/results role bundle.

ConnectionConfig¶

A concrete database endpoint: dialect, host, credentials, target database. Stored in [connections.<name>].

Bases: SecretSafeBaseModel

Complete specification of one physical database connection: server address, credentials, and target database.

Referenced by :attr:DatabaseConfig.connection and :attr:DatabaseConfig.vocab_connection. Each entry under [connections] in config.toml maps to one instance of this model.

Passwords are stored in plaintext for now; secret management support is planned for a future release.

dialect_name `property` ¶

dialect_name: str

SQLAlchemy dialect name (e.g. "postgresql", "sqlite"), derived from dialect alone. Unlike :attr:ResolvedConnection.dialect_name, this needs no other field set (host, database_name, etc.), so it is safe to call before the connection is otherwise complete.

build_url ¶

build_url() -> str

Build the full connection URL, including the plaintext password.

Returns:

Type	Description
`str`	SQLAlchemy-compatible connection URL. For SQLite, returns `sqlite:///<database_name>`; `database_name` must be set explicitly (e.g. to `:memory:`), there is no implicit default.

physical_identity ¶

physical_identity() -> tuple[Any, ...]

Identity tuple (physical connection key, user) used to decide if this connection is the same physical connection as another one.

user is a separate element so callers comparing addresses alone can drop it.

physical_key ¶

physical_key() -> str

Physical identity of the database this connection addresses.

See :func:connection_key, which this defers to so a config entry and a live URL compare through one implementation.

resolve ¶

resolve(name: str) -> ResolvedConnection

Resolve this connection to a concrete, engine-ready target.

safe_url ¶

safe_url() -> str

Build the connection URL with the password redacted.

Safe for logging and display. Identical to build_url() for SQLite connections, which carry no password.

Returns:

Type	Description
`str`	Connection URL with `***` substituted for the password field.

to_env_pairs ¶

to_env_pairs(prefix: str) -> list[str]

Return PREFIX_FIELD=value strings for each non-None field.

Used by :func:~oa_configurator.io.write_env_file to emit env vars for Docker Compose env_file:. Field names are uppercased directly (e.g. host → PREFIX_HOST), so adding a new field here automatically appears in the export without touching io.py. The test_only config-only flag is excluded, since it is not a database connection parameter.

DatabaseKind¶

Bases: str, Enum

Discriminator for :class:DatabaseConfig subclasses.

DatabaseConfig¶

The shared base unifying kind and connection. Every [databases.<name>] entry is one of the two concrete kinds below, chosen by its own kind field. Each kind adds its own schema field on top

schema_name on GenericDatabaseConfig,
cdm_schema on CDMDatabaseConfig.

Both default to None, meaning "no schema override, use the connection's own default" (Postgres's own search_path default, typically public), not a guarantee this library makes or encodes in code.

Bases: SecretSafeBaseModel

Shared interface for every named database: a connection, plus each concrete kind's own schema field(s).

Abstract in practice: kind has no default, so every concrete entry must declare it explicitly via one of the subclasses below. Use this class (not a subclass) for isinstance checks and RefTo targets that accept any kind; use :data:DatabaseEntry for parsing raw config data, which dispatches to the correct subclass based on kind.

Notes

schema dispatch is handled in subclasses. See :class:GenericDatabaseConfig for schema_name and :class:CDMDatabaseConfig for cdm_schema, vocab_schema, and results_schema.

connection_name_for_role ¶

connection_name_for_role(role: Role) -> str

Name of the connection entry to use for role.

Always self.connection here; overridden by CDMDatabaseConfig, whose vocab role may route to a separate connection.

GenericDatabaseConfig¶

kind = "generic". A database with no CDM-specific fields: just connection and schema_name. Used by anything that isn't the CDM itself, e.g. a vector store's own database (see Vector Stores).

Bases: DatabaseConfig

A connection plus one optional schema. No CDM role-splitting.

Used by consumers that need a single database with no vocab/results distinction, e.g. an embedding store or a metadata database.

resolve ¶

resolve(name: str, stack: StackConfig) -> ResolvedDatabase

Resolve this database to a concrete connection and effective schema.

stack must already have passed :meth:StackConfig.validate_references, so self.connection is guaranteed to exist in stack.connections.

CDMDatabaseConfig¶

kind = "cdm". Maps the OMOP logical roles (CDM, vocab, results) to named connections and schema names.

Bases: DatabaseConfig

Maps the OMOP logical roles (CDM, vocab, results) to named connections and schema names.

This is what a CDM-consuming package treats as "its database": the logical CDM/vocab/results bundle, as opposed to :class:ConnectionConfig (the physical server address and credentials underneath it). Most packages only need a single cdm_db database.

connection_name_for_role ¶

connection_name_for_role(role: Role) -> str

Name of the connection entry to use for role: vocab_connection for Role.VOCAB when explicitly configured, else the primary connection.

resolve ¶

resolve(
    name: str, stack: StackConfig
) -> ResolvedCDMDatabase

Resolve this database to concrete connections and effective schema names.

The vocab connection falls back to the primary connection when not explicitly configured; the vocab schema falls back to the CDM schema under the same condition. vocab_schema/results_schema resolve to None when their own connection's dialect has no real multi-schema concept (e.g. SQLite), rather than carrying a dialect-inappropriate string that a later fold would need to correct. stack must already have passed :meth:StackConfig.validate_references, so self.connection/ self.vocab_connection are guaranteed to exist in stack.connections.

Role¶

Selects among a CDM database's several connections at resolve time. Not related to kind: kind decides which fields an entry has at config-authoring time, Role selects among one CDM entry's connections. A generic entry only ever has one connection, so Role has nothing to select there.

Bases: StrEnum

Pre-determined logical schema tags for consumers. Each tag corresponds to a configurable schema name (depending on the database's type [Database vs. CDMDatabase]): - Role.PRIMARY -> schema or cdm_schema: primary tables - Role.VOCAB -> vocab_schema: vocabulary tables - Role.RESULTS -> results_schema: results tables

Also disambiguates which physical connection to use on a CDMDatabase: Role.PRIMARY and Role.VOCAB each select a connection (connection_for_role, create_engine(role=...)); Role.RESULTS has no connection of its own and always resolves through Role.PRIMARY's.

Dialect¶

The SQLAlchemy backend names (Engine.dialect.name / get_backend_name()) this codebase recognizes: currently postgresql and sqlite. Distinct from ConnectionConfig.dialect, which stays a free-form string to carry a driver suffix such as postgresql+psycopg; Dialect is the plain backend-name axis every dialect-keyed dispatch in this codebase branches on. See Dialect support across the stack for the convention every consuming repo follows to add a new one.

Bases: StrEnum

SQLAlchemy backend names (Engine.dialect.name / get_backend_name()) this codebase recognizes, independent of the driver.

Distinct from ConnectionConfig.dialect, which stays a free-form string to carry a driver suffix such as "postgresql+psycopg".

Resolved types¶

ConnectionConfig.resolve(), GenericDatabaseConfig.resolve(), and CDMDatabaseConfig.resolve() produce these. Resolver.resolve_connection()/resolve_database() are thin wrappers around the same methods; resolve_database() returns the resolved subtype matching the entry's own kind.

Concrete physical connection ready for engine creation.

Attributes:

Name	Type	Description
`name`	`str`	Logical name of the connection as declared in the config.
`url`	`str`	Full database URL including credentials. TODO: make this a private attribute to avoid accidental password exposure; requires a factory method or `__post_init__` since dataclass field visibility can't be changed without breaking callers.
`safe_url`	`str`	Database URL with credentials redacted, safe for logging and display.
`_engine_url`	`URL`	SQLAlchemy URL object used for engine creation. Avoids the lossy string round-trip through `url` for SQLite paths containing `?`/`#`.
`test_only`	`bool`	Whether it is a test-only connection.

dialect_name `property` ¶

dialect_name: str

SQLAlchemy dialect name (e.g. "postgresql", "sqlite"), read off the URL directly.

addresses_same_database_as ¶

addresses_same_database_as(
    other: ResolvedConnection,
) -> bool

Do self and other address the same physical database?

Compares physical identity rather than dataclass equality, so a second config entry pointing at one database is recognised as the same database rather than as a separate one.

create_engine ¶

create_engine(**kwargs: Any) -> Engine

Create a SQLAlchemy engine for this connection.

Parameters:

Name	Type	Description	Default
`**kwargs`	`Any`	Forwarded to `sqlalchemy.create_engine`. `pool_pre_ping` defaults to `True`. Pass `pool_pre_ping=False` to opt out.	`{}`

Notes

pool_pre_ping=True checks a pooled connection is still alive before handing it out, avoiding stale-connection failures after a long-idle period. It adds a small overhead to every checkout, so it can be disabled when the database is known to be reliable and the application is latency-sensitive using **kwargs.

Returns:

Type	Description
`Engine`

physical_key ¶

physical_key() -> str

Physical identity of the database this connection addresses.

See :func:connection_key. Two connection entries with different names that address one database share a key, which is what decides whether a role is genuinely hosted elsewhere.

Resolved generic database: one connection, one optional schema.

Attributes:

Name	Type	Description
`name`	`str`	Logical name of the database as declared in the config.
`connection`	`ResolvedConnection`	Resolved connection for this database.
`schema_name`	`str \| None`	Effective schema name for this database, or None for no override (use the connection's own default/search_path).

configured_internal_schema_translate_map ¶

configured_internal_schema_translate_map() -> dict[
    str, str | None
]

Schema translate map for this database. Routing: - "primary" → schema_name (or None if the dialect has no real multi-schema concept, e.g. SQLite)

Notes

This is NOT the full schema_translate_map for an engine! See :meth:create_engine for the full map, which merges the resolver's own config-derived claims with any additional claims passed in by the caller.

connection_for_role ¶

connection_for_role(
    role: Role = Role.PRIMARY,
) -> ResolvedConnection

Return the resolved connection for a given role. Only Role.PRIMARY is valid here as ResolvedDatabase has no vocab/results connection-splitting.

Raises:

Type	Description
`ValueError`	If role is not `Role.PRIMARY`.

connection_for_schema_tag ¶

connection_for_schema_tag(
    schema_tag: Role | str,
) -> ResolvedConnection

Return the connection hosting schema_tag's physical schema.

Mirrors :meth:route_for_schema_tag's routing rule, so a caller asking where a tag lives gets the same answer as a caller routing a statement to it.

create_engine ¶

create_engine(
    *,
    schema_claims: Iterable[SchemaClaim] = (),
    extensions: Sequence[Callable[[Any, Any], None]] = (),
    execution_options: dict[str, Any] | None = None,
    owner: str | None = None,
    register_claims: bool = True,
    **kwargs: Any,
) -> Engine

Create a SQLAlchemy engine with the schema translate map applied.

Parameters are as for :meth:_build_engine, which this calls for the primary connection.

foreign_key_can_span ¶

foreign_key_can_span(
    from_tag: Role | str, to_tag: Role | str
) -> bool

Can a foreign key reference to_tag's tables from from_tag's?

False when the two tags live on different physical databases, where no dialect can express the constraint. Ask this instead of deriving a split flag, so the answer stays correct if a deployment later colocates the two or federates them behind one connection.

occupied_schemas ¶

occupied_schemas(connection: Connection) -> set[str]

Physical schema names this database currently claims that live on connection's own physical server.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	Open connection, used both to match this database's own roles against it and to read its live default schema (search_path-dependent, not a static per-dialect guess). The caller is expected to already have one in scope, not open a fresh one just for this.	required

resolved_physical_schemas ¶

resolved_physical_schemas(
    *, probe_options: _SchemaProbeOptions | None = None
) -> dict[str, str | None]

Physical schema every internal tag resolves to, with unset ones filled in from their own connection's live default schema.

A tag left unset in config means "use the connection's own default schema", which only that connection can answer. Reading it here, per tag, keeps one answer for the whole database, so engines built for different roles agree on where every tag lives.

Stays None for a tag whose connection has no real schema concept, where None is the correct and final answer.

Notes

Opens one connection per distinct connection that has an unset tag, reusing the supplied engine for its own connection and forwarding the same engine options to any remote connection. No connection is opened when config names every schema.

roles_on_connection ¶

roles_on_connection(
    connection: Connection | Engine,
) -> tuple[Role, ...]

Role(s) this database entry has on connection's own physical server (host, database, port), or () if none.

connection may be an already-open Connection or an Engine directly (Engine.engine returns itself, so connection.engine.url works either way).

Only Role.PRIMARY is possible here, since ResolvedDatabase has no vocab/results connection-splitting.

route_for_schema_tag ¶

route_for_schema_tag(
    schema_tag: Role | str, *, vocab: _T, primary: _T
) -> _T

Determines the right connection, engine, or session to use for a given schema_tag in a split-connection database.

Notes

Only Role.PRIMARY and any string are valid here, since ResolvedDatabase has no vocab/results connection-splitting.

Raises:

Type	Description
`TypeError`	If vocab and primary are not the same type, and aren't both members of Bindable.
`ValueError`	If schema_tag is a Role other than Role.PRIMARY, which has no meaning for this database.

schema_for_role ¶

schema_for_role(role: Role = Role.PRIMARY) -> str | None

Return the effective schema for a given role. Only PRIMARY is valid here as ResolvedDatabase has no vocab/results role-splitting.

Raises:

Type	Description
`ValueError`	If role is not `Role.PRIMARY`.

schema_tags ¶

schema_tags() -> tuple[Role, ...]

Role tags whose schema provenance is worth tracking for this database.

tags_share_a_transaction ¶

tags_share_a_transaction(*schema_tags: Role | str) -> bool

Can one transaction cover statements against all of schema_tags?

True only when every tag resolves to the same physical database. Nothing provides atomicity across two engines, so a caller needing all-or-nothing behaviour must check this first. Vacuously true for no tags or one tag.

Bases: ResolvedDatabase

Resolved CDM database: adds vocab/results role-splitting on top of :class:ResolvedDatabase.

Attributes:

Name	Type	Description
`vocab_connection`	`ResolvedConnection`	Resolved vocabulary connection for this database. May be the same as connection if no separate vocab connection is configured.
`vocab_schema`	`str or None`	Effective vocabulary schema name for this database. May be the same as schema_name if no separate vocab schema is configured; None when vocab_connection's dialect has no real multi-schema concept (e.g. SQLite).
`results_schema`	`str or None`	Effective results schema name for this database. May be the same as schema_name if no separate results schema is configured; None when connection's dialect has no real multi-schema concept (e.g. SQLite).

cdm_schema `property` ¶

cdm_schema: str | None

Alias for schema_name, matching OHDSI's own CDM/VOCAB/RESULTS naming.

configured_internal_schema_translate_map ¶

configured_internal_schema_translate_map() -> dict[
    str, str | None
]

SQLAlchemy schema translate map for OMOP ORM models.

Routing: "primary" → cdm_schema "vocab" → vocab_schema, falling back to cdm_schema if unset "results" → results_schema, falling back to cdm_schema if unset

Notes

This is NOT the full schema_translate_map for an engine! See :meth:create_engine for the full map, which merges the resolver's own config-derived claims with any additional claims passed in by the caller.

connection_for_role ¶

connection_for_role(
    role: Role = Role.PRIMARY,
) -> ResolvedConnection

Return the resolved connection for a given role: only Role.PRIMARY or Role.VOCAB, the only two physical connections that exist.

Raises:

Type	Description
`ValueError`	If role is neither `Role.PRIMARY` nor `Role.VOCAB`.

connection_for_schema_tag ¶

connection_for_schema_tag(
    schema_tag: Role | str,
) -> ResolvedConnection

Return the connection hosting schema_tag's physical schema.

Role.VOCAB resolves to vocab_connection; every other tag resolves to the primary connection, matching :meth:route_for_schema_tag.

create_engine ¶

create_engine(**kwargs: Any) -> Engine

Always raises: a CDM database has a primary and a vocabulary engine.

Raises:

Type	Description
`TypeError`	Always. Use :meth:`create_engines` instead.

create_engines ¶

create_engines(
    *,
    schema_claims: Iterable[SchemaClaim] = (),
    extensions: Sequence[Callable[[Any, Any], None]] = (),
    execution_options: dict[str, Any] | None = None,
    owner: str | None = None,
    register_claims: bool = True,
    **kwargs: Any,
) -> tuple[Engine, Engine]

Return (primary_engine, vocab_engine), the only way to get CDM engines.

The vocabulary engine is the primary engine itself when vocab_connection addresses the same physical database, so a colocated deployment opens one pool. Every parameter applies to both engines and is as for :meth:_build_engine.

occupied_schemas ¶

occupied_schemas(connection: Connection) -> set[str]

Physical schema names this database currently claims that live on connection's own physical server.

Scoped by :meth:roles_on_connection, so a schema belonging to a role hosted on a genuinely different physical server (e.g. vocab on its own connection) is never reported just because its name happens to coincide. Unlike checking every role unconditionally, this never needs to open a second engine to a different server.

roles_on_connection ¶

roles_on_connection(
    connection: Connection | Engine,
) -> tuple[Role, ...]

Role(s) this database entry has on connection's own physical server (host, database, port), or () if none.

Role.RESULTS is included whenever Role.PRIMARY is, since it has no connection of its own and always resolves through Role.PRIMARY's. When primary and vocab share one physical server, all three roles are returned.

route_for_schema_tag ¶

route_for_schema_tag(
    schema_tag: Role | str, *, vocab: _T, primary: _T
) -> _T

Determines the right connection, engine, or session to use for a given schema_tag in a split-connection database.

Role.VOCAB routes to vocab; Role.PRIMARY, Role.RESULTS, and any other string tag route to primary (Role.RESULTS has no connection of its own; it always resolves through Role.PRIMARY's).

Raises:

Type	Description
`TypeError`	If vocab and primary are not the same type, and aren't both members of Bindable.

schema_for_role ¶

schema_for_role(role: Role = Role.PRIMARY) -> str | None

Return the effective schema for a given role. See :meth:CDMDatabaseConfig.resolve for how vocab/results roles are handled.

Parameters:

Name	Type	Description	Default
`role`	`Role`	Which schema to return. Defaults to `Role.PRIMARY`.	`PRIMARY`

Returns:

Type	Description
`str or None`	`vocab_schema` for `Role.VOCAB`, `results_schema` for `Role.RESULTS`, `schema_name` for `Role.PRIMARY`.

Raises:

Type	Description
`ValueError`	If role is not recognised as a Role member.

schema_tags ¶

schema_tags() -> tuple[Role, ...]

Role tags whose schema provenance is worth tracking: primary, vocab, and results.

Schema-aware SQL primitives¶

Physical-schema/schema-tag-aware helpers. See Schema translate map for the distinction between a schema tag (an abstract schema_translate_map key) and a physical schema (an actual, literal database schema name).

Quoted, schema-qualified identifier, for raw SQL that's genuinely unavoidable. Utilises IdentifierPreparer.format_table to quote the name according to the dialect's rules.

Parameters:

Name	Type	Description	Default
`bindable`	`Engine \| Connection \| Session \| IdentifierPreparer`	The SQLAlchemy object whose dialect is used to quote the name, or an already-resolved IdentifierPreparer directly, for a caller with no live bindable in hand (e.g. a dialect-only preparer built ahead of any connection).	required
`name`	`str`	Unqualified identifier to quote.	required
`physical_schema`	`str or None`	The already-resolved schema to prefix with, or None for an unqualified name.	required

Returns:

Type	Description
`str`	The quoted identifier, schema-prefixed unless physical_schema is `None`, e.g. `"myschema"."mytable"` or `"mytable"`.

Opens its own transaction for an Engine, or uses an already-open Connection or Session directly, participating in the caller's own transaction.

Parameters:

Name	Type	Description	Default
`bindable`	`sqlalchemy.engine.Engine, sqlalchemy.engine.Connection, or sqlalchemy.orm.Session`	An Engine opens a new connection and transaction scoped to this context manager, committing on a clean exit. A Connection is forwarded as-is; its transaction is owned by the caller, and passing the same Connection into several calls groups them into one shared transaction. A Session reduces to its own live connection (via _as_bind), same ownership as a Connection: its transaction is the Session's own, not opened or committed here.	required

Yields:

Type	Description
`Connection`

Yield a Connection in AUTOCOMMIT isolation mode.

Parameters:

Name	Type	Description	Default
`bindable`	`Engine or Connection`	Given an `Engine`, opens a new connection for the duration of the `with` block and closes it on exit. Given an already-open `Connection`, yields it with the isolation level overridden, restored to what it was on exit rather than left mutated for the rest of the connection's life.	required

Notes

Connection has no .connect() of its own, so the two bindable arg types need different handling rather than one blind call chain.
The connection must not already have an active transaction: SQLAlchemy refuses to change isolation_level once one has started.

Yields:

Type	Description
`Connection`

CREATE SCHEMA IF NOT EXISTS, for Postgres and SQLite; not validated for other dialects.

Given a Connection, executes directly on it (no nested transaction), so it participates in a caller's already-open transaction. Given an Engine, opens its own short-lived transaction.

No-ops on a dialect with no multi-schema concept (e.g. SQLite), when physical_schema is None, or when physical_schema is that dialect's own default schema (e.g. Postgres's "public").

Notes

Unlike the schema primitives above, this parameter is required, not inferred, since creating the wrong schema by silent inference would be far worse than a missing default.

Schema provenance¶

See Schema provenance guard for the full explanation of what this guards against and when it no-ops.

The schema-provenance guard scoped to schema_tag's own physical schema.

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	Connection the guarded DDL runs on. Must have been built by create_engine(): the entry name and test_only are read back off its own execution_options, set there alongside schema_translate_map, so no separate `resolved` argument is needed.	required
`schema_tag`	`Role or str`	Schema tag being guarded.	required

Returns:

Type	Description
`AbstractContextManager[None]`

Raises:

Type	Description
`ValueError`	If connection wasn't built by create_engine() (its execution_options carry neither the entry name nor test_only).

Look up schema_tag's physical schema in the bindable's schema_translate_map.

Parameters:

Name	Type	Description	Default
`bindable`	`Engine \| Connection \| Session`		required
`schema_tag`	`str or None`	The schema_translate_map key to look up. Defaults to Role.PRIMARY.	`PRIMARY`

Returns:

Type	Description
`str or None`	None if schema_tag is None, or if the dialect has no schema concept. Otherwise the mapped physical schema.

Raises:

Type	Description
`UnregisteredSchemaTagError`	If schema_tag has no schema_translate_map entry on a schema-capable dialect. There used to be a fallback here treating schema_tag as a literal, already-registered physical schema name; it was removed because it silently accepted a typo'd tag that happened to collide with some other tag's physical name.

Schemas, other than physical_schema, that already have a table named table_name. Excludes the dialect's own system schemas.

Bases: RuntimeError

A database's physical schema no longer matches its recorded provenance.

Bases: RuntimeError

A schema tag or a reserved physical schema is already claimed by a different owner on the same connection.

Bases: RuntimeError

The schema_registry table on a connection predates its current layout.

Bases: RuntimeError

A schema_tag fell through to a literal physical schema that was never registered on this connection.

Resources¶

ConnectionConfig¶

dialect_name property ¶

build_url ¶

physical_identity ¶

physical_key ¶

resolve ¶

safe_url ¶

to_env_pairs ¶

DatabaseKind¶

DatabaseConfig¶

connection_name_for_role ¶

GenericDatabaseConfig¶

resolve ¶

CDMDatabaseConfig¶

connection_name_for_role ¶

resolve ¶

Role¶

Dialect¶

Resolved types¶

dialect_name property ¶

addresses_same_database_as ¶

create_engine ¶

physical_key ¶

configured_internal_schema_translate_map ¶

connection_for_role ¶

connection_for_schema_tag ¶

create_engine ¶

foreign_key_can_span ¶

occupied_schemas ¶

resolved_physical_schemas ¶

roles_on_connection ¶

route_for_schema_tag ¶

schema_for_role ¶

schema_tags ¶

tags_share_a_transaction ¶

cdm_schema property ¶

configured_internal_schema_translate_map ¶

connection_for_role ¶

connection_for_schema_tag ¶

create_engine ¶

create_engines ¶

occupied_schemas ¶

roles_on_connection ¶

route_for_schema_tag ¶

schema_for_role ¶

schema_tags ¶

Schema-aware SQL primitives¶

Schema provenance¶

dialect_name `property` ¶

dialect_name `property` ¶

cdm_schema `property` ¶