sc_tools.registry — Project Registry#

SQLAlchemy-based registry for tracking projects, datasets, SLURM jobs, agent tasks, and pipeline phase status. SQLite by default; PostgreSQL via SC_TOOLS_REGISTRY_URL environment variable.

from sc_tools.registry import Registry

reg = Registry()
reg.add_project("ggo_visium", platform="visium")
reg.register_dataset("ggo_visium", "results/adata.raw.h5ad", phase="qc_filter")
reg.mark_phase_complete("ggo_visium", "qc_filter")
print(reg.status())

Install: pip install sc-tools[registry]

CLI: python -m sc_tools registry status

class sc_tools.registry.Registry(db_url=None)[source]#

Bases: object

Project and data registry backed by SQLite or PostgreSQL.

Parameters:

db_url (str | None) – SQLAlchemy DB URL. Defaults to env var SC_TOOLS_REGISTRY_URL or sqlite:///~/.sc_tools/registry.db.

add_project(name, platform=None, data_type=None, *, domain=None, imaging_modality=None, project_type='internal', visibility='private')[source]#

Register a new project. Returns project id.

If a project with the same name exists its id is returned without creating a duplicate.

Return type:

int

Parameters:

  • name (str)

  • platform (str | None)

  • data_type (str | None)

  • domain (str | None)

  • imaging_modality (str | None)

  • project_type (str)

  • visibility (str)

get_project(name)[source]#

Return project dict or None if not found.

Return type:

dict[str, Any] | None

Parameters:

name (str)

list_projects()[source]#

Return all registered projects.

Return type:

list[dict[str, Any]]

delete_project(name)[source]#

Delete a project and all its cascade rows. Returns True if deleted.

Return type:

bool

Parameters:

name (str)

register_inventory_item(name, uri, modality, *, data_source_name=None, sample_name=None, platform=None, fmt='h5ad', n_obs=None, n_vars=None, size_mb=None, organism=None, tissue=None, metadata=None)[source]#

Register an inventory item. Returns item id. Idempotent by name.

Return type:

int

Parameters:

  • name (str)

  • uri (str)

  • modality (str)

  • data_source_name (str | None)

  • sample_name (str | None)

  • platform (str | None)

  • fmt (str)

  • n_obs (int | None)

  • n_vars (int | None)

  • size_mb (float | None)

  • organism (str | None)

  • tissue (str | None)

  • metadata (dict | None)

get_inventory_item(name)[source]#

Return inventory item dict by name, or None.

Return type:

dict[str, Any] | None

Parameters:

name (str)

list_inventory_items(modality=None, platform=None)[source]#

List inventory items with optional filters.

Return type:

list[dict[str, Any]]

Parameters:

  • modality (str | None)

  • platform (str | None)

create_dataset(name, *, description=None, fmt='mudata')[source]#

Create a dataset. Returns dataset id. Idempotent for name+version=1.

Return type:

int

Parameters:

  • name (str)

  • description (str | None)

  • fmt (str)

get_dataset(name, version=None)[source]#

Return dataset dict. If version is None, get is_current or highest version.

Return type:

dict[str, Any] | None

Parameters:

  • name (str)

  • version (int | None)

add_dataset_member(dataset_name, inventory_name, modality_key)[source]#

Add an inventory item to a dataset. Returns member id.

Return type:

int

Parameters:

  • dataset_name (str)

  • inventory_name (str)

  • modality_key (str)

remove_dataset_member(dataset_name, modality_key)[source]#

Remove a member from the current version of a dataset.

Return type:

None

Parameters:

  • dataset_name (str)

  • modality_key (str)

get_dataset_members(dataset_name, version=None)[source]#

Return list of member dicts (including inventory item details).

Return type:

list[dict[str, Any]]

Parameters:

  • dataset_name (str)

  • version (int | None)

bump_dataset_version(dataset_name)[source]#

Create a new version of a dataset, copying all members. Returns new dataset id.

Return type:

int

Parameters:

dataset_name (str)

Link a project to a dataset. Returns link id. Idempotent.

Return type:

int

Parameters:

  • project_name (str)

  • dataset_name (str)

  • role (str)

  • notes (str | None)

list_project_datasets(project_name)[source]#

Return all datasets linked to a project.

Return type:

list[dict[str, Any]]

Parameters:

project_name (str)

upsert_phase(project_name, dataset_name, phase_group, subphase, *, status='in_progress', uri=None, n_obs=None, n_vars=None, notes=None, metadata=None)[source]#

Create or update a phase in project_phases.

