Core services

The core package handles problems that have the same meaning in every clinical domain: resolving a source term to an OMOP concept, identifying an event across CDM tables, arranging events on a timeline, and converting measurements to comparable units.

Generic database lifecycle operations do not belong in the clinical toolkit. Use orm-loader to define, create, refresh, index, and drop materialised views. OMOP Alchemy owns the OMOP-specific query and row-grain decisions supplied to that infrastructure; a downstream application owns its registry, dependency policy, and deployment orchestration. See Materialised views for the integration boundary.

Resolve source data to concepts

Suppose an intake system supplies the text Adenocarcinoma of lung rather than an OMOP concept ID. A resolver limits the eligible vocabulary rows and applies the same text normalisation when it builds its lookup and when it handles an incoming value:

from omop_alchemy.toolkit.core.concepts import make_concept_resolver

resolver = make_concept_resolver(
    session,
    name="condition lookup",
    domain_id="Condition",
)

concept_id = resolver.lookup("Adenocarcinoma of lung")

Creating the resolver reads the vocabulary tables, so create it once for a mapping workflow and reuse it. LookupSpec, LookupIndex, and ConceptResolver expose the individual stages when you need to control indexed fields, normalisation, or resolver lifetime. ConceptResolverRegistry provides lazy construction and caching when an application maintains several lookups.

Concept groups answer the complementary question: whether a known concept belongs to a governed set. A resolved group supports both in-memory membership and a SQLAlchemy expression derived from the same specification, so filtering loaded objects and filtering in SQL do not require separate definitions.

Domain packages can declare module-level governed groups with SemanticUnitRef(value_set, unit). The reference loads the optional omop-semantics runtime only when its parent, excluded-parent, or exact IDs are read. It always exposes a complete governed unit; narrower group definitions belong in omop-semantics, not in consumer-side adapters.

Configuration-driven concept sets use RuntimeConceptSetSpec. It records exact and ancestral inclusions and exclusions without touching the database; see Runtime concept sets for the set semantics and current execution boundary.

Resolve concepts to standard concepts

Most source concepts resolve to one standard concept. Some source concepts represent several clinical meanings, however, and OMOP maps those to several standard concepts. standard_concept_mapping_select() therefore returns one row per valid Maps to relationship rather than choosing one target:

from datetime import date

from omop_alchemy.toolkit.core.concepts import (
    StandardConceptMappingSpec,
    standard_concept_mapping_select,
)

mapping_query = standard_concept_mapping_select(
    StandardConceptMappingSpec(
        source_concept_ids=(source_concept_id,),
        valid_on=date(2026, 1, 1),
    )
)

mapping_rows = session.execute(mapping_query).mappings().all()

Each row carries the source and standard concept identifiers, vocabularies, codes, and names alongside the relationship validity dates. Invalid relationships, invalid targets, non-standard targets, and other relationship types are excluded. A standard concept's Maps to self-map is returned normally.

Supplying valid_on makes the relationship and target date ranges part of the query, which is useful when a result must be reproducible against a dated vocabulary release. The query does not follow replacement relationships or Maps to value: those relationships answer different questions and should be handled by purpose-specific queries when a toolkit consumer needs them.

Resolve free-text terms and source codes to OMOP concept IDs.

Source data rarely arrives with concept IDs attached. This module turns a declarative description of which concepts are eligible into a runtime resolver that maps incoming text to those concepts, applying the same normalisation on both sides so that matching is predictable.

Three pieces make up the workflow:

LookupSpec Declares which concepts belong in a lookup — by vocabulary, domain, concept class, or explicit ancestry — and which text fields are indexed.

LookupIndex The materialised table of normalised text keys to concept IDs that a spec produces against the vocabulary tables.

ConceptResolver Wraps an index and resolves terms at runtime, applying the same normalisation used to build the keys.

make_concept_resolver bundles all three for the common case::

from omop_alchemy.toolkit.core.concepts import make_concept_resolver

resolver = make_concept_resolver(
    session,
    name="condition lookup",
    domain_id="Condition",
)
concept_id = resolver.lookup("Adenocarcinoma of lung")

Normalisation is composable. compose_normalizers chains individual rules — normalize_default for whitespace and casing, strip_uicc and make_stage for staging text, site_to_NOS for site generalisation — so that a resolver's matching behaviour is stated explicitly rather than implied.

Building an index queries the vocabulary tables, so resolvers are worth reusing. ConceptResolverRegistry constructs each resolver on first access and caches it for the registry's lifetime.

Governed concept sets are the other half of this module. Where a resolver maps text to concepts, a concept group answers whether a concept ID belongs to a governed clinical set::

from omop_alchemy.toolkit.core.concepts import (
    ConceptGroupSpec,
    resolve_concept_group,
)

RT = ConceptGroupSpec(name="radiotherapy", unit=...)   # governed anchors

group = resolve_concept_group(session, RT)
procedure.procedure_concept_id in group        # Python, O(1)
group.expression(Procedure_Occurrence.procedure_concept_id)   # SQL

Both access paths derive from one spec, so they cannot disagree. Expansions are cached per vocabulary, not per engine, so recreating an engine against the same database does not re-run the closure queries — see :mod:.identity for how an engine declares which vocabulary it reads, and note that a caller building its own engines must register that itself.

Nothing here touches a database at import time: specs are declarative and registries build on first request.

STANDARD_CONCEPT_MAPPING_COLUMNS module-attribute

STANDARD_CONCEPT_MAPPING_COLUMNS: tuple[
    StandardConceptMappingColumn, ...
] = tuple(StandardConceptMappingColumn)

