geno_lewm.data.membership_store

membership_store

Scalable, lineage-bound membership artifacts for v0.3 datasets.

This stable public facade keeps the production-scale membership contract separate from its writer, verifier, storage, lineage, and runtime adapters. PyArrow is imported only by the Parquet build and verification paths.

MEMBERSHIP_STORE_SCHEMA_VERSION module-attribute

MEMBERSHIP_STORE_SCHEMA_VERSION: Final = 'geno-lewm.membership-store.v1'

Closed manifest and row schema for the scalable membership store.

LabeledClinVarMembership dataclass

LabeledClinVarMembership(membership: MembershipRow, clinical_significance: str)

One chromosome-assigned binary ClinVar label and its membership identity.

is_pathogenic property

is_pathogenic: bool

Return the closed binary target for this normalized ClinVar class.

MembershipSourceBinding dataclass

MembershipSourceBinding(source_id: str, kind: str, repository: str, revision: str, namespace: str, artifact_path: str, artifact_sha256: str, artifact_size_bytes: int, artifact_row_count: int, artifact_schema_version: str, verification_kind: str, verification_sha256: str, membership_row_count: int, filtered_row_count: int)

Lineage, staged-byte, and derived-row commitments for one source.

MembershipSourceInput dataclass

MembershipSourceInput(kind: str, path: Path, chromosome: str | None = None)

One local lineage-bound source shard consumed by the builder.

source_id property

source_id: str

Return the stable source identifier recorded on every derived row.

MembershipStoreFile dataclass

MembershipStoreFile(path: str, sha256: str, size_bytes: int)

Exact identity of one physical store file.

MembershipStoreManifest dataclass

MembershipStoreManifest(artifact_id: str, assembly: str, chromosome_roles: ChromosomeRoles, snapshot_lineage: SnapshotLineageBinding, sources: tuple[MembershipSourceBinding, ...], row_count: int, variant_count: int, role_counts: dict[str, int], source_counts: dict[str, int], source_role_counts: dict[str, dict[str, int]], source_kind_role_counts: dict[str, dict[str, int]], clinvar_class_role_counts: dict[str, dict[str, int]], rowset_sha256: str, files: tuple[MembershipStoreFile, ...], content_identity: str, schema_version: str = MEMBERSHIP_STORE_SCHEMA_VERSION)

Closed semantic and physical contract for one membership store.

physical_identity property

physical_identity: str

Return the exact identity of semantic and bound physical bytes.

MembershipStoreVerification dataclass

MembershipStoreVerification(manifest: MembershipStoreManifest, ok: bool = True)

Result of independently scanning a membership store.

SnapshotLineageBinding dataclass

SnapshotLineageBinding(lineage_id: str, sha256: str, size_bytes: int, candidate_snapshot_id: str)

Exact byte and semantic identity of the source snapshot lineage.

evidence_profile property

evidence_profile: str

Return whether the lineage passed official or fixture-only verification.

MembershipStore

MembershipStore(store_dir: Path, *, verify: bool = True)

Read-only, indexed membership lookup without a PyArrow dependency.

Source code in geno_lewm/data/_membership_store_runtime.py

def __init__(self, store_dir: Path, *, verify: bool = True) -> None:
    root = Path(store_dir).absolute()
    capture: _CapturedMembershipStore | None = None
    if verify:
        capture = _capture_membership_store(root)
        try:
            manifest = _verify_captured_membership_store(capture).manifest
        except Exception:
            capture.close()
            raise
        capture.retain_only({_INDEX_NAME})
        lookup_root = capture.root
    else:
        _verify_exact_layout(root)
        manifest = _read_manifest(root / _MANIFEST_NAME)
        lookup_root = root
    path = lookup_root / _INDEX_NAME
    if not path.is_file():
        raise InputError("membership lookup index is missing", details={"path": str(path)})
    self.root = root
    self.manifest = manifest
    self._lookup_root = lookup_root
    self._capture = capture
    self._capture_owner_pid = os.getpid() if capture is not None else None
    self._closed = False
    self._connections: weakref.WeakKeyDictionary[threading.Thread, _ThreadConnection] = (
        weakref.WeakKeyDictionary()
    )
    self._lock = threading.Lock()
    self._owner_pid = os.getpid()

open classmethod

open(store_dir: Path, *, verify: bool = True) -> MembershipStore

Open a store read-only, optionally running its full streaming verifier.

Source code in geno_lewm/data/_membership_store_runtime.py

@classmethod
def open(cls, store_dir: Path, *, verify: bool = True) -> MembershipStore:
    """Open a store read-only, optionally running its full streaming verifier."""
    return cls(store_dir, verify=verify)

close

close() -> None

Close this handle and every per-thread read-only connection it owns.

Source code in geno_lewm/data/_membership_store_runtime.py

