Query contracts

Consider a procedure recorded on the same day as two overlapping treatment episodes. The procedure may already have a valid Episode_Event link, or it may need to be assigned from dates alone. A reliable query has to answer several questions explicitly: what identifies the procedure, whether an explicit link takes precedence, whether fallback may attach it to one or both episodes, and how equally plausible candidates are ordered.

The contracts on this page provide a common vocabulary for those decisions. They are small, immutable values that can be shared by query-building code, configuration, and tests. Projection and predicate helpers translate the contracts into SQLAlchemy statements without opening a database connection.

Construction and execution

Creating a contract, statement, CTE, or predicate is side-effect free. Database access begins only when the resulting statement is executed through a connection or session. The toolkit supplies projection, attachment, hierarchy, temporal, concept-set, mapping, and observation builders.

Start with event identity

OMOP primary keys are scoped to their source tables. These three rows represent three different events even though they all use event ID 7:

Source table	Event ID	Person	Date
Measurement	7	101	20 January 2026
Procedure Occurrence	7	101	20 January 2026
Observation	7	202	20 January 2026

ClinicalEventIdentity keeps the table and numeric ID together:

from omop_alchemy.toolkit.core.events import ClinicalEventIdentity

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

assert len({measurement, procedure, observation}) == 3

A cross-table projection needs more than an identity. ClinicalEventColumn.required_columns() defines the labels a consumer can rely on:

Column	Meaning
person_id	Person who owns the source event
event_id	Primary key in the source table
event_source_table	Table that scopes event_id
event_field_concept_id	OMOP Field concept naming the source table's ID column
event_date	Date used for temporal selection
event_datetime	Source datetime when one is available
event_concept_id	Primary clinical concept carried by the event

Numeric value, value concept, and unit labels are available through ClinicalEventColumn.optional_columns() when a source table supports them.

The Field concept is not interchangeable with the event's clinical concept. For example, a Procedure Occurrence projection uses the Field concept for procedure_occurrence.procedure_occurrence_id as its discriminator and the row's procedure_concept_id as its clinical concept.

Build one shared event stream by passing the source models to canonical_event_union():

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,
).subquery("clinical_events")

All branches expose the same labels. The source table and Field concept are literals derived from model metadata, so they remain available after the tables are combined.

Attach an event to an episode

A complete attachment key adds the episode ID to the table-scoped event identity:

from omop_alchemy.toolkit.episodes.derivation import EpisodeAttachmentIdentity

attachment = EpisodeAttachmentIdentity.from_event(
    procedure,
    episode_id=1002,
)

assert attachment.event == procedure

Before accepting an explicit link, the query confirms that the event ID and Field concept identify a row in the supplied projection and that the event and episode belong to the same person. A link carrying another table's Field concept is outside that projection's scope. This is important because event IDs are unique only within their OMOP table: a Procedure Occurrence 7 link says nothing about Measurement 7, even when both numbers happen to be present.

Attachment diagnostics therefore do not infer a discriminator error or missing target from non-matching rows. An arbitrary event projection may intentionally be filtered, so absence from it does not prove anything about the underlying OMOP table. ResolvedEpisodeEvent.event_resolution_diagnostics checks the target table named by the Field concept and reports unsupported fields or dangling_event when the row truly does not exist. A link that names an existing row but is clinically incorrect cannot be identified from the linkage columns alone.

Once valid, an explicit link takes precedence under either explicit-first policy. Suppose Procedure Occurrence 7 is linked to episode 1002, while its date also falls inside the windows of episodes 1001 and 1002. The result is only (procedure_occurrence, 7, 1002): fallback must not add episode 1001 or duplicate episode 1002.

EpisodeAttachmentPolicy controls what happens when no valid explicit link exists:

Policy	Fallback behaviour
explicit_only	Leave the event unattached
explicit_first_ranked	Select one date-eligible episode using a separate ranking specification
explicit_first_all_in_window	Retain every date-eligible episode