Parameters:

  • project_name (str) – Project name (must exist).

  • dataset_name (str) – Dataset name (must exist, uses is_current version).

  • phase_group (str) – Phase group: 'data_processing' or 'discovery'.

  • subphase (str) – Subphase slug (e.g. 'qc_filter', 'clustering_v1').

  • status (str) – Phase status string.

  • uri (str | None) – Optional checkpoint URI.

  • n_obs (int | None) – Number of observations.

  • n_vars (int | None) – Number of variables.

  • notes (str | None) – Free-text notes (stored in metadata).

  • metadata (dict | None) – Additional metadata dict.

Return type:

None

get_phase(project_name, phase_group, subphase)[source]#

Return phase info from project_phases.

Return type:

dict[str, Any] | None

Parameters:

  • project_name (str)

  • phase_group (str)

  • subphase (str)

list_phases(project_name, phase_group=None)[source]#

Return phase summaries for a project from project_phases.

Return type:

list[dict[str, Any]]

Parameters:

  • project_name (str)

  • phase_group (str | None)

mark_phase_complete(project_name, phase_group, subphase)[source]#

Mark a phase as ‘ready’ in project_phases.

Return type:

None

Parameters:

  • project_name (str)

  • phase_group (str)

  • subphase (str)

record_provenance(tool, *, tool_version=None, reference_genome=None, reference_dataset=None, signature_source=None, n_input_obs=None, n_output_obs=None, params=None, environment=None, script_uri=None, agent=None, phase_id=None, dataset_id=None)[source]#

Record provenance for a transformation. Returns provenance id.

Exactly one of phase_id or dataset_id must be provided.

Return type:

int

Parameters:

  • tool (str)

  • tool_version (str | None)

  • reference_genome (str | None)

  • reference_dataset (str | None)

  • signature_source (str | None)

  • n_input_obs (int | None)

  • n_output_obs (int | None)

  • params (dict | None)

  • environment (dict | None)

  • script_uri (str | None)

  • agent (str | None)

  • phase_id (int | None)

  • dataset_id (int | None)

get_provenance(*, phase_id=None, dataset_id=None)[source]#

Return provenance records for a given target.

Return type:

list[dict[str, Any]]

Parameters:

  • phase_id (int | None)

  • dataset_id (int | None)

register_data_source(name, uri, *, description=None, platform=None, domain=None, imaging_modality=None, source_type=None, organism=None, tissue=None, disease=None, n_samples=None, n_cells=None, publication=None, access_notes=None, status='available', metadata=None)[source]#

Register a data source. Returns data source id. Idempotent.

Return type:

int

Parameters:

  • name (str)

  • uri (str)

  • description (str | None)

  • platform (str | None)

  • domain (str | None)

  • imaging_modality (str | None)

  • source_type (str | None)

  • organism (str | None)

  • tissue (str | None)

  • disease (str | None)

  • n_samples (int | None)

  • n_cells (int | None)

  • publication (str | None)

  • access_notes (str | None)

  • status (str)

  • metadata (dict | None)

get_data_source(name)[source]#

Return data source dict or None if not found.

Return type:

dict[str, Any] | None

Parameters:

name (str)

list_data_sources(platform=None, source_type=None, disease=None, tissue=None)[source]#

Return data sources, optionally filtered.

Return type:

list[dict[str, Any]]

Parameters:

  • platform (str | None)

  • source_type (str | None)

  • disease (str | None)

  • tissue (str | None)

list_datasets(project_name=None)[source]#

List datasets, optionally filtered by project via project_datasets.

Return type:

list[dict[str, Any]]

Parameters:

project_name (str | None)

get_dataset_uri(project_name, dataset_name)[source]#

Return the URI for a dataset linked to a project.

Return type:

str | None

Parameters:

  • project_name (str)

  • dataset_name (str)

project_data_summary(project_name)[source]#

Return summary of datasets and their members for a project.

Return type:

dict[str, Any]

Parameters:

project_name (str)

add_patient(patient_id, metadata=None, **clinical_kwargs)[source]#

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 (str) – Unique de-identified identifier (e.g. "PT001").

  • metadata (dict | None) – JSONB metadata dict with clinical info (organism, sex, diagnosis, etc.).

  • **clinical_kwargs (Any) – Individual clinical fields merged into metadata dict.

Return type:

int

add_subject(subject_id, organism='human', **clinical_kwargs)[source]#

Backward-compatible alias for add_patient().

Maps subject_id -> patient_id and packs clinical columns into metadata.

Return type:

int

Parameters:

  • subject_id (str)

  • organism (str)

  • clinical_kwargs (Any)

get_patient(patient_id)[source]#

Return patient dict by de-identified ID, or None.