Columns exposed by a standard concept mapping query.

STANDARD_CONCEPT_MAPPING_UNIQUENESS module-attribute

STANDARD_CONCEPT_MAPPING_UNIQUENESS: tuple[
    StandardConceptMappingColumn, ...
] = (source_concept_id, standard_concept_id)

Executable uniqueness key of a standard concept mapping result.

CacheStats dataclass

CacheStats(
    entries: int = 0,
    cached_bytes: int = 0,
    evictions: int = 0,
    rebuilds_after_evict: int = 0,
)

Observability for a bounded registry.

rebuilds_after_evict is the load-bearing number. Evicting an entry that is never asked for again is exactly what a bound is for; evicting one that is then rebuilt is thrashing, and it is otherwise invisible because it presents as ordinary slowness rather than as a cache problem. While it stays at zero the bound is correct.

ConceptGroupAnchors

Bases: Protocol

Role-aware anchors consumed by :class:ConceptGroupSpec.

ConceptGroupRegistry

ConceptGroupRegistry(
    engine: Engine,
    *,
    max_bytes: int = DEFAULT_MAX_CACHE_BYTES,
)

Bases: _LazyBoundedRegistry[ResolvedConceptGroup]

Lazy registry for governed concept groups, scoped to one vocabulary.

Obtain one through :func:concept_group_registry rather than constructing it directly, so registries are shared per vocabulary identity instead of per engine.

register_spec

register_spec(spec: ConceptGroupSpec) -> None

Register spec under its governed name, if not already present.

ConceptGroupSpec dataclass

ConceptGroupSpec(
    name: str,
    unit: ConceptGroupAnchors,
    include_descendants: bool = True,
    require_standard: bool = False,
    include_classification: bool = True,
)

A governed omop-semantics semantic unit plus how to expand it.

Declarative and side-effect free — constructing one performs no I/O and touches no database.

Parameters:

Name	Type	Description	Default
name	str	Stable identifier for this group, used as the cache key. Use the governed semantic-unit name so the key derives from the governed identity rather than a locally invented label.	required
unit	ConceptGroupAnchors	The omop-semantics RuntimeSemanticUnit supplying anchors. Read lazily, so a spec can be declared at module scope without loading the semantics runtime. The unit supplies mixed-role anchors: parent_ids expand through descendants while exact_ids are matched directly.	required
include_descendants	bool	Expand parent_ids through concept_ancestor. When False only the anchors themselves are members.	True
require_standard	bool	Restrict expansion to concepts carrying a standardness flag. Defaults to False, so expansion reads concept_ancestor without a standard filter. OMOPConceptSource.descendants defaults to requiring standard concepts.	False
include_classification	bool	Widens require_standard to admit classification ('C') concepts. Defaults to True so a governed group can be anchored on a classification node (ATC, for example) without silently losing it. Has no effect unless require_standard is set.	True

exact_ids

exact_ids() -> tuple[int, ...]

Governed members matched directly, without descendant expansion.

excluded_parent_ids

excluded_parent_ids() -> tuple[int, ...]

Governed anchors whose descendants are subtracted.

expression_for

expression_for(
    column: SQLColumnExpression[Any],
) -> sa.ColumnElement[bool]

SQL membership for this group, as a subquery over concept_ancestor.

Available on the spec because the SQL form needs no session: the traversal is performed by the database when the query runs. That is what lets a hybrid_property expose the same governed set at class level, where no session exists.

Deliberately a subquery rather than a literal IN list built from a resolved group: a closure of tens of thousands of IDs degrades query plans and can exceed driver parameter limits.

parent_ids

parent_ids() -> tuple[int, ...]

Governed descendant-expanding anchors.

ConceptResolver

ConceptResolver(
    index: LookupIndex,
    *,
    normalizer: Normaliser | None = None,
    corrections: list[Callable[[str], str]] | None = None,
)

Runtime resolver for mapping free-text terms to OMOP concept IDs.

A ConceptResolver wraps a pre-built LookupIndex and applies runtime normalisation and optional correction passes to resolve arbitrary input strings to concept IDs. It is intentionally lightweight and stateless: all semantic scope and vocabulary constraints are encoded upstream in the LookupSpec and LookupIndex.

Resolution proceeds in ordered stages: 1. Apply the primary normaliser to the input term and attempt a direct lookup. 2. If no hit is found, apply each correction function in turn, re-normalise, and retry the lookup. 3. If no match is found, return the configured unknown concept ID.

This design allows simple, explicit handling of common data quality issues (e.g. formatting differences, legacy codes, mild normalisation errors) without introducing fuzzy matching, probabilistic scoring, or hidden inference logic.

Parameters:

Name	Type	Description	Default
index	LookupIndex	Pre-built LookupIndex providing the normalised key → concept_id mapping.	required
normalizer	Normaliser | None	Optional normalisation function applied to input terms at lookup time. Defaults to normalize_default. This should be compatible with the normaliser used when constructing the LookupIndex.	None
corrections	list[Callable[[str], str]] | None	Optional ordered list of correction functions applied to the raw input term prior to normalisation and lookup. Each correction is tried in sequence until a match is found.	None

Notes

• ConceptResolver performs no database access and no dynamic expansion of vocabularies; it operates over the materialised LookupIndex.
• Resolution is deterministic and transparent: there is no fuzzy matching, ranking, or probabilistic inference.
• Correction functions are applied conservatively and in-order; later corrections do not override earlier successful matches.
• lookup_exact bypasses correction passes and performs a single normalised lookup, which is useful for validation and debugging.

Examples:

>>> resolver = ConceptResolver(index)
>>> resolver.lookup("Stage III")
123456
>>> resolver.lookup("stage-3")
123456

Bind a materialized index to runtime normalization and corrections.

all_concepts cached property

all_concepts: set[int]

Every concept ID reachable through this resolver's index.

Cached because the index is fixed at construction. Returned by reference, so treat it as read-only.

__contains__

__contains__(item: str | int) -> bool

Test corrected text membership or direct concept-ID membership.

estimated_bytes

estimated_bytes() -> int

Approximate retained size, for cache accounting.

A name/code/synonym index costs several times more per concept than a bare ID set, which is why the cache bound is measured in bytes rather than entry counts. Measured at ~350 bytes per concept for OMOP-length names and codes with one synonym each.

lookup

lookup(term: str | None) -> int | None

Resolve a term, trying direct lookup before ordered corrections.

lookup_exact

lookup_exact(term: str | None) -> int | None

Resolve only the normalized input, bypassing correction functions.

ConceptResolverRegistry

ConceptResolverRegistry(
    engine: Engine,
    *,
    max_bytes: int = DEFAULT_MAX_CACHE_BYTES,
)

Bases: _LazyBoundedRegistry[ConceptResolver]

Lazy registry for ConceptResolvers.

Resolvers are constructed on first access and cached for the lifetime of this registry instance. The registry is scoped to a SQLAlchemy Engine, ensuring vocab lookups are built once per database.

LookupIndex dataclass

LookupIndex(
    name: str, unknown: int | None, mapping: dict[str, int]
)

Materialised lookup table from normalised text keys to OMOP concept IDs.

A LookupIndex is the runtime artifact produced by a LookupSpec and a ConceptSource. It represents a flat, precomputed mapping from one or more normalised string representations (e.g. concept names, codes, synonyms) to OMOP concept IDs.

Attributes:

Name	Type	Description
name	str	Human-readable identifier for the lookup.
unknown	int | None	Concept ID to return when a lookup fails, or None if failures should propagate as null.
mapping	dict[str, int]	Dictionary mapping normalised string keys to OMOP concept IDs. Keys are expected to already be normalised at build time.

Notes

The mapping may contain multiple textual representations pointing to the same concept ID (e.g. name + code + synonym).

all_concepts property

all_concepts: set[int]

Return the concept IDs represented by this materialized index.

__contains__

__contains__(item: str | int) -> bool

Test membership by indexed key or by reachable concept ID.

lookup

lookup(term: str | None) -> int | None

Resolve an already-normalized key, returning the configured fallback.

LookupSpec dataclass

LookupSpec(
    name: str,
    unknown: int | None = 0,
    domain_id: str | None = None,
    concept_class_id: list[str] | None = None,
    vocabulary_id: list[str] | None = None,
    require_standard: bool = True,
    include_classification: bool = True,
    require_active: bool = False,
    code_filter: str | None = None,
    parents: list[int] | None = None,
    include_non_standard_descendants: bool = False,
    include_synonyms: bool = False,
    normalizer: Normaliser = normalize_default,
    include: tuple[str, ...] = (
        "concept_name",
        "concept_code",
    ),
)

Declarative specification for constructing a vocabulary lookup index.

A LookupSpec defines what concepts should be included in a lookup and which textual representations should be indexed for resolution.

The spec is consumed by a OMOPConceptSource, which materialises a LookupIndex by querying an OMOP vocabulary source and extracting the requested fields.

This separation allows lookup semantics (domain, vocabulary, hierarchy, standardness, synonyms, normalisation) to be expressed explicitly and versioned independently of runtime resolution logic.

Attributes:

Name	Type	Description
name	str	Stable identifier for this lookup specification.
unknown	int | None	Concept ID to return for unmatched terms. Set to None to preserve nulls, or to a sentinel concept ID to force closed-world behaviour.
domain_id	str | None	Optional OMOP domain filter
concept_class_id	list[str] | None	Optional list of OMOP concept_class_id values to restrict the lookup
vocabulary_id	list[str] | None	Optional list of OMOP vocabulary_id values to restrict the lookup
require_standard	bool	If True, restricts the lookup to concepts carrying a standardness flag. Named to match ConceptFilter.require_standard, which it delegates to.
include_classification	bool	Widens require_standard to admit classification ('C') concepts. Defaults to True: a lookup exists to recognise vocabulary terms, and a classification concept is a legitimate thing to recognise even though it is not a valid mapping target. Set False for selection-shaped lookups.
require_active	bool	If True, excludes concepts with an invalid_reason. Defaults to False so recognition stays permissive — a deprecated concept that matches the text can still be resolved forward through "Maps to" / "Concept replaced by", whereas filtering it out here discards the term entirely.
code_filter	str | None	Optional substring filter applied to concept_code (ILIKE-based). Useful for coarse scoping (e.g. AJCC-only codes).
parents	list[int] | None	Optional list of ancestor concept IDs from which to expand the lookup via the Concept_Ancestor table.
include_non_standard_descendants	bool	If True, includes non-standard concepts when expanding from parents. Has no effect if parents is None.
include_synonyms	bool	If True, include Concept_Synonym entries in the lookup keys.
normalizer	Normaliser	Function applied to all indexed strings at build time. This should match (or be compatible with) the normalisation used at resolution time by ConceptResolver.
include	tuple[str, ...]	Tuple of ConceptRow attribute names to index as keys (e.g. ("concept_name", "concept_code")). This controls which textual fields become resolvable inputs.

Notes

