Source code for sc_tools.registry

"""Data registry for sc_tools projects.

Tracks data objects (AnnData checkpoints, images, etc.) and patients across
all active projects.  Uses **SQLite** by default (zero-config) with optional
**PostgreSQL** support via the ``SC_TOOLS_REGISTRY_URL`` environment variable.

Quick start
-----------
::

    from sc_tools.registry import Registry

    reg = Registry()
    reg.add_project("ggo_visium", platform="visium",
                    domain="spatial_transcriptomics")
    ds_id = reg.register_data(
        "ggo_visium", phase="qc_filter",
        uri="sftp://brb//athena/.../adata.raw.h5ad",
        fmt="h5ad",
        file_role="primary",
    )

CLI
---
::

    python -m sc_tools registry status

Environment
-----------
``SC_TOOLS_REGISTRY_URL``
    Optional SQLAlchemy-compatible DB URL.  Defaults to
    ``sqlite:///~/.sc_tools/registry.db``.

Phase slugs
-----------
    ingest_raw       p0a
    ingest_load      p0b
    qc_filter        p1
    metadata_attach  p2
    preprocess       p3
    demographics     p3.5
    scoring          p3.5b
    celltype_manual  p4
    biology          p5
    meta_analysis    p6/p7
"""

from __future__ import annotations

import logging
import os
import warnings
from datetime import datetime, timezone
from pathlib import Path
from types import SimpleNamespace
from typing import Any

logger = logging.getLogger(__name__)

_DEFAULT_DB_PATH = Path.home() / ".sc_tools" / "registry.db"


# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------


def _get_db_url() -> str:
    """Return the database URL from env or default SQLite path."""
    url = os.environ.get("SC_TOOLS_REGISTRY_URL")
    if url:
        return url
    db_path = _DEFAULT_DB_PATH
    db_path.parent.mkdir(parents=True, exist_ok=True)
    return f"sqlite:///{db_path}"


def _require_sqlalchemy() -> Any:
    try:
        import sqlalchemy

        return sqlalchemy
    except ImportError as e:
        raise ImportError(
            "SQLAlchemy is required for the registry. Install with: pip install sc-tools[registry]"
        ) from e


def _utcnow() -> str:
    return datetime.now(tz=timezone.utc).isoformat()  # noqa: UP017


_URI_SCHEMES = ("sftp://", "s3://", "gs://", "az://", "http://", "https://", "phase://")


def _validate_uri(uri: str) -> str:
    """Ensure URI is an absolute path or has a recognized scheme."""
    if any(uri.startswith(s) for s in _URI_SCHEMES):
        return uri
    if os.path.isabs(uri):
        return uri
    raise ValueError(
        f"URI must be an absolute path or use a scheme ({', '.join(_URI_SCHEMES)}). Got: {uri!r}"
    )


# ---------------------------------------------------------------------------
# ORM models -- built lazily to avoid top-level SQLAlchemy import errors
# ---------------------------------------------------------------------------