flowchart TD Start["Event from canonical projection"] --> Valid{"Valid explicit
Episode_Event link?"} Valid -- "yes" --> Explicit["Attach via the explicit link
(precedence; fallback is not applied)"] Valid -- "no" --> Policy{"EpisodeAttachmentPolicy"} Policy -- "explicit_only" --> Unattached["Leave unattached"] Policy -- "explicit_first_ranked" --> Ranked["Rank date-eligible episodes
(TemporalRankingSpec)"] --> One["Attach to one episode"] Policy -- "explicit_first_all_in_window" --> Window["Every date-eligible episode
in the window"] --> Many["Attach to each eligible episode"]

Choosing between ranked and all-in-window fallback is a statement about result grain. Ranked fallback produces at most one episode per event. All-in-window fallback intentionally allows one event to appear against several overlapping episodes.

episode_attachment_queries() applies the complete precedence rule. It accepts a canonical event statement or one supported event model, validates explicit links against Episode_Event, and applies fallback only to events that have no valid explicit link:

from omop_alchemy.toolkit.episodes.derivation import (
    EpisodeAttachmentPolicy,
    EpisodeAttachmentDiagnostic,
    TemporalRankingSpec,
    TemporalSelectionPolicy,
    episode_attachment_queries,
)

attachment_queries = episode_attachment_queries(
    events,
    policy=EpisodeAttachmentPolicy.explicit_first_ranked,
    ranking=TemporalRankingSpec(
        policy=TemporalSelectionPolicy.nearest,
        stable_id_column="episode_id",
    ),
    include_diagnostics=True,
)

attachments = session.execute(attachment_queries.attachments).mappings().all()
assert attachment_queries.diagnostics is not None
diagnostics = session.execute(attachment_queries.diagnostics).mappings().all()
typed_diagnostics = [
    EpisodeAttachmentDiagnostic.from_mapping(row)
    for row in diagnostics
]

The attachment result preserves the event projection and adds episode_id and attachment_method. Its uniqueness key is (event_source_table, event_id, episode_id). A valid explicit link may legitimately connect an event to more than one episode; each relationship remains a separate attachment under that key.

Diagnostics are advisory rows and do not change the attachments. They identify cross-person links, fallback ambiguity, and events for which no valid explicit link or fallback candidate exists. EpisodeAttachmentDiagnostic.from_mapping() converts a raw SQLAlchemy mapping into a typed value carrying the event identity, projected and linked Field concepts, episode, candidate count, and message.

Rank fallback candidates

Ranking has two independent parts: which side of the anchor date should be considered first, and how candidates on that side should be ordered. Keeping them separate supports both symmetric nearest-date matching and the common preference for an episode that had already started when the event occurred.

For an event on 20 January, consider these episode starts:

Episode	Start date	Absolute distance	State on 20 January
1001	15 January	5 days	Already started
1003	21 January	1 day	Not yet started

A side-neutral nearest policy selects episode 1003:

from omop_alchemy.toolkit.episodes.derivation import (
    TemporalRankingSpec,
    TemporalSelectionPolicy,
)

absolute_nearest = TemporalRankingSpec(
    policy=TemporalSelectionPolicy.nearest,
    stable_id_column="episode_id",
)

If the analysis should prefer an episode that was underway when the event happened, apply a side preference before distance:

from omop_alchemy.toolkit.episodes.derivation import TemporalSidePreference

already_started_first = TemporalRankingSpec(
    policy=TemporalSelectionPolicy.nearest,
    stable_id_column="episode_id",
    side_preference=TemporalSidePreference.on_or_before_anchor,
)

flowchart TD A["Event on 20 January"] --> N{"policy = nearest"} A --> S{"policy = nearest,
side_preference =
on_or_before_anchor"} N -->|"closest by absolute distance"| E1003["Episode 1003
starts 21 Jan · 1 day away"] S -->|"already-started side considered first"| E1001["Episode 1001
starts 15 Jan · 5 days away"]

This policy selects episode 1001. Absolute distance still orders episodes within the preferred side; it simply does not allow a closer future episode to outrank every episode that had already started. on_or_after_anchor expresses the corresponding future-first rule.

Apply the policy to SQLAlchemy columns with temporal_order_expressions():

from omop_alchemy.cdm.model.structural import Episode
from omop_alchemy.toolkit.episodes.derivation import temporal_order_expressions

ordering = temporal_order_expressions(
    Episode.episode_start_date,
    events.c.event_date,
    Episode.episode_id,
    already_started_first,
)

candidate_episodes = candidate_episodes.order_by(*ordering)

earliest and latest are available when chronological position, rather than distance from the anchor, defines the result. Every policy ends with the named stable ID column in ascending order. If episodes 1001 and 1002 are otherwise tied, 1001 wins consistently rather than relying on database return order.

Date boundaries

Lower and upper bounds are inclusive by default and can be changed independently through EpisodeWindowSpec. For an episode starting 15 January 2026 with a 90-day prior window, 17 October 2025 lies exactly on the lower boundary and is included under the default. If the episode ends on 5 February, that date is included while 6 February is not.

Window admission and candidate ranking are separate decisions. EpisodeWindowSpec owns the finite interval and its open or closed boundaries. TemporalRankingSpec is accepted only by ranked fallback and describes how already-admitted candidates are ordered.

episode_window_predicate() uses the same finite defaults as the in-memory episode window. It honours a recorded episode end and substitutes a bounded post-start end only when the end is missing:

from omop_alchemy.toolkit.episodes.derivation import (
    EpisodeWindowSpec,
    episode_window_predicate,
)

window = EpisodeWindowSpec(
    days_prior=90,
    open_end_fallback_days=365,
    include_lower_bound=True,
    include_upper_bound=True,
)

inside_episode_window = episode_window_predicate(
    events.c.event_date,
    Episode.episode_start_date,
    Episode.episode_end_date,
    window=window,
)

Select one repeated observation

Repeated observations need the same explicit treatment of direction, grouping, and ties. For an anchor date of 20 January, suppose a person has these rows:

Observation ID	Date	Value
21	1 January	earlier
22	20 January	anchor-a
23	20 January	anchor-b
24	21 January	after-anchor

The following specification chooses the latest observation on or before the anchor, grouping rows by person and observation concept:

from omop_alchemy.toolkit.episodes.derivation import (
    ObservationSelectionPolicy,
    ObservationSelectionSpec,
)

selection = ObservationSelectionSpec(
    policy=ObservationSelectionPolicy.latest_on_or_before_anchor,
    partition_by=("person_id", "observation_concept_id"),
    stable_id_column="observation_id",
    include_anchor_date=True,
)

Use ranked_observation_select() to apply the anchor filter before calculating row numbers:

from datetime import date

from sqlalchemy import literal, select

from omop_alchemy.cdm.model import Observation
from omop_alchemy.toolkit.episodes.derivation import ranked_observation_select

ranked = ranked_observation_select(
    Observation.__table__,
    selection,
    anchor_date=literal(date(2026, 1, 20)),
).subquery("ranked_observations")

selected = select(ranked).where(ranked.c.observation_rank == 1)

Observation 24 is after the anchor and is therefore excluded. Observations 22 and 23 tie on date, so the stable ID selects 22. That tie-break creates reproducible output; it does not claim that one same-day clinical value is more correct. If every same-day value is meaningful, retain them by choosing a result grain that includes the observation ID instead of reducing the group to one row.

Add episode_id or another field to partition_by when selection must occur separately within those groups. The partition is part of the clinical meaning of the result, not merely an optimisation detail.

Runtime concept sets

Applications often receive concept selection as configuration rather than as a compile-time governed unit. RuntimeConceptSetSpec records four inputs: exact concepts and descendants to include, and exact concepts and descendants to exclude.

from omop_alchemy.toolkit.core.concepts import RuntimeConceptSetSpec

concepts = RuntimeConceptSetSpec(
    include_ancestor_ids=(100,),
    include_exact_ids=(900,),
    exclude_ancestor_ids=(400,),
    exclude_exact_ids=(901,),
    require_standard=True,
    include_classification=False,
)

The intended set is:

(descendants of 100 OR exact concept 900)
AND NOT (descendants of 400 OR exact concept 901)

Exclusion wins when a concept is reached from both sides. With no inclusion, the set matches nothing. IDs are sorted and deduplicated when the specification is created.

require_standard and include_classification apply while expanding ancestor descendants. Exact IDs are explicit configuration and are not removed if the deployed vocabulary is temporarily out of step with the configuration source, including after a concept has been de-standardised. The specification does not decide whether an exact numeric ID is present, active, standard, or classification in a particular vocabulary; validate and report those expectations at the configuration boundary without changing membership silently.

runtime_concept_predicate() translates the specification into database-side concept_ancestor and concept predicates:

from sqlalchemy import select

from omop_alchemy.cdm.model import Procedure_Occurrence
from omop_alchemy.toolkit.core.concepts import runtime_concept_predicate

matching_procedures = select(Procedure_Occurrence).where(
    runtime_concept_predicate(
        Procedure_Occurrence.procedure_concept_id,
        concepts,
    )
)

Constructing the specification or predicate performs no hierarchy expansion and no database access. Descendants are resolved by the database when the surrounding statement is executed. Exact IDs are rendered as parameters, so very large externally supplied lists should be staged as rows and joined rather than pushed through a single IN predicate.

Some applications compose positive and negative rules independently rather than collecting them into one runtime set. descendant_concept_select() provides the lower-level hierarchy operation for that case and returns each matching descendant once:

from omop_alchemy.toolkit.core.concepts import descendant_concept_select

matching_procedures = select(Procedure_Occurrence).where(
    Procedure_Occurrence.procedure_concept_id.in_(
        descendant_concept_select((100, 200))
    ),
    Procedure_Occurrence.procedure_concept_id.not_in(
        descendant_concept_select((400,))
    ),
)

Use RuntimeConceptSetSpec when the inclusions and exclusions form one configured set with exclusion precedence. Use descendant_concept_select() when the surrounding query or rule model owns how separate predicates are combined.

Canonical modifier queries

ModifierIdentity(modifier_source_table, modifier_id) and ModifierTargetIdentity(target_field_concept_id, target_event_id) make both table scopes explicit. The canonical source columns are:

Role	Columns
Source identity	modifier_source_table, modifier_id
Target identity	target_field_concept_id, target_event_id
Clinical row	person_id, modifier_date, modifier_datetime, modifier_concept_id
Nullable values	value_as_number, value_as_concept_id, unit_concept_id, value_as_string

Target validation is intentionally performed before selection. A valid link matches target Field concept, target ID, and person. This prevents a modifier for Condition Occurrence 7 from competing with one for Procedure Occurrence 7, and prevents a malformed cross-person link from replacing valid evidence.

from omop_alchemy.cdm.model import Condition_Occurrence, Measurement
from omop_alchemy.toolkit.core.modifiers import (
    ModifierSelectionPolicy,
    ModifierSelectionSpec,
    modifier_target_queries,
    selected_modifier_select,
)

resolved = modifier_target_queries(Measurement, Condition_Occurrence)
selected = selected_modifier_select(
    resolved.matches,
    spec=ModifierSelectionSpec(policy=ModifierSelectionPolicy.earliest),
)

The default selection partition includes person and both target identity columns. If one input contains several modifier categories and selection should occur separately for each, filter to one category before ranking or add that category discriminator to partition_by. Incomplete target identities are excluded. The stable source table and modifier ID are always the final tie-breakers under the default contract.

modifier_target_queries(..., diagnostics=True) returns a second advisory query covering missing_target_identity, unsupported_target_field, missing_target_event, and person_mismatch. Diagnostics do not change the valid result. For a caller-supplied selectable, only missing identity and observed person mismatches are reported: a filtered result cannot prove that an event is absent from the underlying table or that its Field is unsupported.

Diagnostics support inspection and validation independently of selection. Consume SQLAlchemy mappings directly, or convert them with the thin typed adapters. The adapters do not execute queries or alter the valid matches:

from omop_alchemy.toolkit.core.modifiers import ModifierTargetDiagnostic

checked = modifier_target_queries(
    Measurement, Condition_Occurrence, diagnostics=True,
)
assert checked.diagnostics is not None
diagnostic_rows = session.execute(checked.diagnostics).mappings().all()
typed_diagnostics = [
    ModifierTargetDiagnostic.from_mapping(row) for row in diagnostic_rows
]

EpisodeAttachmentDiagnostic provides the corresponding convenience for attachment diagnostics, as shown above. These are exported downstream validation contracts; their presence does not imply an in-package workflow or a scheduled consumer integration.

Oncology stage preference composes with this generic selector. Its public default is pathological, clinical, then unclassified, followed by earliest time. StageSelectionSpec.clinical_first(), StageSelectionSpec.chronological_only(), and a latest temporal policy are explicit query-scoped alternatives.

API reference

Canonical, domain-neutral clinical-event identities and row shapes.

Event tables use different native column names, but cross-table analytical queries need one stable vocabulary. This area provides both the shared row contracts and SQLAlchemy projections. Building a projection is side-effect free; the database is accessed only when a caller executes the returned statement.

ClinicalEventColumn

Bases: StrEnum

Canonical labels emitted by a cross-table clinical-event projection.

optional_columns classmethod

optional_columns() -> tuple[ClinicalEventColumn, ...]

Nullable value labels, excluding inherited required fields.

required_columns classmethod

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

Required projection labels in row-contract order.

ClinicalEventIdentity dataclass

ClinicalEventIdentity(
    event_source_table: str, event_id: int
)

Cross-table event identity.

OMOP event IDs are unique only within their source table. A Measurement and a Procedure Occurrence may legitimately have the same numeric ID, so the table is a mandatory part of identity.

ClinicalEventRow

Bases: Protocol

Value-level view of the required canonical event projection.

SQLAlchemy Row objects and small dataclasses can both satisfy this protocol. It describes the output consumed by downstream tools; it does not require a session-bound ORM entity.

ValuedClinicalEventRow

Bases: ClinicalEventRow, Protocol

Canonical event row extended with nullable value and unit fields.

canonical_event_projection

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

Project one supported OMOP event model to canonical event columns.

canonical_event_union

canonical_event_union(
    *models: type[Any], include_values: bool = True
) -> sa.Select[Any] | sa.CompoundSelect[Any]

Combine supported event models into one canonical UNION ALL query.

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.

Construct episodes and resolve the relationships between them.

Episodes in the CDM are rows that reference each other through episode_parent_id and resolve clinical facts through Episode_Event.

The public contracts in this area define episode-attachment identities and policies used by query builders. Shared clinical-event row names and identities live in toolkit.core.events.

Projection, attachment, and ranking helpers return SQLAlchemy statements or expressions without executing them.

CANONICAL_ATTACHMENT_DIAGNOSTIC_COLUMNS module-attribute

CANONICAL_ATTACHMENT_DIAGNOSTIC_COLUMNS: tuple[
    AttachmentDiagnosticColumn, ...
] = tuple(AttachmentDiagnosticColumn)

Columns exposed by an attachment diagnostics query.

CANONICAL_EPISODE_COLUMNS module-attribute

CANONICAL_EPISODE_COLUMNS: tuple[EpisodeColumn, ...] = (
    tuple(
        column
        for column in EpisodeColumn
        if column is not episode_concept_name
    )
)

Columns exposed by the canonical episode projection.

CANONICAL_EPISODE_OPTIONAL_COLUMNS module-attribute

CANONICAL_EPISODE_OPTIONAL_COLUMNS: tuple[
    EpisodeColumn, ...
] = (episode_concept_name,)

Opt-in descriptive columns exposed by an enriched episode projection.

AttachmentDiagnosticCode

Bases: StrEnum

Stable categories for explaining rejected or ambiguous attachment rows.

AttachmentDiagnosticColumn

Bases: StrEnum

Stable labels emitted by an attachment diagnostics query.

EpisodeAttachmentDiagnostic dataclass

EpisodeAttachmentDiagnostic(
    code: AttachmentDiagnosticCode,
    event: ClinicalEventIdentity,
    event_field_concept_id: int,
    message: str,
    linked_event_field_concept_id: int | None = None,
    episode_id: int | None = None,
    candidate_count: int | None = None,
)

Typed advisory result returned by an attachment diagnostics query.

from_mapping classmethod

from_mapping(
    row: _EpisodeAttachmentDiagnosticMapping,
) -> "EpisodeAttachmentDiagnostic"

Convert one SQLAlchemy mapping result without leaking column-name handling.

EpisodeAttachmentIdentity dataclass

EpisodeAttachmentIdentity(
    event_source_table: str, event_id: int, episode_id: int
)

Unique identity of one event attached to one episode.

event property

event: ClinicalEventIdentity

The event portion of this attachment identity.

from_event classmethod

from_event(
    event: ClinicalEventIdentity, *, episode_id: int
) -> EpisodeAttachmentIdentity

Add an episode to an already canonical cross-table event identity.

EpisodeAttachmentMethod

Bases: StrEnum

How an event-to-episode attachment was established.

EpisodeAttachmentPolicy

Bases: StrEnum

Named precedence and fallback-cardinality policies.

A valid explicit link always wins in the two explicit-first policies. The difference is what happens to an event that has no valid explicit link.

permits_fallback_fanout property

permits_fallback_fanout: bool

Whether one fallback event may attach to several episodes.

requires_fallback_ranking property

requires_fallback_ranking: bool

Whether fallback needs a separate temporal ranking specification.

uses_fallback property

uses_fallback: bool

Whether unlinked events may be attached by a date window.

EpisodeAttachmentQueries dataclass

EpisodeAttachmentQueries(
    attachments: Select[Any],
    diagnostics: Select[Any] | None = None,
)

Attachment results and, when requested, their advisory diagnostics.

attachments preserves the input event columns and appends episode_id and attachment_method. Its executable uniqueness key is (event_source_table, event_id, episode_id).

diagnostics is None unless requested. Diagnostics explain rejected explicit links and fallback outcomes; they do not alter attachment rows.

EpisodeColumn

Bases: StrEnum

Canonical labels emitted by an episode projection.

EpisodeWindowSpec dataclass

EpisodeWindowSpec(
    days_prior: int = DEFAULT_EPISODE_WINDOW_DAYS_PRIOR,
    open_end_fallback_days: int = DEFAULT_EPISODE_OPEN_END_FALLBACK_DAYS,
    include_lower_bound: bool = True,
    include_upper_bound: bool = True,
)

Finite episode-relative window used to admit fallback event candidates.

InvalidAttachmentSourceError

Bases: ValueError

Raised when an attachment input does not expose its required columns.

ObservationSelectionPolicy

Bases: StrEnum

Supported deterministic choices for repeated longitudinal observations.

ObservationSelectionSpec dataclass

ObservationSelectionSpec(
    policy: ObservationSelectionPolicy,
    partition_by: tuple[str, ...] = (
        "person_id",
        "observation_concept_id",
    ),
    stable_id_column: str = "observation_id",
    include_anchor_date: bool = True,
)

Partition and tie-break contract for repeated-observation selection.

The default partition describes one observation concept for one person. Callers may add an episode or another grouping field when their declared result grain requires it.

requires_anchor property

requires_anchor: bool

Whether selection requires a caller-supplied anchor date.

TemporalRankingSpec dataclass

TemporalRankingSpec(
    policy: TemporalSelectionPolicy,
    stable_id_column: str,
    side_preference: TemporalSidePreference = TemporalSidePreference.none,
)

Temporal ranking contract for SQL builders.

A side preference, when present, is applied before the selection policy. nearest then means the smallest absolute distance within that tier. All policies use the named stable ID column as their final ascending tie-breaker, so the same source rows cannot alternate across executions.

has_side_preference property

has_side_preference: bool

Whether candidates are tiered by their side of the anchor first.

uses_absolute_distance property

uses_absolute_distance: bool

Whether ranking uses absolute distance after any side-preference tier.

TemporalSelectionPolicy

Bases: StrEnum

How one row is selected from several temporal candidates.

TemporalSidePreference

Bases: StrEnum

Which side of an anchor is preferred before temporal ranking.

Candidate dates are ranked relative to a caller-supplied anchor. For event attachment where the event is the anchor and episode starts are candidates, on_or_before_anchor prefers an episode that had already started.

absolute_day_delta

absolute_day_delta(
    candidate_date: ColumnElement[Any],
    anchor_date: ColumnElement[Any],
) -> sa.ColumnElement[int]

Return absolute calendar-day distance between candidate and anchor.

bounded_temporal_predicate

bounded_temporal_predicate(
    value: ColumnElement[Any],
    lower_bound: ColumnElement[Any],
    upper_bound: ColumnElement[Any],
    *,
    include_lower_bound: bool = True,
    include_upper_bound: bool = True,
) -> sa.ColumnElement[bool]

Test a value against independently open or closed temporal bounds.

canonical_episode_projection

canonical_episode_projection(
    episode_model: EpisodeSource = Episode,
    *,
    include_concept_label: bool = False,
) -> sa.Select[Any]

Project stable episode fields, optionally including its concept name.

direct_episode_relationship_projection

direct_episode_relationship_projection(
    episode_model: EpisodeSource = Episode,
) -> sa.Select[Any]

Select direct parent-child pairs with a depth of one.

episode_attachment_queries

episode_attachment_queries(
    events: type[Any] | FromClause | SelectBase,
    *,
    policy: EpisodeAttachmentPolicy,
    episodes: type[Episode]
    | FromClause
    | SelectBase = Episode,
    episode_events: type[Episode_Event]
    | FromClause
    | SelectBase = Episode_Event,
    ranking: TemporalRankingSpec | None = None,
    window: EpisodeWindowSpec = EpisodeWindowSpec(),
    include_diagnostics: bool = False,
) -> EpisodeAttachmentQueries

Build explicit-first attachments from canonical event and episode inputs.

Parameters:

Name	Type	Description	Default
events	type[Any] | FromClause | SelectBase	A supported event model or selectable exposing the canonical event columns.	required
policy	EpisodeAttachmentPolicy	Whether to use explicit links only, ranked fallback, or every eligible episode in the fallback window.	required
episodes	type[Episode] | FromClause | SelectBase	An Episode model or selectable exposing episode identity, person and date bounds.	Episode
episode_events	type[Episode_Event] | FromClause | SelectBase	An Episode_Event model or selectable containing episode/event links and their Field-concept discriminator.	Episode_Event
ranking	TemporalRankingSpec | None	Temporal ranking used by explicit_first_ranked. It must be omitted for policies that do not rank fallback candidates.	None
window	EpisodeWindowSpec	Episode-relative date window used to admit fallback candidates.	EpisodeWindowSpec()
include_diagnostics	bool	If True, return an advisory diagnostic selectable as well as the attachment query. Building the queries does not execute them.	False

Returns:

Type	Description
EpisodeAttachmentQueries	attachments preserves event columns and appends episode_id and attachment_method. diagnostics is None unless requested.

Notes

Resolution proceeds in three stages: validate explicit links, admit same-person fallback candidates within the episode-relative window, then retain every admitted candidate or apply temporal ranking according to policy. Explicit links suppress fallback only after the event ID, Field-concept discriminator, episode ID and person all agree; the final result is deduplicated by event source, event ID and episode ID. Fallback ambiguity counts distinct eligible episodes, even when inputs repeat rows.

Diagnostics explain rejected links and fallback outcomes without changing attachment rows. Because inputs may be filtered selectables, the builder does not infer missing source rows or unsupported discriminators from their absence.

episode_descendants

episode_descendants(
    *,
    root_episode_id: int | ColumnElement[Any] | None = None,
    episode_model: EpisodeSource = Episode,
    include_root: bool = True,
    max_depth: int = 100,
    name: str = "episode_descendants",
) -> sa.CTE

Build a recursive root-to-descendant projection with bounded depth.

episode_event_hierarchy_projection

episode_event_hierarchy_projection(
    *,
    root_episode_id: int | ColumnElement[Any] | None = None,
    episode_model: EpisodeSource = Episode,
    episode_event_model: EpisodeEventSource = Episode_Event,
    include_root: bool = True,
    max_depth: int = 100,
) -> sa.Select[Any]

Select linked events with their root episode, owning episode, and depth.

episode_window_bounds

episode_window_bounds(
    episode_start_date: ColumnElement[Any],
    episode_end_date: ColumnElement[Any],
    *,
    window: EpisodeWindowSpec = EpisodeWindowSpec(),
) -> tuple[sa.ColumnElement[Any], sa.ColumnElement[Any]]

Build the bounded SQL interval used for date-admitted episode facts.

Closed episodes use their recorded end date. Open episodes use the configured fallback horizon so an absent end date cannot silently produce an unbounded join.

episode_window_predicate

episode_window_predicate(
    event_date: ColumnElement[Any],
    episode_start_date: ColumnElement[Any],
    episode_end_date: ColumnElement[Any],
    *,
    window: EpisodeWindowSpec = EpisodeWindowSpec(),
) -> sa.ColumnElement[bool]

Test whether an event lies inside a bounded episode-relative window.

observation_eligibility_predicate

observation_eligibility_predicate(
    observation_date: ColumnElement[Any],
    spec: ObservationSelectionSpec,
    *,
    anchor_date: ColumnElement[Any] | None = None,
) -> sa.ColumnElement[bool]

Return the date predicate required by an observation selection policy.

observation_order_expressions

observation_order_expressions(
    observation_date: ColumnElement[Any],
    stable_id: ColumnElement[Any],
    spec: ObservationSelectionSpec,
) -> tuple[sa.ColumnElement[Any], ...]

Build date and stable-ID ordering for repeated observations.

observation_row_number

observation_row_number(
    columns: Any,
    *,
    observation_date_column: str,
    spec: ObservationSelectionSpec,
    label: str = "observation_rank",
) -> sa.ColumnElement[int]

Return a deterministic row number using the declared observation grain.

The caller supplies column names rather than mapped attributes so this helper works against raw OMOP sources and projected/aliased selects alike.

ranked_observation_select

ranked_observation_select(
    source: FromClause,
    spec: ObservationSelectionSpec,
    *,
    observation_date_column: str = "observation_date",
    anchor_date: ColumnElement[Any] | None = None,
    rank_label: str = "observation_rank",
) -> sa.Select[Any]

Select source columns with deterministic rank and eligibility.

Eligibility is applied in the same select that computes the window rank, keeping out-of-window rows from competing for rank one. The source's existing columns are preserved and the rank is appended for downstream projection or filtering.

shift_date

shift_date(
    value: ColumnElement[Any], *, days: int
) -> sa.ColumnElement[Any]

Shift a date by a fixed number of days on PostgreSQL or SQLite.

signed_day_delta

signed_day_delta(
    candidate_date: ColumnElement[Any],
    anchor_date: ColumnElement[Any],
) -> sa.ColumnElement[int]

Return candidate minus anchor in whole calendar days.

temporal_order_expressions

temporal_order_expressions(
    candidate_date: ColumnElement[Any],
    anchor_date: ColumnElement[Any],
    stable_id: ColumnElement[Any],
    ranking: TemporalRankingSpec,
) -> tuple[sa.ColumnElement[Any], ...]

Build deterministic ordering for a temporal ranking contract.

temporal_row_number

temporal_row_number(
    candidate_date: ColumnElement[Any],
    anchor_date: ColumnElement[Any],
    stable_id: ColumnElement[Any],
    ranking: TemporalRankingSpec,
    *,
    partition_by: Iterable[ColumnElement[Any]] = (),
    label: str = "temporal_rank",
) -> sa.ColumnElement[int]

Return a deterministic row number for temporal candidates.