• LookupSpec encodes semantic intent; LookupIndex encodes runtime state.
• Specs are designed to be stable, inspectable configuration objects that can be versioned and reviewed as part of phenotype or ETL definitions.
• Normalisation and correction policies are intentionally split between build-time (this spec) and runtime (ConceptResolver) to make lookup behaviour explicit and testable.

OMOPConceptSource

Concrete ConceptSource backed by OMOP CDM vocabulary tables.

It is a thin, explicit adapter between SQLAlchemy + OMOP CDM and higher-level vocabulary indexing logic.

Used exclusively to builds a query based on provided parameters (adds filter for each non-None parameter, and joins to Concept_Ancestor if parents are specified).

build_lookup staticmethod

build_lookup(
    session: Session, spec: LookupSpec
) -> LookupIndex

Materialize one scoped lookup and its optional synonym keys.

The returned index is intentionally detached from the session. If multiple selected representations normalize to the same key, the later materialized assignment wins; callers that need collision-free semantics should narrow the LookupSpec rather than rely on row ordering.

descendants staticmethod

descendants(
    session: Session,
    parents: list[int],
    *,
    include_non_standard: bool = False,
) -> list[int]

Return descendant IDs using the source's standardness policy.

fetch_concepts staticmethod

fetch_concepts(
    session: Session,
    *,
    domain_id: str | None = None,
    concept_class_id: Iterable[str] | None = None,
    vocabulary_id: Iterable[str] | None = None,
    require_standard: bool = True,
    include_classification: bool = True,
    require_active: bool = False,
    code_filter: str | None = None,
    parents: Iterable[int] | None = None,
    include_non_standard_descendants: bool = False,
) -> list[ConceptRow]

Fetch concepts matching the provided constraints.

This method supports two primary modes: 1. Flat filtering by domain / class / vocabulary 2. Hierarchical expansion from parent concept(s)

Domain, vocabulary, standardness and validity are delegated to :class:~omop_alchemy.cdm.query.ConceptFilter so this layer shares one implementation of the OMOP flag rules with the rest of the package. In particular the flag comparisons are normalised, so a blank or whitespace-padded standard_concept is classified the same way here as it is by Concept.is_standard.

Recognition is permissive by default, because this layer resolves text to a concept, not a concept to a mapping target. Both defaults follow from that:

• include_classification is True — classification concepts are legitimate things to recognise, even though they are not valid mapping targets.
• require_active is False — a deprecated concept that matches the text is a useful hit, because the caller can follow Maps to / Concept replaced by to a valid successor. Filtering it out at recognition time discards the term entirely, with nothing to resolve forward from.

Pass include_classification=False / require_active=True when the index feeds concept selection rather than recognition, where the strict OMOP mapping-target rules apply.

fetch_synonyms staticmethod

fetch_synonyms(
    session: Session,
    *,
    concept_ids: Iterable[int] | None = None,
) -> list[tuple[int, str]]

Return (concept_id, synonym) pairs for concept synonyms.

The join to concept scopes the result to synonyms whose concept actually exists, and concept_ids narrows it further to a caller-supplied set. Both are applied in SQL: without them this streams every synonym row in the vocabulary back to Python to be discarded, which on a full Athena load is millions of rows for a lookup covering a handful of domains.

ResolvedConceptGroup dataclass

ResolvedConceptGroup(
    spec: ConceptGroupSpec, ids: frozenset[int]
)

A governed group expanded against one vocabulary.

Membership is (descendants(parents) - descendants(excluded parents)) | exact_ids. Exact members are explicit inclusions and are never subtracted by the exclusion step, matching omop-semantics' composition contract.

estimated_bytes

estimated_bytes() -> int

Approximate retained size, for cache accounting.

Measured at ~80 bytes per ID for a frozenset of OMOP-magnitude concept IDs: a distinct PyLong each, plus hash-table overhead at a 0.6 load factor.

expression

expression(
    column: SQLColumnExpression[Any],
) -> sa.ColumnElement[bool]

SQL membership for this group — delegates to the spec.

Kept here so a caller holding a resolved group can reach either rendering without going back to the spec, but the definition lives in one place.

RuntimeConceptSetSpec dataclass

RuntimeConceptSetSpec(
    include_ancestor_ids: tuple[int, ...] = (),
    include_exact_ids: tuple[int, ...] = (),
    exclude_ancestor_ids: tuple[int, ...] = (),
    exclude_exact_ids: tuple[int, ...] = (),
    require_standard: bool = False,
    include_classification: bool = True,
)

Runtime include/exclude inputs for a database-side concept predicate.

The intended expression is::

(included ancestor descendants OR included exact IDs)
AND NOT (excluded ancestor descendants OR excluded exact IDs)

Exclusion therefore wins if the same concept is reached by both sides. Empty inclusions describe an always-false set. Constructing the spec is side-effect free and preserves no session-bound vocabulary objects.

require_standard and include_classification deliberately match ConceptFilter and ConceptGroupSpec for descendant expansion. Exact IDs remain explicit inclusions even when the deployed vocabulary has de-standardised one of them; configuration validation may report that drift without silently changing set membership. Descendant rendering delegates to the existing normalised Concept flag expressions.

IDs are sorted and deduplicated only. Validity rules for configuration or a local vocabulary belong at those boundaries, not in this generic spec.

has_inclusions property

has_inclusions: bool

Whether the predicate can match at least one configured input.

SemanticUnitRef dataclass

SemanticUnitRef(value_set: str, unit: str)

Lazy, immutable reference to one complete governed semantic unit.

StandardConceptMappingColumn

Bases: StrEnum