def _build_models(Base: Any) -> SimpleNamespace:
    """Return ORM model classes as a SimpleNamespace.

    Attributes
    ----------
    Project, DataSource, InventoryItem, Dataset, DatasetMember,
    ProjectDataset, ProjectPhase, Patient, Sample, Provenance
    """
    from sqlalchemy import (
        JSON as JSONType,
    )
    from sqlalchemy import (
        Boolean,
        Column,
        Float,
        ForeignKey,
        Integer,
        String,
        Text,
    )
    from sqlalchemy.orm import relationship

    # ------------------------------------------------------------------
    # Existing tables (unchanged)
    # ------------------------------------------------------------------

    class Project(Base):
        __tablename__ = "projects"

        id = Column(Integer, primary_key=True, autoincrement=True)
        name = Column(String, unique=True, nullable=False)
        status = Column(String, default="active")
        created_at = Column(String, default=_utcnow)

        phase_links = relationship(
            "ProjectPhase", back_populates="project", cascade="all, delete-orphan"
        )
        dataset_links = relationship(
            "ProjectDataset", back_populates="project", cascade="all, delete-orphan"
        )

    class Patient(Base):
        __tablename__ = "patients"

        id = Column(Integer, primary_key=True, autoincrement=True)
        patient_id = Column(String, unique=True, nullable=False)
        meta = Column("metadata", JSONType, default=dict)
        created_at = Column(String, default=_utcnow)

        sample_links = relationship(
            "Sample", back_populates="patient", cascade="all, delete-orphan"
        )

    class Sample(Base):
        """A biological specimen collected from a patient."""

        __tablename__ = "samples"

        id = Column(Integer, primary_key=True, autoincrement=True)
        patient_id = Column(Integer, ForeignKey("patients.id", ondelete="SET NULL"), nullable=True)
        sample_id = Column(String, unique=True, nullable=False)
        tissue = Column(String)
        collection_date = Column(String)
        meta = Column("metadata", JSONType, default=dict)
        created_at = Column(String, default=_utcnow)

        patient = relationship("Patient", back_populates="sample_links")
        inventory_items = relationship("InventoryItem", back_populates="sample")

    # ------------------------------------------------------------------
    # Layer 0: data_sources
    # ------------------------------------------------------------------

    class DataSource(Base):
        """A raw data source -- HPC directory, public dataset, or external repository."""

        __tablename__ = "data_sources"

        id = Column(Integer, primary_key=True, autoincrement=True)
        name = Column(String, unique=True, nullable=False)
        description = Column(Text)
        uri = Column(Text, nullable=False)
        platform = Column(String)
        domain = Column(String)
        imaging_modality = Column(String)
        source_type = Column(String)
        organism = Column(String)
        tissue = Column(String)
        disease = Column(String)
        n_samples = Column(Integer)
        n_cells = Column(Integer)
        publication = Column(String)
        access_notes = Column(Text)
        status = Column(String, default="discovered")
        meta = Column("metadata", JSONType, default=dict)
        created_at = Column(String, default=_utcnow)

        inventory_items = relationship("InventoryItem", back_populates="data_source")

    # ------------------------------------------------------------------
    # New Layer 1: inventory_items
    # ------------------------------------------------------------------

    class InventoryItem(Base):
        """A clean, ingested data file (h5ad) produced from a data source."""

        __tablename__ = "inventory_items"

        id = Column(Integer, primary_key=True, autoincrement=True)
        data_source_id = Column(
            Integer, ForeignKey("data_sources.id", ondelete="SET NULL"), nullable=True
        )
        sample_id = Column(Integer, ForeignKey("samples.id", ondelete="SET NULL"), nullable=True)
        name = Column(String, unique=True, nullable=False)
        uri = Column(Text, nullable=False)
        modality = Column(String, nullable=False)
        platform = Column(String)
        format = Column(String, default="h5ad")
        n_obs = Column(Integer)
        n_vars = Column(Integer)
        size_mb = Column(Float)
        organism = Column(String)
        tissue = Column(String)
        meta = Column("metadata", JSONType, default=dict)
        created_at = Column(String, default=_utcnow)
        updated_at = Column(String, default=_utcnow)

        data_source = relationship("DataSource", back_populates="inventory_items")
        sample = relationship("Sample", back_populates="inventory_items")
        dataset_memberships = relationship("DatasetMember", back_populates="inventory_item")

    # ------------------------------------------------------------------
    # New Layer 2: datasets + dataset_members
    # ------------------------------------------------------------------

    class Dataset(Base):
        """A versioned assembly of inventory items (MuData or AnnData)."""

        __tablename__ = "datasets"

        id = Column(Integer, primary_key=True, autoincrement=True)
        name = Column(String, nullable=False)
        version = Column(Integer, nullable=False, default=1)
        description = Column(Text)
        uri = Column(Text)
        format = Column(String, default="mudata")
        n_obs = Column(Integer)
        size_mb = Column(Float)
        is_current = Column(Boolean, default=True)
        meta = Column("metadata", JSONType, default=dict)
        created_at = Column(String, default=_utcnow)
        updated_at = Column(String, default=_utcnow)

        members = relationship(
            "DatasetMember", back_populates="dataset", cascade="all, delete-orphan"
        )
        project_links = relationship(
            "ProjectDataset", back_populates="dataset", cascade="all, delete-orphan"
        )
        phases = relationship("ProjectPhase", back_populates="dataset")

    class DatasetMember(Base):
        """Composition link: which inventory items belong to a dataset."""

        __tablename__ = "dataset_members"

        id = Column(Integer, primary_key=True, autoincrement=True)
        dataset_id = Column(Integer, ForeignKey("datasets.id", ondelete="CASCADE"), nullable=False)
        inventory_id = Column(
            Integer, ForeignKey("inventory_items.id", ondelete="RESTRICT"), nullable=False
        )
        modality_key = Column(String, nullable=False)

        dataset = relationship("Dataset", back_populates="members")
        inventory_item = relationship("InventoryItem", back_populates="dataset_memberships")

    # ------------------------------------------------------------------
    # New Layer 3: project_datasets + project_phases
    # ------------------------------------------------------------------

    class ProjectDataset(Base):
        """Many-to-many join between Projects and Datasets."""

        __tablename__ = "project_datasets"

        id = Column(Integer, primary_key=True, autoincrement=True)
        project_id = Column(Integer, ForeignKey("projects.id", ondelete="CASCADE"), nullable=False)
        dataset_id = Column(Integer, ForeignKey("datasets.id", ondelete="CASCADE"), nullable=False)
        role = Column(String, default="primary")
        notes = Column(Text)

        project = relationship("Project", back_populates="dataset_links")
        dataset = relationship("Dataset", back_populates="project_links")

    class ProjectPhase(Base):
        """Phase tracking per project + dataset combination."""

        __tablename__ = "project_phases"

        id = Column(Integer, primary_key=True, autoincrement=True)
        project_id = Column(Integer, ForeignKey("projects.id", ondelete="CASCADE"), nullable=False)
        dataset_id = Column(Integer, ForeignKey("datasets.id", ondelete="RESTRICT"), nullable=False)
        phase_group = Column(String, nullable=False)
        subphase = Column(String, nullable=False)
        status = Column(String, default="pending")
        uri = Column(Text)
        n_obs = Column(Integer)
        n_vars = Column(Integer)
        meta = Column("metadata", JSONType, default=dict)
        created_at = Column(String, default=_utcnow)
        updated_at = Column(String, default=_utcnow)

        project = relationship("Project", back_populates="phase_links")
        dataset = relationship("Dataset", back_populates="phases")

    # ------------------------------------------------------------------
    # Provenance
    # ------------------------------------------------------------------

    class Provenance(Base):
        """Tracks tool versions, params, and environment for transformations."""

        __tablename__ = "provenance"

        id = Column(Integer, primary_key=True, autoincrement=True)
        target_type = Column(String, nullable=False)
        phase_id = Column(
            Integer, ForeignKey("project_phases.id", ondelete="SET NULL"), nullable=True
        )
        dataset_id = Column(Integer, ForeignKey("datasets.id", ondelete="SET NULL"), nullable=True)
        tool = Column(String, nullable=False)
        tool_version = Column(String)
        reference_genome = Column(String)
        reference_dataset = Column(Text)
        signature_source = Column(Text)
        n_input_obs = Column(Integer)
        n_output_obs = Column(Integer)
        params = Column(JSONType, default=dict)
        environment = Column(JSONType, default=dict)
        script_uri = Column(Text)
        agent = Column(String)
        created_at = Column(String, default=_utcnow)

        phase = relationship("ProjectPhase")
        dataset = relationship("Dataset")

    return SimpleNamespace(
        Project=Project,
        DataSource=DataSource,
        InventoryItem=InventoryItem,
        Dataset=Dataset,
        DatasetMember=DatasetMember,
        ProjectDataset=ProjectDataset,
        ProjectPhase=ProjectPhase,
        Patient=Patient,
        Sample=Sample,
        Provenance=Provenance,
    )


# ---------------------------------------------------------------------------
# Registry class
# ---------------------------------------------------------------------------