Return type:

dict[str, Any] | None

Parameters:

patient_id (str)

get_subject(subject_id)[source]#

Backward-compatible alias for get_patient().

Return type:

dict[str, Any] | None

Parameters:

subject_id (str)

list_patients(diagnosis=None, tissue=None)[source]#

Return patients, optionally filtered by metadata fields.

Return type:

list[dict[str, Any]]

Parameters:

  • diagnosis (str | None)

  • tissue (str | None)

list_subjects(project_name=None, diagnosis=None, tissue=None)[source]#

Backward-compatible alias for list_patients().

The project_name filter checks for patients linked via patient -> sample -> inventory_item -> dataset_member -> dataset -> project.

Return type:

list[dict[str, Any]]

Parameters:

  • project_name (str | None)

  • diagnosis (str | None)

  • tissue (str | None)

register_sample(sample_id, *, patient_id=None, tissue=None, collection_date=None, metadata=None)[source]#

Register a sample. Returns sample DB id. Idempotent by sample_id.

Return type:

int

Parameters:

  • sample_id (str)

  • patient_id (str | None)

  • tissue (str | None)

  • collection_date (str | None)

  • metadata (dict | None)

get_sample(sample_id)[source]#

Return sample dict by sample_id, or None.

Return type:

dict[str, Any] | None

Parameters:

sample_id (str)

list_samples(patient_id=None, tissue=None)[source]#

List samples with optional filters.

Return type:

list[dict[str, Any]]

Parameters:

  • patient_id (str | None)

  • tissue (str | None)

status()[source]#

Return a high-level summary of current registry state.

Return type:

dict[str, Any]

register_data(project_name, phase, uri, *, fmt='h5ad', platform=None, category=None, status='ready', file_role='primary', n_obs=None, n_vars=None, size_mb=None, patient_id=None, metadata=None, sample_id=None)[source]#

Deprecated. Use register_inventory_item() instead.

Return type:

int

Parameters:

  • project_name (str)

  • phase (str)

  • uri (str)

  • fmt (str | None)

  • platform (str | None)

  • category (str | None)

  • status (str)

  • file_role (str)

  • n_obs (int | None)

  • n_vars (int | None)

  • size_mb (float | None)

  • patient_id (str | None)

  • metadata (dict | None)

  • sample_id (str | None)

register_dataset(project_name, phase, uri, *, sample_id=None, fmt='h5ad', size_mb=None, md5=None, status='ready', file_role='primary', validated=False, n_obs=None, n_vars=None)[source]#

Deprecated. Use register_inventory_item() instead.

Return type:

int

Parameters:

  • project_name (str)

  • phase (str)

  • uri (str)

  • sample_id (str | None)

  • fmt (str)

  • size_mb (float | None)

  • md5 (str | None)

  • status (str)

  • file_role (str)

  • validated (bool)

  • n_obs (int | None)

  • n_vars (int | None)

register_biodata(project_name, category, platform, uri, *, fmt=None, status='pending', file_role='primary', phase=None, n_obs=None, n_vars=None, size_mb=None, md5=None, subject_id=None, sample_db_id=None, **type_kwargs)[source]#

Deprecated. Use register_inventory_item() instead.

Return type:

int

Parameters:

  • project_name (str)

  • category (str)

  • platform (str)

  • uri (str)

  • fmt (str | None)

  • status (str)

  • file_role (str)

  • phase (str | None)

  • n_obs (int | None)

  • n_vars (int | None)

  • size_mb (float | None)

  • md5 (str | None)

  • subject_id (str | None)

  • sample_db_id (int | None)

  • type_kwargs (Any)

list_biodata(project_name=None, category=None, platform=None, sample_db_id=None, phase=None)[source]#

Deprecated. Use list_inventory_items() instead.

Return type:

list[dict[str, Any]]

Parameters:

  • project_name (str | None)

  • category (str | None)

  • platform (str | None)

  • sample_db_id (int | None)

  • phase (str | None)

Deprecated. Use link_project_dataset() instead.

Return type:

int

Parameters:

  • project_name (str)

  • data_source_name (str)

  • role (str)

  • notes (str | None)

list_project_data_sources(project_name)[source]#

Deprecated. Use list_project_datasets() instead.

Return type:

list[dict[str, Any]]

Parameters:

project_name (str)

get_biodata(biodata_id)[source]#

Return an inventory item dict by id.

Return type:

dict[str, Any] | None

Parameters:

biodata_id (int)

update_dataset_status(dataset_id, status)[source]#

Update dataset status.

Return type:

None

Parameters:

  • dataset_id (int)

  • status (str)