Stable labels emitted by :func:standard_concept_mapping_select.

StandardConceptMappingSpec dataclass

StandardConceptMappingSpec(
    source_concept_ids: tuple[int, ...] = (),
    valid_on: date | None = None,
)

Select valid Maps to relationships for the requested source concepts.

An empty source_concept_ids tuple selects every source. valid_on can be supplied when a reproducible historical vocabulary view is required; it applies to both the relationship and its standard target. Invalid relationships and targets are always excluded.

build_concept_group

build_concept_group(
    session: Session, spec: ConceptGroupSpec
) -> ResolvedConceptGroup

Expand a governed group against the vocabulary behind session.

Performs up to two queries — the include closure and, when the group declares exclusions, the exclude closure. Callers should normally go through the cache in registry rather than calling this directly.

clear_concept_group_cache

clear_concept_group_cache() -> None

Drop every cached group expansion, across all vocabularies.

Per-vocabulary keying means this is rarely needed — moving database gives a different identity and therefore a different registry. It remains an escape hatch for a dataset reloaded in place under an unchanged identity.

clear_vocabulary_identity

clear_vocabulary_identity(engine: Engine) -> None

Forget engine's registered identity, if it had one.

concept_group_cache_stats

concept_group_cache_stats() -> dict[
    str | int, dict[str, int]
]

Per-scope cache statistics, for monitoring the bound.

Watch rebuilds_after_evict: while it is zero the shared bound is holding, and if it climbs the per-scope byte totals are the evidence for raising it or splitting the budget by payload kind.

concept_group_registry

concept_group_registry(
    session: Session,
) -> ConceptGroupRegistry

Return the group registry for the vocabulary behind session.

Registries are keyed on vocabulary identity where one has been registered (see :mod:.identity), so recreating an engine against the same dataset reuses expansions. Otherwise they are keyed weakly on the engine, which is still built-once-per-engine but is not shared across engines.

In-memory SQLite intentionally lands in the second case: two such engines are separate databases despite identical configuration, so cross-engine sharing would serve one database's concept sets for another.

descendant_concept_select

descendant_concept_select(
    ancestor_ids: Iterable[int],
    *,
    require_standard: bool = False,
    include_classification: bool = True,
) -> sa.Select[Any]

Select each descendant ID once for the supplied ancestors.

make_concept_resolver

make_concept_resolver(
    session: Session,
    *,
    name: str,
    unknown: int | None = 0,
    domain_id: str | None = None,
    concept_class_id: list[str] | None = None,
    vocabulary_id: list[str] | None = None,
    require_standard: bool = True,
    include_classification: bool = True,
    require_active: bool = False,
    code_filter: str | None = None,
    parents: list[int] | None = None,
    include_non_standard_descendants: bool = False,
    include_synonyms: bool = False,
    include: tuple[str, ...] = (
        "concept_name",
        "concept_code",
    ),
    build_normalizer: Normaliser = normalize_default,
    runtime_normalizer: Normaliser | None = None,
    corrections: list[Callable[[str], str]] | None = None,
) -> ConceptResolver

Convenience factory for constructing a ConceptResolver from declarative inputs.

This function bundles the common workflow of: - defining a LookupSpec - materialising a LookupIndex from OMOP - constructing a ConceptResolver for runtime use

Parameters:

Name	Type	Description	Default
session	Session	Active SQLAlchemy session connected to the OMOP CDM database.	required
name	str	Stable identifier for this lookup specification, used in logging and debugging.	required
unknown	int | None	Concept ID to return for unmatched terms. Set to None to preserve nulls, or to a sentinel concept ID to force closed-world behaviour.	0
domain_id	str | None	Optional OMOP domain filter for the concepts to include in the lookup.	None
concept_class_id	list[str] | None	Optional list of OMOP concept_class_id values to restrict the lookup.	None
vocabulary_id	list[str] | None	Optional list of OMOP vocabulary_id values to restrict the lookup.	None
require_standard	bool	If True, restricts the lookup to concepts carrying a standardness flag. Named to match ConceptFilter.require_standard, which it delegates to.	True
include_classification	bool	Widens require_standard to admit classification ('C') concepts. Defaults to True: a lookup exists to recognise vocabulary terms, and a classification concept is a legitimate thing to recognise even though it is not a valid mapping target. Set False for selection-shaped lookups.	True
require_active	bool	If True, excludes concepts with an invalid_reason. Defaults to False so recognition stays permissive — a deprecated concept that matches the text can still be resolved forward through "Maps to" / "Concept replaced by", whereas filtering it out here discards the term entirely.	False
code_filter	str | None	Optional substring filter applied to concept_code (ILIKE-based). Useful for coarse scoping (e.g. AJCC-only codes).	None
parents	list[int] | None	Optional list of ancestor concept IDs from which to expand the lookup via the Concept_Ancestor table.	None
include_non_standard_descendants	bool	If True, includes non-standard concepts when expanding from parents. Has no effect if parents is None.	False
include_synonyms	bool	If True, include Concept_Synonym entries in the lookup keys.	False
include	tuple[str, ...]	Tuple of ConceptRow attribute names to index as keys (e.g. ("concept_name", "concept_code")). This controls which textual fields become resolvable inputs.	('concept_name', 'concept_code')
build_normalizer	Normaliser	Normalisation function applied to all indexed strings at build time. This should match (or be compatible with) the normaliser used at resolution time by ConceptResolver.	normalize_default
runtime_normalizer	Normaliser | None	Optional normalisation function applied to input terms at lookup time. Defaults to normalize_default. This should be compatible with the normaliser used when constructing the LookupIndex.	None
corrections	list[Callable[[str], str]] | None	Optional ordered list of correction functions applied to the raw input term prior to normalisation and lookup	None