[docs] class Registry: """Project and data registry backed by SQLite or PostgreSQL. Parameters ---------- db_url SQLAlchemy DB URL. Defaults to env var ``SC_TOOLS_REGISTRY_URL`` or ``sqlite:///~/.sc_tools/registry.db``. """ def __init__(self, db_url: str | None = None) -> None: sa = _require_sqlalchemy() from sqlalchemy.orm import DeclarativeBase class Base(DeclarativeBase): pass self._Base = Base models = _build_models(Base) self._Project = models.Project self._Patient = models.Patient self._Sample = models.Sample self._DataSource = models.DataSource self._InventoryItem = models.InventoryItem self._Dataset = models.Dataset self._DatasetMember = models.DatasetMember self._ProjectDataset = models.ProjectDataset self._ProjectPhase = models.ProjectPhase self._Provenance = models.Provenance url = db_url or _get_db_url() self._engine = sa.create_engine(url, echo=False) Base.metadata.create_all(self._engine) logger.info("Registry connected: %s", url) def _session(self) -> Any: from sqlalchemy.orm import Session return Session(self._engine) # ------------------------------------------------------------------ # Projects # ------------------------------------------------------------------
[docs] def add_project( self, name: str, platform: str | None = None, # ignored (column dropped) data_type: str | None = None, # ignored *, domain: str | None = None, # ignored (column dropped) imaging_modality: str | None = None, # ignored project_type: str = "internal", # ignored visibility: str = "private", # ignored ) -> int: """Register a new project. Returns project id. If a project with the same name exists its id is returned without creating a duplicate. """ with self._session() as sess: existing = sess.query(self._Project).filter_by(name=name).first() if existing: logger.info("Project '%s' already registered (id=%d)", name, existing.id) return existing.id proj = self._Project(name=name) sess.add(proj) sess.commit() sess.refresh(proj) logger.info("Registered project '%s' (id=%d)", name, proj.id) return proj.id
[docs] def get_project(self, name: str) -> dict[str, Any] | None: """Return project dict or None if not found.""" with self._session() as sess: row = sess.query(self._Project).filter_by(name=name).first() return self._to_dict(row) if row else None
[docs] def list_projects(self) -> list[dict[str, Any]]: """Return all registered projects.""" with self._session() as sess: return [self._to_dict(r) for r in sess.query(self._Project).all()]
[docs] def delete_project(self, name: str) -> bool: """Delete a project and all its cascade rows. Returns True if deleted.""" with self._session() as sess: proj = sess.query(self._Project).filter_by(name=name).first() if proj is None: return False sess.delete(proj) sess.commit() logger.info("Deleted project '%s'", name) return True
# ------------------------------------------------------------------ # Inventory Items (NEW -- Layer 1) # ------------------------------------------------------------------
[docs] def register_inventory_item( self, name: str, uri: str, modality: str, *, data_source_name: str | None = None, sample_name: str | None = None, platform: str | None = None, fmt: str = "h5ad", n_obs: int | None = None, n_vars: int | None = None, size_mb: float | None = None, organism: str | None = None, tissue: str | None = None, metadata: dict | None = None, ) -> int: """Register an inventory item. Returns item id. Idempotent by name.""" _validate_uri(uri) with self._session() as sess: existing = sess.query(self._InventoryItem).filter_by(name=name).first() if existing: logger.info("InventoryItem '%s' already registered (id=%d)", name, existing.id) return existing.id data_source_id = None if data_source_name is not None: ds = sess.query(self._DataSource).filter_by(name=data_source_name).first() if ds is None: raise ValueError(f"DataSource '{data_source_name}' not found.") data_source_id = ds.id sample_id = None if sample_name is not None: sample = sess.query(self._Sample).filter_by(sample_id=sample_name).first() if sample is None: raise ValueError(f"Sample '{sample_name}' not found.") sample_id = sample.id now = _utcnow() item = self._InventoryItem( name=name, uri=uri, modality=modality, data_source_id=data_source_id, sample_id=sample_id, platform=platform, format=fmt, n_obs=n_obs, n_vars=n_vars, size_mb=size_mb, organism=organism, tissue=tissue, meta=metadata or {}, created_at=now, updated_at=now, ) sess.add(item) sess.commit() sess.refresh(item) logger.info("Registered inventory item '%s' (id=%d)", name, item.id) return item.id
[docs] def get_inventory_item(self, name: str) -> dict[str, Any] | None: """Return inventory item dict by name, or None.""" with self._session() as sess: row = sess.query(self._InventoryItem).filter_by(name=name).first() return self._to_dict(row) if row else None
[docs] def list_inventory_items( self, modality: str | None = None, platform: str | None = None, ) -> list[dict[str, Any]]: """List inventory items with optional filters.""" with self._session() as sess: q = sess.query(self._InventoryItem) if modality is not None: q = q.filter_by(modality=modality) if platform is not None: q = q.filter_by(platform=platform) return [self._to_dict(r) for r in q.order_by(self._InventoryItem.name).all()]
# ------------------------------------------------------------------ # Datasets (NEW -- Layer 2) # ------------------------------------------------------------------
[docs] def create_dataset( self, name: str, *, description: str | None = None, fmt: str = "mudata", ) -> int: """Create a dataset. Returns dataset id. Idempotent for name+version=1.""" with self._session() as sess: existing = sess.query(self._Dataset).filter_by(name=name, version=1).first() if existing: logger.info("Dataset '%s' v1 already exists (id=%d)", name, existing.id) return existing.id now = _utcnow() ds = self._Dataset( name=name, version=1, description=description, format=fmt, is_current=True, created_at=now, updated_at=now, ) sess.add(ds) sess.commit() sess.refresh(ds) logger.info("Created dataset '%s' v1 (id=%d)", name, ds.id) return ds.id
[docs] def get_dataset( self, name: str, version: int | None = None, ) -> dict[str, Any] | None: """Return dataset dict. If version is None, get is_current or highest version.""" with self._session() as sess: if version is not None: row = sess.query(self._Dataset).filter_by(name=name, version=version).first() else: # Prefer is_current=True, fall back to highest version row = sess.query(self._Dataset).filter_by(name=name, is_current=True).first() if row is None: row = ( sess.query(self._Dataset) .filter_by(name=name) .order_by(self._Dataset.version.desc()) .first() ) return self._to_dict(row) if row else None
[docs] def add_dataset_member( self, dataset_name: str, inventory_name: str, modality_key: str, ) -> int: """Add an inventory item to a dataset. Returns member id.""" with self._session() as sess: ds = sess.query(self._Dataset).filter_by(name=dataset_name, is_current=True).first() if ds is None: raise ValueError(f"Dataset '{dataset_name}' not found (no current version).") inv = sess.query(self._InventoryItem).filter_by(name=inventory_name).first() if inv is None: raise ValueError(f"Inventory item '{inventory_name}' not found.") # Check for duplicate modality_key existing = ( sess.query(self._DatasetMember) .filter_by(dataset_id=ds.id, modality_key=modality_key) .first() ) if existing: raise ValueError( f"Modality key '{modality_key}' already exists in dataset " f"'{dataset_name}' v{ds.version}." ) member = self._DatasetMember( dataset_id=ds.id, inventory_id=inv.id, modality_key=modality_key, ) sess.add(member) sess.commit() sess.refresh(member) logger.info( "Added member '%s' (%s) to dataset '%s' v%d", inventory_name, modality_key, dataset_name, ds.version, ) return member.id
[docs] def remove_dataset_member( self, dataset_name: str, modality_key: str, ) -> None: """Remove a member from the current version of a dataset.""" with self._session() as sess: ds = sess.query(self._Dataset).filter_by(name=dataset_name, is_current=True).first() if ds is None: raise ValueError(f"Dataset '{dataset_name}' not found (no current version).") member = ( sess.query(self._DatasetMember) .filter_by(dataset_id=ds.id, modality_key=modality_key) .first() ) if member is None: raise ValueError( f"Modality key '{modality_key}' not found in dataset " f"'{dataset_name}' v{ds.version}." ) sess.delete(member) sess.commit() logger.info( "Removed member '%s' from dataset '%s' v%d", modality_key, dataset_name, ds.version, )
[docs] def get_dataset_members( self, dataset_name: str, version: int | None = None, ) -> list[dict[str, Any]]: """Return list of member dicts (including inventory item details).""" with self._session() as sess: if version is not None: ds = sess.query(self._Dataset).filter_by(name=dataset_name, version=version).first() else: ds = sess.query(self._Dataset).filter_by(name=dataset_name, is_current=True).first() if ds is None: ds = ( sess.query(self._Dataset) .filter_by(name=dataset_name) .order_by(self._Dataset.version.desc()) .first() ) if ds is None: return [] members = sess.query(self._DatasetMember).filter_by(dataset_id=ds.id).all() result = [] for m in members: inv = sess.get(self._InventoryItem, m.inventory_id) d = { "member_id": m.id, "dataset_id": m.dataset_id, "modality_key": m.modality_key, "inventory_id": m.inventory_id, } if inv: d["inventory_name"] = inv.name d["inventory_uri"] = inv.uri d["inventory_modality"] = inv.modality d["inventory_platform"] = inv.platform d["inventory_format"] = inv.format d["n_obs"] = inv.n_obs d["n_vars"] = inv.n_vars result.append(d) return result
[docs] def bump_dataset_version(self, dataset_name: str) -> int: """Create a new version of a dataset, copying all members. Returns new dataset id.""" with self._session() as sess: current = ( sess.query(self._Dataset).filter_by(name=dataset_name, is_current=True).first() ) if current is None: raise ValueError(f"Dataset '{dataset_name}' not found (no current version).") old_version = current.version current.is_current = False current.updated_at = _utcnow() now = _utcnow() new_ds = self._Dataset( name=dataset_name, version=old_version + 1, description=current.description, uri=current.uri, format=current.format, n_obs=current.n_obs, size_mb=current.size_mb, is_current=True, meta=dict(current.meta or {}), created_at=now, updated_at=now, ) sess.add(new_ds) sess.flush() # get new_ds.id # Copy members old_members = sess.query(self._DatasetMember).filter_by(dataset_id=current.id).all() for m in old_members: new_member = self._DatasetMember( dataset_id=new_ds.id, inventory_id=m.inventory_id, modality_key=m.modality_key, ) sess.add(new_member) sess.commit() sess.refresh(new_ds) logger.info( "Bumped dataset '%s' from v%d to v%d (id=%d)", dataset_name, old_version, new_ds.version, new_ds.id, ) return new_ds.id
# ------------------------------------------------------------------ # Project-Dataset links (NEW) # ------------------------------------------------------------------
[docs] def list_project_datasets(self, project_name: str) -> list[dict[str, Any]]: """Return all datasets linked to a project.""" with self._session() as sess: proj = sess.query(self._Project).filter_by(name=project_name).first() if proj is None: return [] links = sess.query(self._ProjectDataset).filter_by(project_id=proj.id).all() result = [] for lnk in links: ds = sess.get(self._Dataset, lnk.dataset_id) if ds: d = self._to_dict(ds) d["link_role"] = lnk.role d["link_notes"] = lnk.notes d["link_id"] = lnk.id result.append(d) return result
# ------------------------------------------------------------------ # Project Phases (REWRITTEN -- uses project_phases table) # ------------------------------------------------------------------
[docs] def upsert_phase( self, project_name: str, dataset_name: str, phase_group: str, subphase: str, *, status: str = "in_progress", uri: str | None = None, n_obs: int | None = None, n_vars: int | None = None, notes: str | None = None, metadata: dict | None = None, ) -> None: """Create or update a phase in project_phases. Parameters ---------- project_name: Project name (must exist). dataset_name: Dataset name (must exist, uses is_current version). phase_group: Phase group: ``'data_processing'`` or ``'discovery'``. subphase: Subphase slug (e.g. ``'qc_filter'``, ``'clustering_v1'``). status: Phase status string. uri: Optional checkpoint URI. n_obs: Number of observations. n_vars: Number of variables. notes: Free-text notes (stored in metadata). metadata: Additional metadata dict. """ with self._session() as sess: proj = sess.query(self._Project).filter_by(name=project_name).first() if proj is None: raise ValueError(f"Project '{project_name}' not found in registry") ds = sess.query(self._Dataset).filter_by(name=dataset_name, is_current=True).first() if ds is None: raise ValueError( f"Dataset '{dataset_name}' not found (no current version). " "Create the dataset and link it to the project first." ) existing = ( sess.query(self._ProjectPhase) .filter_by( project_id=proj.id, dataset_id=ds.id, phase_group=phase_group, subphase=subphase, ) .first() ) now = _utcnow() if existing is not None: existing.status = status existing.updated_at = now if uri is not None: existing.uri = uri if n_obs is not None: existing.n_obs = n_obs if n_vars is not None: existing.n_vars = n_vars meta = dict(existing.meta or {}) if notes is not None: meta["notes"] = notes if metadata: meta.update(metadata) existing.meta = meta else: meta = dict(metadata or {}) if notes is not None: meta["notes"] = notes row = self._ProjectPhase( project_id=proj.id, dataset_id=ds.id, phase_group=phase_group, subphase=subphase, status=status, uri=uri, n_obs=n_obs, n_vars=n_vars, meta=meta, created_at=now, updated_at=now, ) sess.add(row) sess.commit()
[docs] def get_phase( self, project_name: str, phase_group: str, subphase: str, ) -> dict[str, Any] | None: """Return phase info from project_phases.""" with self._session() as sess: proj = sess.query(self._Project).filter_by(name=project_name).first() if proj is None: return None row = ( sess.query(self._ProjectPhase) .filter_by( project_id=proj.id, phase_group=phase_group, subphase=subphase, ) .first() ) if row is None: return None meta = row.meta or {} return { "phase_group": row.phase_group, "subphase": row.subphase, "status": row.status, "uri": row.uri, "n_obs": row.n_obs, "n_vars": row.n_vars, "dataset_id": row.dataset_id, "notes": meta.get("notes"), "updated_at": row.updated_at, }
[docs] def list_phases( self, project_name: str, phase_group: str | None = None, ) -> list[dict[str, Any]]: """Return phase summaries for a project from project_phases.""" with self._session() as sess: proj = sess.query(self._Project).filter_by(name=project_name).first() if proj is None: return [] q = sess.query(self._ProjectPhase).filter_by(project_id=proj.id) if phase_group is not None: q = q.filter_by(phase_group=phase_group) rows = q.order_by( self._ProjectPhase.phase_group, self._ProjectPhase.subphase, ).all() result = [] for r in rows: rmeta = r.meta or {} result.append( { "phase_group": r.phase_group, "subphase": r.subphase, "status": r.status, "uri": r.uri, "n_obs": r.n_obs, "n_vars": r.n_vars, "dataset_id": r.dataset_id, "notes": rmeta.get("notes"), "updated_at": r.updated_at, } ) return result
[docs] def mark_phase_complete( self, project_name: str, phase_group: str, subphase: str, ) -> None: """Mark a phase as 'ready' in project_phases.""" with self._session() as sess: proj = sess.query(self._Project).filter_by(name=project_name).first() if proj is None: raise ValueError(f"Project '{project_name}' not found in registry") rows = ( sess.query(self._ProjectPhase) .filter_by( project_id=proj.id, phase_group=phase_group, subphase=subphase, ) .all() ) if not rows: raise ValueError( f"Phase '{phase_group}/{subphase}' not found for project '{project_name}'" ) now = _utcnow() for r in rows: r.status = "ready" r.updated_at = now sess.commit()
# ------------------------------------------------------------------ # Provenance (NEW) # ------------------------------------------------------------------
[docs] def record_provenance( self, tool: str, *, tool_version: str | None = None, reference_genome: str | None = None, reference_dataset: str | None = None, signature_source: str | None = None, n_input_obs: int | None = None, n_output_obs: int | None = None, params: dict | None = None, environment: dict | None = None, script_uri: str | None = None, agent: str | None = None, phase_id: int | None = None, dataset_id: int | None = None, ) -> int: """Record provenance for a transformation. Returns provenance id. Exactly one of phase_id or dataset_id must be provided. """ # Validate exactly one FK fk_count = sum(x is not None for x in (phase_id, dataset_id)) if fk_count != 1: raise ValueError( f"Exactly one of phase_id or dataset_id must be provided. Got {fk_count}." ) if phase_id is not None: target_type = "phase" else: target_type = "dataset" with self._session() as sess: row = self._Provenance( target_type=target_type, phase_id=phase_id, dataset_id=dataset_id, tool=tool, tool_version=tool_version, reference_genome=reference_genome, reference_dataset=reference_dataset, signature_source=signature_source, n_input_obs=n_input_obs, n_output_obs=n_output_obs, params=params or {}, environment=environment or {}, script_uri=script_uri, agent=agent, created_at=_utcnow(), ) sess.add(row) sess.commit() sess.refresh(row) logger.info( "Recorded provenance id=%d (tool=%s, target_type=%s)", row.id, tool, target_type, ) return row.id
[docs] def get_provenance( self, *, phase_id: int | None = None, dataset_id: int | None = None, ) -> list[dict[str, Any]]: """Return provenance records for a given target.""" with self._session() as sess: q = sess.query(self._Provenance) if phase_id is not None: q = q.filter_by(target_type="phase", phase_id=phase_id) elif dataset_id is not None: q = q.filter_by(target_type="dataset", dataset_id=dataset_id) else: # No filter -- return all pass return [self._to_dict(r) for r in q.order_by(self._Provenance.id).all()]
# ------------------------------------------------------------------ # Data sources (uses new data_sources table) # ------------------------------------------------------------------
[docs] def register_data_source( self, name: str, uri: str, *, description: str | None = None, platform: str | None = None, domain: str | None = None, imaging_modality: str | None = None, source_type: str | None = None, organism: str | None = None, tissue: str | None = None, disease: str | None = None, n_samples: int | None = None, n_cells: int | None = None, publication: str | None = None, access_notes: str | None = None, status: str = "available", metadata: dict | None = None, ) -> int: """Register a data source. Returns data source id. Idempotent.""" _validate_uri(uri) with self._session() as sess: existing = sess.query(self._DataSource).filter_by(name=name).first() if existing: logger.info("DataSource '%s' already registered (id=%d)", name, existing.id) return existing.id src = self._DataSource( name=name, uri=uri, description=description, platform=platform, domain=domain, imaging_modality=imaging_modality, source_type=source_type, organism=organism, tissue=tissue, disease=disease, n_samples=n_samples, n_cells=n_cells, publication=publication, access_notes=access_notes, status=status, meta=metadata or {}, ) sess.add(src) sess.commit() sess.refresh(src) logger.info("Registered data source '%s' (id=%d)", name, src.id) return src.id
[docs] def get_data_source(self, name: str) -> dict[str, Any] | None: """Return data source dict or None if not found.""" with self._session() as sess: row = sess.query(self._DataSource).filter_by(name=name).first() return self._to_dict(row) if row else None
[docs] def list_data_sources( self, platform: str | None = None, source_type: str | None = None, disease: str | None = None, tissue: str | None = None, ) -> list[dict[str, Any]]: """Return data sources, optionally filtered.""" with self._session() as sess: q = sess.query(self._DataSource) if platform is not None: q = q.filter_by(platform=platform) if source_type is not None: q = q.filter_by(source_type=source_type) if disease is not None: q = q.filter(self._DataSource.disease.ilike(f"%{disease}%")) if tissue is not None: q = q.filter(self._DataSource.tissue.ilike(f"%{tissue}%")) return [self._to_dict(r) for r in q.order_by(self._DataSource.name).all()]
# ------------------------------------------------------------------ # Datasets query (REWRITTEN) # ------------------------------------------------------------------
[docs] def list_datasets( self, project_name: str | None = None, ) -> list[dict[str, Any]]: """List datasets, optionally filtered by project via project_datasets.""" with self._session() as sess: if project_name is not None: proj = sess.query(self._Project).filter_by(name=project_name).first() if proj is None: return [] links = sess.query(self._ProjectDataset).filter_by(project_id=proj.id).all() dataset_ids = [lnk.dataset_id for lnk in links] if not dataset_ids: return [] rows = ( sess.query(self._Dataset) .filter(self._Dataset.id.in_(dataset_ids)) .order_by(self._Dataset.name, self._Dataset.version) .all() ) else: rows = ( sess.query(self._Dataset) .order_by(self._Dataset.name, self._Dataset.version) .all() ) return [self._to_dict(r) for r in rows]
[docs] def get_dataset_uri( self, project_name: str, dataset_name: str, ) -> str | None: """Return the URI for a dataset linked to a project.""" with self._session() as sess: proj = sess.query(self._Project).filter_by(name=project_name).first() if proj is None: return None # Find the current version of the dataset ds = sess.query(self._Dataset).filter_by(name=dataset_name, is_current=True).first() if ds is None: return None # Verify it is linked to the project link = ( sess.query(self._ProjectDataset) .filter_by(project_id=proj.id, dataset_id=ds.id) .first() ) if link is None: return None return ds.uri
[docs] def project_data_summary(self, project_name: str) -> dict[str, Any]: """Return summary of datasets and their members for a project.""" with self._session() as sess: proj = sess.query(self._Project).filter_by(name=project_name).first() if proj is None: return { "project": project_name, "n_datasets": 0, "n_inventory_items": 0, "datasets": [], } links = sess.query(self._ProjectDataset).filter_by(project_id=proj.id).all() datasets_info = [] all_inventory_ids: set[int] = set() for lnk in links: ds = sess.get(self._Dataset, lnk.dataset_id) if ds is None: continue members = sess.query(self._DatasetMember).filter_by(dataset_id=ds.id).all() member_info = [] for m in members: all_inventory_ids.add(m.inventory_id) inv = sess.get(self._InventoryItem, m.inventory_id) member_info.append( { "modality_key": m.modality_key, "inventory_name": inv.name if inv else None, "modality": inv.modality if inv else None, "n_obs": inv.n_obs if inv else None, } ) datasets_info.append( { "name": ds.name, "version": ds.version, "is_current": ds.is_current, "format": ds.format, "role": lnk.role, "n_members": len(members), "members": member_info, } ) return { "project": project_name, "n_datasets": len(datasets_info), "n_inventory_items": len(all_inventory_ids), "datasets": datasets_info, }
# ------------------------------------------------------------------ # Patients (UNCHANGED) # ------------------------------------------------------------------
[docs] def add_patient( self, patient_id: str, metadata: dict | None = None, **clinical_kwargs: Any, ) -> int: """Register a patient. Returns patient DB id. Idempotent: if a patient with the same ``patient_id`` exists, its DB id is returned without creating a duplicate. Parameters ---------- patient_id: Unique de-identified identifier (e.g. ``"PT001"``). metadata: JSONB metadata dict with clinical info (organism, sex, diagnosis, etc.). **clinical_kwargs: Individual clinical fields merged into metadata dict. """ combined_meta = dict(metadata or {}) combined_meta.update(clinical_kwargs) with self._session() as sess: existing = sess.query(self._Patient).filter_by(patient_id=patient_id).first() if existing: logger.info("Patient '%s' already registered (id=%d)", patient_id, existing.id) return existing.id pat = self._Patient( patient_id=patient_id, meta=combined_meta, ) sess.add(pat) sess.commit() sess.refresh(pat) logger.info("Registered patient '%s' (id=%d)", patient_id, pat.id) return pat.id
[docs] def add_subject( self, subject_id: str, organism: str = "human", **clinical_kwargs: Any, ) -> int: """Backward-compatible alias for :meth:`add_patient`. Maps subject_id -> patient_id and packs clinical columns into metadata. """ meta = {"organism": organism} meta.update(clinical_kwargs) return self.add_patient(subject_id, metadata=meta)
[docs] def get_patient(self, patient_id: str) -> dict[str, Any] | None: """Return patient dict by de-identified ID, or None.""" with self._session() as sess: row = sess.query(self._Patient).filter_by(patient_id=patient_id).first() return self._to_dict(row) if row else None
# Backward compat alias
[docs] def get_subject(self, subject_id: str) -> dict[str, Any] | None: """Backward-compatible alias for :meth:`get_patient`.""" return self.get_patient(subject_id)
[docs] def list_patients( self, diagnosis: str | None = None, tissue: str | None = None, ) -> list[dict[str, Any]]: """Return patients, optionally filtered by metadata fields.""" with self._session() as sess: q = sess.query(self._Patient) # Filter using JSONB operators on PostgreSQL if diagnosis is not None: q = q.filter(self._Patient.meta["diagnosis"].astext.ilike(f"%{diagnosis}%")) if tissue is not None: q = q.filter(self._Patient.meta["tissue_of_origin"].astext.ilike(f"%{tissue}%")) return [self._to_dict(r) for r in q.order_by(self._Patient.patient_id).all()]
[docs] def list_subjects( self, project_name: str | None = None, diagnosis: str | None = None, tissue: str | None = None, ) -> list[dict[str, Any]]: """Backward-compatible alias for :meth:`list_patients`. The project_name filter checks for patients linked via patient -> sample -> inventory_item -> dataset_member -> dataset -> project. """ with self._session() as sess: q = sess.query(self._Patient) if project_name is not None: proj = sess.query(self._Project).filter_by(name=project_name).first() if proj is None: return [] # patient -> sample -> inventory_item -> dataset_member -> project_dataset patient_ids_subq = ( sess.query(self._Sample.patient_id) .join(self._InventoryItem, self._InventoryItem.sample_id == self._Sample.id) .join( self._DatasetMember, self._DatasetMember.inventory_id == self._InventoryItem.id, ) .join(self._Dataset, self._DatasetMember.dataset_id == self._Dataset.id) .join(self._ProjectDataset, self._ProjectDataset.dataset_id == self._Dataset.id) .filter(self._ProjectDataset.project_id == proj.id) .filter(self._Sample.patient_id.isnot(None)) .distinct() .scalar_subquery() ) q = q.filter(self._Patient.id.in_(patient_ids_subq)) if diagnosis is not None: q = q.filter(self._Patient.meta["diagnosis"].astext.ilike(f"%{diagnosis}%")) if tissue is not None: q = q.filter(self._Patient.meta["tissue_of_origin"].astext.ilike(f"%{tissue}%")) results = [] for r in q.order_by(self._Patient.patient_id).all(): d = self._to_dict(r) # Flatten metadata for backward compat meta = d.get("meta") or {} d["subject_id"] = d["patient_id"] d["organism"] = meta.get("organism") d["sex"] = meta.get("sex") d["diagnosis"] = meta.get("diagnosis") d["tissue_of_origin"] = meta.get("tissue_of_origin") results.append(d) return results
# ------------------------------------------------------------------ # Samples # ------------------------------------------------------------------
[docs] def register_sample( self, sample_id: str, *, patient_id: str | None = None, tissue: str | None = None, collection_date: str | None = None, metadata: dict | None = None, ) -> int: """Register a sample. Returns sample DB id. Idempotent by sample_id.""" with self._session() as sess: existing = sess.query(self._Sample).filter_by(sample_id=sample_id).first() if existing: logger.info("Sample '%s' already registered (id=%d)", sample_id, existing.id) return existing.id pat_db_id = None if patient_id is not None: pat = sess.query(self._Patient).filter_by(patient_id=patient_id).first() if pat is None: raise ValueError(f"Patient '{patient_id}' not found.") pat_db_id = pat.id sample = self._Sample( sample_id=sample_id, patient_id=pat_db_id, tissue=tissue, collection_date=collection_date, meta=metadata or {}, ) sess.add(sample) sess.commit() sess.refresh(sample) logger.info("Registered sample '%s' (id=%d)", sample_id, sample.id) return sample.id
[docs] def get_sample(self, sample_id: str) -> dict[str, Any] | None: """Return sample dict by sample_id, or None.""" with self._session() as sess: row = sess.query(self._Sample).filter_by(sample_id=sample_id).first() return self._to_dict(row) if row else None
[docs] def list_samples( self, patient_id: str | None = None, tissue: str | None = None, ) -> list[dict[str, Any]]: """List samples with optional filters.""" with self._session() as sess: q = sess.query(self._Sample) if patient_id is not None: pat = sess.query(self._Patient).filter_by(patient_id=patient_id).first() if pat is None: return [] q = q.filter_by(patient_id=pat.id) if tissue is not None: q = q.filter(self._Sample.tissue.ilike(f"%{tissue}%")) return [self._to_dict(r) for r in q.order_by(self._Sample.sample_id).all()]
# ------------------------------------------------------------------ # Status summary (REWRITTEN) # ------------------------------------------------------------------
[docs] def status(self) -> dict[str, Any]: """Return a high-level summary of current registry state.""" with self._session() as sess: n_projects = sess.query(self._Project).count() n_data_sources = sess.query(self._DataSource).count() n_inventory_items = sess.query(self._InventoryItem).count() n_datasets = sess.query(self._Dataset).filter_by(is_current=True).count() n_patients = sess.query(self._Patient).count() n_samples = sess.query(self._Sample).count() projects = [p.name for p in sess.query(self._Project).filter_by(status="active").all()] # Per-project phase summary from project_phases from sqlalchemy import func phase_summary: dict[str, dict[str, int]] = {} for proj_name in projects: proj = sess.query(self._Project).filter_by(name=proj_name).first() if proj is None: continue phase_counts = ( sess.query(self._ProjectPhase.status, func.count()) .filter_by(project_id=proj.id) .group_by(self._ProjectPhase.status) .all() ) counts = dict(phase_counts) if counts: phase_summary[proj_name] = counts return { "n_projects": n_projects, "n_data_sources": n_data_sources, "n_inventory_items": n_inventory_items, "n_datasets": n_datasets, "n_patients": n_patients, "n_samples": n_samples, # Backward-compat keys "n_subjects": n_patients, "n_data": n_inventory_items, "n_biodata": n_inventory_items, "active_slurm_jobs": 0, "running_agent_tasks": 0, "active_projects": projects, "phase_summary": phase_summary, }
# ------------------------------------------------------------------ # Backward compatibility wrappers (DEPRECATED) # ------------------------------------------------------------------
[docs] def register_data( self, project_name: str, phase: str, uri: str, *, fmt: str | None = "h5ad", platform: str | None = None, category: str | None = None, status: str = "ready", file_role: str = "primary", n_obs: int | None = None, n_vars: int | None = None, size_mb: float | None = None, patient_id: str | None = None, metadata: dict | None = None, sample_id: str | None = None, # backward compat, ignored ) -> int: """Deprecated. Use register_inventory_item() instead.""" warnings.warn( "register_data() is deprecated. Use register_inventory_item() instead.", DeprecationWarning, stacklevel=2, ) name = f"{project_name}_{phase}_{_utcnow().replace(':', '-')}" return self.register_inventory_item( name=name, uri=uri, modality=category or "unknown", platform=platform, fmt=fmt or "h5ad", n_obs=n_obs, n_vars=n_vars, size_mb=size_mb, metadata=metadata, )
[docs] def register_dataset( self, project_name: str, phase: str, uri: str, *, sample_id: str | None = None, fmt: str = "h5ad", size_mb: float | None = None, md5: str | None = None, status: str = "ready", file_role: str = "primary", validated: bool = False, n_obs: int | None = None, n_vars: int | None = None, ) -> int: """Deprecated. Use register_inventory_item() instead.""" warnings.warn( "register_dataset() is deprecated. Use register_inventory_item() instead.", DeprecationWarning, stacklevel=2, ) meta: dict[str, Any] = {} if md5: meta["md5"] = md5 if validated: meta["validated"] = True name = f"{project_name}_{phase}_{_utcnow().replace(':', '-')}" return self.register_inventory_item( name=name, uri=uri, modality="unknown", fmt=fmt, n_obs=n_obs, n_vars=n_vars, size_mb=size_mb, metadata=meta if meta else None, )
[docs] def register_biodata( self, project_name: str, category: str, platform: str, uri: str, *, fmt: str | None = None, status: str = "pending", file_role: str = "primary", phase: str | None = None, n_obs: int | None = None, n_vars: int | None = None, size_mb: float | None = None, md5: str | None = None, subject_id: str | None = None, sample_db_id: int | None = None, **type_kwargs: Any, ) -> int: """Deprecated. Use register_inventory_item() instead.""" warnings.warn( "register_biodata() is deprecated. Use register_inventory_item() instead.", DeprecationWarning, stacklevel=2, ) meta: dict[str, Any] = {} if md5: meta["md5"] = md5 if type_kwargs: meta.update(type_kwargs) name = f"{project_name}_{category}_{platform}_{_utcnow().replace(':', '-')}" return self.register_inventory_item( name=name, uri=uri, modality=category, platform=platform, fmt=fmt or "h5ad", n_obs=n_obs, n_vars=n_vars, size_mb=size_mb, metadata=meta if meta else None, )
[docs] def list_biodata( self, project_name: str | None = None, category: str | None = None, platform: str | None = None, sample_db_id: int | None = None, phase: str | None = None, ) -> list[dict[str, Any]]: """Deprecated. Use list_inventory_items() instead.""" warnings.warn( "list_biodata() is deprecated. Use list_inventory_items() instead.", DeprecationWarning, stacklevel=2, ) return self.list_inventory_items( modality=category, platform=platform, )
[docs] def list_project_data_sources(self, project_name: str) -> list[dict[str, Any]]: """Deprecated. Use list_project_datasets() instead.""" warnings.warn( "list_project_data_sources() is deprecated. Use list_project_datasets() instead.", DeprecationWarning, stacklevel=2, ) return self.list_project_datasets(project_name)
# ------------------------------------------------------------------ # Legacy methods kept for burn-in period # ------------------------------------------------------------------
[docs] def get_biodata(self, biodata_id: int) -> dict[str, Any] | None: """Return an inventory item dict by id.""" with self._session() as sess: row = sess.get(self._InventoryItem, biodata_id) if row is None: return None return self._to_dict(row)
[docs] def update_dataset_status(self, dataset_id: int, status: str) -> None: """Update dataset status.""" with self._session() as sess: row = sess.get(self._Dataset, dataset_id) if row is None: raise ValueError(f"Dataset id={dataset_id} not found") row.updated_at = _utcnow() sess.commit()
# ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ @staticmethod def _to_dict(row: Any) -> dict[str, Any]: """Convert an ORM row to a dict.""" from sqlalchemy import inspect as sa_inspect mapper = sa_inspect(type(row)) return {attr.key: getattr(row, attr.key) for attr in mapper.column_attrs}
# --------------------------------------------------------------------------- # CLI helpers # --------------------------------------------------------------------------- def _cli_status() -> None: """Print a registry status summary to stdout.""" reg = Registry() s = reg.status() print("\nsc_tools Registry Status") print("=" * 40) print(f" Projects : {s['n_projects']}") print(f" Data sources : {s['n_data_sources']}") print(f" Inventory items : {s['n_inventory_items']}") print(f" Datasets : {s['n_datasets']}") print(f" Patients : {s['n_patients']}") print(f" Samples : {s['n_samples']}") if s["active_projects"]: print("\n Active projects:") for proj_name in s["active_projects"]: print(f" - {proj_name}") phase_summary = s.get("phase_summary", {}).get(proj_name, {}) if phase_summary: parts = ", ".join(f"{k}={v}" for k, v in sorted(phase_summary.items())) print(f" phase status: {parts}") datasets = reg.list_project_datasets(project_name=proj_name) for ds in datasets[-5:]: role = ds.get("link_role", "primary") print( f" dataset={ds['name']} v{ds['version']} role={role} " f"format={ds['format']}" ) print() __all__ = ["Registry"]