def close(self) -> None:
    """Close this handle and every per-thread read-only connection it owns."""
    if not self._closed:
        with self._lock:
            for connection in self._connections.values():
                connection.close()
            self._connections.clear()
        self._closed = True
        if self._capture is not None and self._capture_owner_pid == os.getpid():
            self._capture.close()
        self._capture = None

__getstate__

__getstate__() -> dict[str, object]

Serialize stable bindings only; live SQLite handles never cross processes.

Source code in geno_lewm/data/_membership_store_runtime.py

def __getstate__(self) -> dict[str, object]:
    """Serialize stable bindings only; live SQLite handles never cross processes."""
    return {
        "root": str(self.root),
        "manifest": self.manifest,
        "closed": self._closed,
    }

contains_variant

contains_variant(variant: CanonicalVariant | str, *, roles: Sequence[str] = ('validation', 'evaluation'), sources: Sequence[str] | None = None) -> bool

Return whether variant has any indexed membership in roles.

Source code in geno_lewm/data/_membership_store_runtime.py

def contains_variant(
    self,
    variant: CanonicalVariant | str,
    *,
    roles: Sequence[str] = ("validation", "evaluation"),
    sources: Sequence[str] | None = None,
) -> bool:
    """Return whether ``variant`` has any indexed membership in ``roles``."""
    self._require_open()
    normalized = _canonical_variant(variant)
    selected_roles = _normalize_roles(roles)
    selected_sources = _normalize_sources(sources, self.manifest.source_counts)
    role_placeholders = ",".join("?" for _ in selected_roles)
    source_clause = ""
    parameters: tuple[object, ...] = (normalized.key, *selected_roles)
    if selected_sources is not None:
        source_placeholders = ",".join("?" for _ in selected_sources)
        source_clause = f" AND source IN ({source_placeholders})"
        parameters = (*parameters, *selected_sources)
    row = (
        self._connection()
        .execute(
            f"SELECT 1 FROM memberships WHERE variant_key = ? "
            f"AND role IN ({role_placeholders}){source_clause} LIMIT 1",
            parameters,
        )
        .fetchone()
    )
    return row is not None

overlaps_interval

overlaps_interval(chrom: str, *, start_bp: int, end_bp: int, roles: Sequence[str] = ('validation', 'evaluation'), sources: Sequence[str] | None = None) -> bool

Return whether an indexed variant intersects a 0-based half-open interval.

Source code in geno_lewm/data/_membership_store_runtime.py

def overlaps_interval(
    self,
    chrom: str,
    *,
    start_bp: int,
    end_bp: int,
    roles: Sequence[str] = ("validation", "evaluation"),
    sources: Sequence[str] | None = None,
) -> bool:
    """Return whether an indexed variant intersects a 0-based half-open interval."""
    self._require_open()
    chromosome = canonicalize_chromosome(chrom)
    _require_nonnegative_int(start_bp, "membership interval start_bp")
    _require_positive_int(end_bp, "membership interval end_bp")
    if end_bp <= start_bp:
        raise InputError("membership interval end_bp must be greater than start_bp")
    selected_roles = _normalize_roles(roles)
    selected_sources = _normalize_sources(sources, self.manifest.source_counts)
    role_placeholders = ",".join("?" for _ in selected_roles)
    source_clause = ""
    parameters: tuple[object, ...] = (
        _CHROMOSOME_RANK[chromosome],
        _CHROMOSOME_RANK[chromosome],
        end_bp,
        start_bp,
        *selected_roles,
    )
    if selected_sources is not None:
        source_placeholders = ",".join("?" for _ in selected_sources)
        source_clause = f" AND memberships.source IN ({source_placeholders})"
        parameters = (*parameters, *selected_sources)
    row = (
        self._connection()
        .execute(
            "SELECT 1 FROM membership_intervals AS intervals "
            "CROSS JOIN memberships AS memberships "
            "ON memberships.membership_id = intervals.membership_id "
            "WHERE intervals.chrom_min <= ? AND intervals.chrom_max >= ? "
            "AND intervals.start_min < ? AND intervals.end_max > ? "
            f"AND memberships.role IN ({role_placeholders}){source_clause} LIMIT 1",
            parameters,
        )
        .fetchone()
    )
    return row is not None

iter_role

iter_role(role: str, *, batch_size: int = 65536) -> Iterator[MembershipRow]

Stream one split role in canonical order with bounded Python memory.

Source code in geno_lewm/data/_membership_store_runtime.py

def iter_role(self, role: str, *, batch_size: int = 65_536) -> Iterator[MembershipRow]:
    """Stream one split role in canonical order with bounded Python memory."""
    self._require_open()
    selected_role = _normalize_roles((role,))[0]
    _require_positive_int(batch_size, "membership iteration batch_size")
    cursor = self._connection().execute(
        _SELECT_ROWS + " WHERE role = ? " + _ORDER_BY,
        (selected_role,),
    )
    while batch := cursor.fetchmany(batch_size):
        for raw in batch:
            yield _membership_row_from_sql(raw)

iter_labeled_clinvar