register_vocabulary_identity

register_vocabulary_identity(
    engine: Engine, identity: str
) -> None

Declare which vocabulary dataset engine reads.

Concept-set caches keyed on this identity are shared by every engine registered under it, so recreating an engine reuses the expansion.

Pass the engine your factory returns. ResolvedCDMDatabase.create_engine ends with execution_options(schema_translate_map=...), which yields a derived OptionEngine; that is the object sessions bind to, and the one lookups will see.

Do not register an identity for an ephemeral database — notably in-memory SQLite, where two engines built from identical configuration are genuinely separate databases. Those correctly fall back to per-engine caching.

resolve_concept_group

resolve_concept_group(
    session: Session, spec: ConceptGroupSpec
) -> ResolvedConceptGroup

Expand spec against session's vocabulary, cached.

The cache key is spec.name within the vocabulary's registry, so it derives from the governed semantic-unit name rather than a locally invented label.

runtime_concept_predicate

runtime_concept_predicate(
    column: SQLColumnExpression[Any],
    spec: RuntimeConceptSetSpec,
) -> sa.ColumnElement[bool]

Render runtime concept membership entirely as database predicates.

standard_concept_mapping_select

standard_concept_mapping_select(
    spec: StandardConceptMappingSpec,
) -> sa.Select[Any]

Select each valid, single-hop Maps to standard-concept mapping.

Results remain relational because OMOP permits one source concept to map to more than one standard concept. Standard-concept self-maps are returned as ordinary rows; no recursive traversal is required.

Identify events across CDM tables

A Measurement and a Procedure Occurrence may have the same numeric primary key. Code that combines tables must therefore carry the source table as part of event identity:

from omop_alchemy.toolkit.core.events import ClinicalEventIdentity

measurement = ClinicalEventIdentity("measurement", 7)
procedure = ClinicalEventIdentity("procedure_occurrence", 7)

assert measurement != procedure

ClinicalEventColumn defines the common labels used when heterogeneous event tables are projected into one result. The required shape includes the person, table-scoped event identity, event date and datetime, clinical concept, and OMOP Field concept that identifies the source ID column. Optional labels cover numeric values, value concepts, and units.

ClinicalEventColumn.required_columns() and optional_columns() expose these groups as ordered enum tuples derived from the field-only row contracts.

canonical_event_union() turns supported event models into that shared shape. Measurement and Observation retain numeric values, value concepts and units; Observation string values are outside this projection. Sources without those fields receive typed nulls so every branch of the union remains compatible:

from omop_alchemy.cdm.model import (
    Measurement,
    Observation,
    Procedure_Occurrence,
)
from omop_alchemy.toolkit.core.events import canonical_event_union

events = canonical_event_union(
    Measurement,
    Observation,
    Procedure_Occurrence,
)

for event in session.execute(events).mappings():
    print(event["event_source_table"], event["event_id"], event["event_date"])

The projection resolves its ID, clinical concept, date, source table, and Field concept through stable CDM event metadata shared with episode-event resolution. Bare Measurement, Observation, and Device_Exposure classes remain lightweight mappings for ETL, while their analytical views provide reference context, domain validation, and episode-event resolution. Importing analytics modules cannot change either the Core projection metadata or the default resolution target. UnsupportedClinicalEventModelError is raised before SQL execution when no supported CDM definition exists.

The six clinical analytical Views inherit ClinicalEventMixin, which extends ModifierTargetMixin with has_complete_metadata() and clinical_event_model_spec(). The spec method can validate either the View's columns or those of a supplied bare source model. The explicit registry still controls event membership: EpisodeView is a modifier target rather than a clinical event, and custom modifier sources may combine ModifierSourceMixin with ClinicalEventMixin without registering as episode-event targets.

from omop_alchemy.cdm.model.clinical import Measurement, MeasurementView

assert MeasurementView.has_complete_metadata()
spec = MeasurementView.clinical_event_model_spec(Measurement)