iter_labeled_clinvar(role: str, *, batch_size: int = 65536) -> Iterator[LabeledClinVarMembership]

Stream binary-labeled ClinVar rows for one chromosome-assigned role.

Source code in geno_lewm/data/_membership_store_runtime.py

def iter_labeled_clinvar(
    self, role: str, *, batch_size: int = 65_536
) -> Iterator[LabeledClinVarMembership]:
    """Stream binary-labeled ClinVar rows for one chromosome-assigned role."""
    self._require_open()
    selected_role = _normalize_roles((role,))[0]
    _require_positive_int(batch_size, "membership iteration batch_size")
    cursor = self._connection().execute(
        _SELECT_UNIQUE_CLINVAR_ROWS,
        (selected_role,),
    )
    while batch := cursor.fetchmany(batch_size):
        for raw in batch:
            yield _labeled_clinvar_membership_from_sql(raw)

MembershipStoreHoldoutPolicy

MembershipStoreHoldoutPolicy(store: MembershipStore)

Bases: HoldoutPolicy

Existing tuple-builder holdout adapter backed by :class:MembershipStore.

Source code in geno_lewm/data/_membership_store_runtime.py

def __init__(self, store: MembershipStore) -> None:
    if not isinstance(store, MembershipStore):
        raise InputError("membership holdout adapter requires MembershipStore")
    super().__init__(
        holdout_chroms=(
            *store.manifest.chromosome_roles.validation,
            *store.manifest.chromosome_roles.evaluation,
        )
    )
    object.__setattr__(self, "store", store)

excludes_window

excludes_window(window: WindowContext) -> bool

Exclude every validation/evaluation chromosome before tuple emission.

Source code in geno_lewm/data/_membership_store_runtime.py

def excludes_window(self, window: WindowContext) -> bool:
    """Exclude every validation/evaluation chromosome before tuple emission."""
    if not isinstance(window, WindowContext):
        raise InputError("window must be a WindowContext")
    _chromosome, role = self._placed_role(window)
    return role != "train"

excludes_edit

excludes_edit(window: WindowContext, edit: RelEdit) -> bool

Exclude a held chromosome or a validation/evaluation variant lookup.

Source code in geno_lewm/data/_membership_store_runtime.py

def excludes_edit(self, window: WindowContext, edit: RelEdit) -> bool:
    """Exclude a held chromosome or a validation/evaluation variant lookup."""
    if not isinstance(window, WindowContext):
        raise InputError("window must be a WindowContext")
    if not isinstance(edit, RelEdit):
        raise InputError("edit must be a RelEdit")
    _chromosome, role = self._placed_role(window)
    return role != "train"

to_dict

to_dict() -> dict[str, object]

Return a compact policy binding, never the complete key set.

Source code in geno_lewm/data/_membership_store_runtime.py

def to_dict(self) -> dict[str, object]:
    """Return a compact policy binding, never the complete key set."""
    return {
        "schema_version": MEMBERSHIP_STORE_SCHEMA_VERSION,
        "membership_content_identity": self.store.manifest.content_identity,
        "excluded_chromosomes": list(self.holdout_chroms),
        "selection": "chromosome_roles",
        "lookup": _INDEX_NAME,
    }

identity

identity() -> str

Return the semantic identity of this indexed holdout policy.

Source code in geno_lewm/data/_membership_store_runtime.py

def identity(self) -> str:
    """Return the semantic identity of this indexed holdout policy."""
    return canonical_json_sha256(self.to_dict())

build_membership_store

build_membership_store(*, artifact_id: str, snapshot_lineage_path: Path, expected_snapshot_lineage_sha256: str, builder_git_commit: str, container_image: str, sources: Sequence[MembershipSourceInput], output_dir: Path) -> MembershipStoreManifest

Build one immutable store from expected lineage bytes and pinned builder provenance.

Source code in geno_lewm/data/membership_store.py

def build_membership_store(
    *,
    artifact_id: str,
    snapshot_lineage_path: Path,
    expected_snapshot_lineage_sha256: str,
    builder_git_commit: str,
    container_image: str,
    sources: Sequence[MembershipSourceInput],
    output_dir: Path,
) -> MembershipStoreManifest:
    """Build one immutable store from expected lineage bytes and pinned builder provenance."""
    return _build_membership_store(
        artifact_id=artifact_id,
        snapshot_lineage_path=snapshot_lineage_path,
        expected_snapshot_lineage_sha256=expected_snapshot_lineage_sha256,
        builder_git_commit=builder_git_commit,
        container_image=container_image,
        sources=sources,
        output_dir=output_dir,
    )

verify_membership_store

verify_membership_store(store_dir: Path) -> MembershipStoreVerification

Independently verify manifest, files, Parquet rows, and SQLite rows.

Source code in geno_lewm/data/membership_store.py

def verify_membership_store(store_dir: Path) -> MembershipStoreVerification:
    """Independently verify manifest, files, Parquet rows, and SQLite rows."""
    return _verify_membership_store(store_dir)