flowchart LR M["Measurement
measurement_id"] --> U["canonical_event_union()"] O["Observation
observation_id"] --> U P["Procedure_Occurrence
procedure_occurrence_id"] --> U U --> S["canonical shape
person_id · event_id · event_source_table
event_field_concept_id · event_date · event_concept_id
(+ value / value_concept / unit where supported)"]

Each source model contributes its own ID column and Field concept; branches without a value or unit receive typed nulls so the union stays one consistent shape regardless of which models are combined.

The query contracts explain how the canonical shape participates in episode attachment.

Resolve OMOP modifiers

Measurement and Observation both implement the OMOP polymorphic modifier link, but use different physical column names. canonical_modifier_projection() and canonical_modifier_union() normalize those tables to one shape containing a table-scoped modifier identity, a Field-concept-scoped target identity, the modifier date and concept, and all four OMOP value representations.

ModifierColumn.required_columns() and value_columns() expose the ordered modifier label groups without adding methods to the row Protocols.

Supported source and target models are declared explicitly in immutable metadata. Shared event fields and the six clinical target definitions are derived from the clinical-event registry; Episode is the only target extension. This keeps imports deterministic without maintaining a second copy of the CDM Field concepts and native event columns.

from omop_alchemy.cdm.model import Measurement, Observation
from omop_alchemy.toolkit.core.modifiers import canonical_modifier_union

modifiers = canonical_modifier_union(Measurement, Observation).subquery()

A numeric modifier ID is unique only inside its source table. Likewise, a numeric target ID is meaningful only with target_field_concept_id. Preserve both parts of both identities when a query is joined, ranked, or materialized.

Use modifier_target_queries() before reducing repeated modifiers. The accepted link must agree on target ID, target Field concept, and person. Supported targets are Condition Occurrence, Measurement, Observation, Procedure Occurrence, Drug Exposure, Device Exposure, and Episode. Episode is deliberately a modifier target without being added to the clinical-event union.

from omop_alchemy.cdm.model import Condition_Occurrence, Measurement
from omop_alchemy.toolkit.core.modifiers import modifier_target_queries

result = modifier_target_queries(
    Measurement,
    Condition_Occurrence,
    diagnostics=True,
)

valid_modifiers = result.matches
rejected_links = result.diagnostics

Diagnostics distinguish incomplete target identities, unsupported target Field concepts, missing target events, and cross-person links. Missing-row diagnostics for a caller-supplied filtered target projection are relative to that projection; use a complete target model when absence from the CDM itself is the question.

selected_modifier_select() then applies an explicit earliest/latest policy. The default partition is (person_id, target_field_concept_id, target_event_id); date, non-null datetime, source table, and modifier ID form its deterministic order. Rows without a complete target identity do not participate.

Canonical modifier projections, target validation, and selection.

InvalidModifierSourceError

Bases: ValueError

Raised when a modifier selection input lacks a required column.

InvalidModifierTargetSourceError

Bases: ValueError

Raised when a supplied projection lacks canonical columns.

ModifierColumn

Bases: StrEnum

Stable labels emitted by Measurement/Observation modifier projections.

Native columns such as measurement_event_id and observation_event_id converge on these names so downstream query code never needs to branch on the physical modifier source.

required_columns classmethod

required_columns() -> tuple[ModifierColumn, ...]

Required projection labels in row-contract order.

value_columns classmethod

value_columns() -> tuple[ModifierColumn, ...]

Nullable value labels, excluding inherited required fields.

ModifierIdentity dataclass

ModifierIdentity(
    modifier_source_table: str, modifier_id: int
)

Source-table-scoped identity of the modifier row itself.

ModifierRow

Bases: Protocol

Structural typing contract for the canonical identity and clinical row.

ModifierSelectionPolicy

Bases: StrEnum

Supported temporal directions after any caller-supplied priority.

ModifierSelectionSpec dataclass

ModifierSelectionSpec(
    policy: ModifierSelectionPolicy = ModifierSelectionPolicy.earliest,
    partition_by: tuple[str, ...] = (
        str(ModifierColumn.person_id),
        str(ModifierColumn.target_field_concept_id),
        str(ModifierColumn.target_event_id),
    ),
    date_column: str = str(ModifierColumn.modifier_date),
    datetime_column: str = str(
        ModifierColumn.modifier_datetime
    ),
    stable_identity_columns: tuple[str, ...] = (
        str(ModifierColumn.modifier_source_table),
        str(ModifierColumn.modifier_id),
    ),
)

Deterministic selection policy within a modifier target partition.

ModifierTargetDiagnostic dataclass

ModifierTargetDiagnostic(
    diagnostic_code: ModifierTargetDiagnosticCode,
    modifier_source_table: str,
    modifier_id: int,
    target_field_concept_id: int | None,
    target_event_id: int | None,
    message: str,
)

Typed value representation of one target-resolution diagnostic row.

ModifierTargetDiagnosticCode

Bases: StrEnum

Reasons a canonical modifier could not resolve to its supplied target.

ModifierTargetDiagnosticColumn

Bases: StrEnum

Stable output labels for advisory target-resolution diagnostics.

ModifierTargetIdentity dataclass

ModifierTargetIdentity(
    target_field_concept_id: int, target_event_id: int
)

Field-concept-scoped identity of the row being modified.

ModifierTargetModelSpec dataclass

ModifierTargetModelSpec(
    event_id_attribute: str,
    event_field_concept_id: int,
    event_source_table: str,
)

Native target identity and its canonical OMOP Field discriminator.

UnsupportedModifierSourceModelError

UnsupportedModifierSourceModelError(
    model: object, reason: str
)

Bases: UnsupportedModelError

Raised when a model cannot provide a canonical modifier projection.

UnsupportedModifierTargetError

UnsupportedModifierTargetError(model: object, reason: str)

Bases: UnsupportedModelError

Raised when a model cannot be a canonical modifier target.

ValuedModifierRow

Bases: ModifierRow, Protocol

Canonical modifier row extended with all nullable value representations.

canonical_modifier_projection

canonical_modifier_projection(
    model: type[Any], *, include_values: bool = True
) -> sa.Select[Any]

Project Measurement or Observation into one modifier row shape.

canonical_modifier_target_projection

canonical_modifier_target_projection(
    model: type[Any],
) -> sa.Select[Any]

Project a supported clinical event or Episode to target identity columns.

modifier_order_expressions

modifier_order_expressions(
    columns: ReadOnlyColumnCollection[str, Any],
    spec: ModifierSelectionSpec,
    *,
    priority: Sequence[ColumnElement[Any]] = (),
) -> tuple[sa.ColumnElement[Any], ...]

Return the complete, portable order for a modifier selection policy.

modifier_row_number

modifier_row_number(
    columns: ReadOnlyColumnCollection[str, Any],
    spec: ModifierSelectionSpec,
    *,
    priority: Sequence[ColumnElement[Any]] = (),
    label: str = MODIFIER_RANK,
) -> sa.ColumnElement[int]

Build the deterministic window rank for canonical modifier columns.

modifier_source_model_spec

modifier_source_model_spec(
    model: type[Any],
) -> ClinicalEventModelSpec

Resolve and validate metadata for any model declaring the modifier link.

modifier_target_model_spec

modifier_target_model_spec(
    model: type[Any],
) -> ModifierTargetModelSpec

Resolve and validate immutable metadata for a modifier target model.

modifier_target_queries

modifier_target_queries(
    modifier_source: type[Any] | FromClause | SelectBase,
    target_source: type[Any] | FromClause | SelectBase,
    *,
    include_unmatched: bool = False,
    diagnostics: bool = False,
) -> ModifierTargetQueries

Resolve valid target links and optionally explain rejected modifier rows.

Parameters:

Name	Type	Description	Default
modifier_source	type[Any] | FromClause | SelectBase	A supported modifier model or a selectable exposing the canonical modifier columns.	required
target_source	type[Any] | FromClause | SelectBase	A supported ORM target model or a selectable exposing target identity columns. Selectables are treated as the caller's supplied scope.	required
include_unmatched	bool	If True, retain modifier rows without a valid target in matches. Otherwise only valid links are returned.	False
diagnostics	bool	If True, also build an advisory diagnostic selectable. Building the queries does not execute either selectable.	False

Returns:

Type	Description
ModifierTargetQueries	matches contains the modifier columns plus resolved target identity columns. diagnostics is None unless requested.

Notes

A link is valid only when target event ID, target Field concept and person ID all agree. For an ORM target model, diagnostics can report an unsupported target field or a missing target event. For a caller-supplied selectable, those absences may be caused by filtering, so only missing identity and observed person mismatches are reported.

ranked_modifier_select

ranked_modifier_select(
    source: FromClause | SelectBase,
    spec: ModifierSelectionSpec = ModifierSelectionSpec(),
    *,
    priority: Sequence[ColumnElement[Any]] = (),
    rank_label: str = MODIFIER_RANK,
) -> sa.Select[Any]

Rank bound modifiers, with caller priorities preceding temporal policy.

selected_modifier_select

selected_modifier_select(
    source: FromClause | SelectBase,
    spec: ModifierSelectionSpec = ModifierSelectionSpec(),
    *,
    priority: Sequence[ColumnElement[Any]] = (),
) -> sa.Select[Any]

Select the first deterministically ranked modifier in each partition.

Parameters:

Name	Type	Description	Default
source	FromClause | SelectBase	A selectable containing canonical modifier columns and target identity columns.	required
spec	ModifierSelectionSpec	Temporal direction, partition columns and stable identity columns used to define the selection contract.	ModifierSelectionSpec()
priority	Sequence[ColumnElement[Any]]	Optional SQL expressions placed before the temporal policy, such as a domain-specific stage preference.	()

Returns:

Type	Description
Select	A selectable containing the source columns, with one row at rank one for each target partition. Rows missing either target identity column are excluded before selection.

Notes

The final tie-breakers come from spec.stable_identity_columns. This keeps the result deterministic when dates, datetimes and caller priorities are equal.

Work with a patient timeline

The timeline adapter presents conditions, measurements, observations, and drug exposures as a single ordered sequence while retaining each row's source identity and value semantics. Use it when an application needs to display or serialise a patient's chronology rather than build a set-based analytical query.

The timeline has a dedicated guide with session requirements, event mappings, and extension points: Patient timelines.

Convert body measurements

Body-size calculations require weights and heights to use consistent units. The default conversion rules use the unit concept recorded on each measurement:

from omop_alchemy.toolkit.core.units import default_body_unit_conversion_rules

rules = default_body_unit_conversion_rules()
weight_kg = rules.normalize_weight_kg(180.0, rules.units.lb)
height_cm = rules.normalize_height_cm(70.0, rules.units.inch)

An unknown unit, a missing unit, or a missing value produces None; values are never passed through as though they were already normalised. Deployments with local unit concepts can construct BodyUnitConversionRules with their own BodySizeUnitConcepts mapping.

Clinical choices built on those conversions, including which measurements constitute baseline weight and how change is graded, belong to analytics.body_metrics and analytics.adverse_events.

Convert measurement values to canonical units.

Measurement rows carry whatever unit the source system recorded. Before values can be compared, trended, or fed into a calculation they need to agree on a unit, and that conversion must be explicit about what it will and will not accept.

BodyUnitConversionRules converts anthropometric measurements to kilograms and centimetres, driven by the unit concept on each row rather than by guesswork. default_body_unit_conversion_rules supplies the standard rule set; INCH_TO_CM and LB_US_TO_KG are the underlying factors where a caller needs them directly.

A reading whose unit concept is unrecognised is not converted and not silently passed through — callers are told the value could not be normalised, so an unconvertible unit never reaches a calculation disguised as a valid one.

BodySizeUnitConcepts dataclass

BodySizeUnitConcepts(
    kg: int, lb: int, cm: int, inch: int, m2: int
)

The unit concepts anthropometric conversion recognises.

Kilograms, pounds, centimetres, inches, and square metres mean the same thing in every clinical domain, so the mapping from concept ID to unit lives here rather than with any one domain's measurement logic.

BodyUnitConversionRules dataclass

BodyUnitConversionRules(units: BodySizeUnitConcepts)

Unit normalisation for anthropometric measurements.

Conversion is driven by the unit concept recorded on each row. A value whose unit concept is not in units converts to None rather than being passed through, so a reading in an unrecognised unit cannot reach a calculation disguised as a valid one.

Deployments that record body size against non-standard unit concepts can supply their own units instead of the governed defaults.