diff --git a/alembic/versions/0007_add_qc_read_tables.py b/alembic/versions/0007_add_qc_read_tables.py new file mode 100644 index 0000000..c2f62d4 --- /dev/null +++ b/alembic/versions/0007_add_qc_read_tables.py @@ -0,0 +1,304 @@ +"""Add qc_read, qc_read_file, qc_read_submission; drop read_submission. + +Revision ID: 0007_add_qc_read_tables +Revises: 0006_consolidated_refactor +Create Date: 2026-04-29 00:00:00.000000 +""" + +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +from alembic import op + +revision = "0007_add_qc_read_tables" +down_revision = "0006_consolidated_refactor" +branch_labels = None +depends_on = None + + +def upgrade(): + # 1. Extend the entity_type enum to include 'qc_read'. + # ALTER TYPE … ADD VALUE must be committed before the new value can be used. + # Since Alembic runs migrations in a transaction, we must manually commit, + # add the enum value, then begin a new transaction. + connection = op.get_bind() + connection.execute(sa.text("COMMIT")) + connection.execute(sa.text("ALTER TYPE entity_type ADD VALUE IF NOT EXISTS 'qc_read'")) + connection.execute(sa.text("BEGIN")) + + # 2. Create qc_read table. + op.create_table( + "qc_read", + sa.Column( + "id", + postgresql.UUID(as_uuid=True), + primary_key=True, + server_default=sa.text("uuid_generate_v4()"), + ), + sa.Column( + "experiment_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("experiment.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column("base_count", sa.BigInteger, nullable=False), + sa.Column("read_count", sa.BigInteger, nullable=False), + sa.Column("qc_bases_removed", sa.BigInteger, nullable=False), + sa.Column("qc_reads_removed", sa.BigInteger, nullable=False), + sa.Column("mean_gc_content", sa.Float, nullable=False), + sa.Column("n50_length", sa.BigInteger, nullable=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("NOW()"), + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("NOW()"), + ), + ) + op.create_index("idx_qc_read_experiment_id", "qc_read", ["experiment_id"]) + + # 3. Create qc_read_file table. + op.create_table( + "qc_read_file", + sa.Column( + "id", + postgresql.UUID(as_uuid=True), + primary_key=True, + server_default=sa.text("uuid_generate_v4()"), + ), + sa.Column( + "qc_read_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("qc_read.id", ondelete="CASCADE"), + nullable=False, + ), + # 'cram', 'fastq_r1', or 'fastq_r2' + sa.Column("file_type", sa.Text, nullable=False), + sa.Column("storage_backend", sa.Text, nullable=False), + sa.Column("storage_profile", sa.Text, nullable=False), + sa.Column("bucket_name", sa.Text, nullable=False), + sa.Column("path_to_file", sa.Text, nullable=False), + sa.Column("md5_checksum", sa.Text, nullable=False), + sa.Column("sha256_checksum", sa.Text, nullable=False), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("NOW()"), + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("NOW()"), + ), + sa.CheckConstraint( + "file_type IN ('cram', 'fastq_r1', 'fastq_r2')", + name="ck_qc_read_file_type", + ), + sa.CheckConstraint( + "md5_checksum ~ '^[a-f0-9]{32}$'", + name="ck_qc_read_file_md5", + ), + sa.CheckConstraint( + "sha256_checksum ~ '^[a-f0-9]{64}$'", + name="ck_qc_read_file_sha256", + ), + ) + op.create_index("idx_qc_read_file_qc_read_id", "qc_read_file", ["qc_read_id"]) + + # 4. Create qc_read_submission table (mirrors read_submission; FK to qc_read). + op.create_table( + "qc_read_submission", + sa.Column( + "id", + postgresql.UUID(as_uuid=True), + primary_key=True, + server_default=sa.text("uuid_generate_v4()"), + ), + sa.Column( + "qc_read_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("qc_read.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column( + "experiment_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("experiment.id", ondelete="CASCADE"), + nullable=True, + ), + sa.Column( + "authority", + postgresql.ENUM("ENA", "NCBI", "DDBJ", name="authority_type", create_type=False), + nullable=False, + server_default="ENA", + ), + sa.Column( + "status", + postgresql.ENUM( + "draft", + "ready", + "submitting", + "accepted", + "rejected", + "replaced", + name="submission_status", + create_type=False, + ), + nullable=False, + server_default="draft", + ), + sa.Column("prepared_payload", postgresql.JSONB, nullable=False), + sa.Column("response_payload", postgresql.JSONB, nullable=True), + sa.Column("accession", sa.Text, nullable=True), + sa.Column( + "entity_type_const", + postgresql.ENUM( + "organism", + "sample", + "experiment", + "read", + "assembly", + "project", + "qc_read", + name="entity_type", + create_type=False, + ), + nullable=False, + server_default="qc_read", + ), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("NOW()"), + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("NOW()"), + ), + sa.Column("attempt_id", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("finalised_attempt_id", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("lock_acquired_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("lock_expires_at", sa.DateTime(timezone=True), nullable=True), + sa.CheckConstraint( + "entity_type_const = 'qc_read'", + name="ck_qc_read_submission_entity_type", + ), + # Deferred FK to accession_registry, matching the read_submission pattern. + sa.ForeignKeyConstraint( + ["accession", "authority", "entity_type_const", "qc_read_id"], + [ + "accession_registry.accession", + "accession_registry.authority", + "accession_registry.entity_type", + "accession_registry.entity_id", + ], + name="fk_qc_read_submission_accession", + deferrable=True, + initially="DEFERRED", + ), + ) + op.create_index("idx_qc_read_submission_attempt", "qc_read_submission", ["attempt_id"]) + op.create_index( + "idx_qc_read_submission_finalised_attempt", + "qc_read_submission", + ["finalised_attempt_id"], + ) + op.create_index("idx_qc_read_submission_status", "qc_read_submission", ["status"]) + op.create_index( + "idx_qc_read_submission_lock_expires_at", "qc_read_submission", ["lock_expires_at"] + ) + op.create_index("idx_qc_read_submission_experiment_id", "qc_read_submission", ["experiment_id"]) + # Partial unique index: at most one accepted accession per (qc_read, authority). + op.execute( + """ + CREATE UNIQUE INDEX uq_qc_read_one_accepted + ON qc_read_submission (qc_read_id, authority) + WHERE status = 'accepted' AND accession IS NOT NULL + """ + ) + + # 5. Drop read_submission (no data migration required). + op.execute("DROP TABLE IF EXISTS read_submission CASCADE") + + +def downgrade(): + # Re-create read_submission + op.create_table( + "read_submission", + sa.Column( + "id", + postgresql.UUID(as_uuid=True), + primary_key=True, + server_default=sa.text("uuid_generate_v4()"), + ), + sa.Column( + "read_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("read.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column( + "authority", + postgresql.ENUM("ENA", "NCBI", "DDBJ", name="authority_type", create_type=False), + nullable=False, + server_default="ENA", + ), + sa.Column( + "status", + postgresql.ENUM( + "draft", + "ready", + "submitting", + "accepted", + "rejected", + "replaced", + name="submission_status", + create_type=False, + ), + nullable=False, + server_default="draft", + ), + sa.Column("prepared_payload", postgresql.JSONB, nullable=False), + sa.Column("response_payload", postgresql.JSONB, nullable=True), + sa.Column( + "experiment_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("experiment.id", ondelete="CASCADE"), + nullable=True, + ), + sa.Column("accession", sa.Text, nullable=True), + sa.Column("entity_type_const", sa.Text, nullable=False, server_default="read"), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("NOW()"), + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("NOW()"), + ), + sa.Column("attempt_id", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("finalised_attempt_id", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("lock_acquired_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("lock_expires_at", sa.DateTime(timezone=True), nullable=True), + ) + + # Drop qc_read_submission (cascades indexes) + op.drop_table("qc_read_submission") + # Drop qc_read_file and qc_read (cascade) + op.drop_table("qc_read_file") + op.drop_table("qc_read") + # Note: cannot remove enum values in PostgreSQL; qc_read stays in entity_type enum. diff --git a/alembic/versions/0008_add_base_url_to_experiment.py b/alembic/versions/0008_add_base_url_to_experiment.py new file mode 100644 index 0000000..100f0ef --- /dev/null +++ b/alembic/versions/0008_add_base_url_to_experiment.py @@ -0,0 +1,23 @@ +"""Add base_url column to experiment table. + +Revision ID: 0008_add_base_url_to_experiment +Revises: 0007_add_qc_read_tables +Create Date: 2026-04-30 00:00:00.000000 +""" + +import sqlalchemy as sa + +from alembic import op + +revision = "0008_add_base_url_to_experiment" +down_revision = "0007_add_qc_read_tables" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column("experiment", sa.Column("base_url", sa.Text(), nullable=True)) + + +def downgrade(): + op.drop_column("experiment", "base_url") diff --git a/app/api/v1/api.py b/app/api/v1/api.py index 185a7ff..446011c 100644 --- a/app/api/v1/api.py +++ b/app/api/v1/api.py @@ -11,7 +11,7 @@ genome_notes, organisms, projects, - read_submissions, + qc_reads, reads, sample_submissions, samples, @@ -44,9 +44,7 @@ ) api_router.include_router(projects.router, prefix="/projects", tags=["projects"]) api_router.include_router(reads.router, prefix="/reads", tags=["reads"]) -api_router.include_router( - read_submissions.router, prefix="/read-submissions", tags=["read-submissions"] -) +api_router.include_router(qc_reads.router, prefix="/qc-reads", tags=["qc-reads"]) api_router.include_router(genome_notes.router, prefix="/genome-notes", tags=["genome-notes"]) # XML export endpoints diff --git a/app/api/v1/endpoints/assemblies.py b/app/api/v1/endpoints/assemblies.py index 4d9bc67..c6ccc1e 100644 --- a/app/api/v1/endpoints/assemblies.py +++ b/app/api/v1/endpoints/assemblies.py @@ -353,9 +353,12 @@ def create_assembly_intent( tax_id: int, intent_in: Optional[AssemblyIntent] = Body(default=None), current_user: User = Depends(get_current_active_user), -) -> AssemblyIntentResponse: +) -> Response: """ - Reserve the next assembly version and return a manifest. + Reserve the next assembly version and return a manifest as YAML. + + Returns: + YAML response with manifest data including assembly_run_id, version, status, and manifest fields """ organism, selected_sample, reads, experiments = _get_manifest_inputs_by_tax_id(db, tax_id) try: @@ -391,15 +394,22 @@ def create_assembly_intent( db.refresh(run) sample_metadata_by_id = _build_sample_metadata_by_id(db, experiments) - yaml_content = generate_assembly_manifest( + manifest_yaml = generate_assembly_manifest( organism, reads, experiments, run.tol_id, run.version, sample_metadata_by_id ) - return AssemblyIntentResponse( - assembly_run_id=run.id, - version=run.version, - status=run.status, - manifest_yaml=yaml_content, - ) + + # Build complete YAML response including metadata and manifest + import yaml + + response_data = { + "assembly_run_id": str(run.id), + "version": run.version, + "status": run.status, + "manifest": yaml.safe_load(manifest_yaml), + } + yaml_response = yaml.dump(response_data, default_flow_style=False, sort_keys=False) + + return Response(content=yaml_response, media_type="application/x-yaml") @router.post("/intent/{tax_id}/cancel") diff --git a/app/api/v1/endpoints/broker.py b/app/api/v1/endpoints/broker.py index 2d476d6..43d28b4 100644 --- a/app/api/v1/endpoints/broker.py +++ b/app/api/v1/endpoints/broker.py @@ -17,7 +17,7 @@ from app.models.experiment import Experiment, ExperimentSubmission from app.models.organism import Organism from app.models.project import Project, ProjectSubmission -from app.models.read import Read, ReadSubmission +from app.models.qc_read import QcRead, QcReadSubmission from app.models.sample import Sample, SampleSubmission from app.models.user import User from app.schemas.broker_contract import ( @@ -79,20 +79,22 @@ class ClaimRequest(BaseModel): # Optional explicit selection by submission IDs; if provided, organism_key is not enforced sample_submission_ids: Optional[List[UUID]] = None experiment_submission_ids: Optional[List[UUID]] = None - read_submission_ids: Optional[List[UUID]] = None + qc_read_submission_ids: Optional[List[UUID]] = None project_submission_ids: Optional[List[UUID]] = None # option to override lease, in mins lease_duration_minutes: Optional[int] = Field(default=None, ge=1, le=180) class ClaimByEntityRequest(BaseModel): - """Request to claim specific samples, experiments, and reads by their entity IDs.""" + """Request to claim specific samples, experiments, and qc_reads by their entity IDs.""" sample_ids: Optional[List[UUID]] = Field(default=None, description="Sample entity IDs to claim") experiment_ids: Optional[List[UUID]] = Field( default=None, description="Experiment entity IDs to claim" ) - read_ids: Optional[List[UUID]] = Field(default=None, description="Read entity IDs to claim") + qc_read_ids: Optional[List[UUID]] = Field( + default=None, description="QcRead entity IDs to claim" + ) project_ids: Optional[List[UUID]] = Field( default=None, description="Project entity IDs to claim" ) @@ -195,8 +197,8 @@ def _create_new_draft_submission_after_rejection( model_cls = ExperimentSubmission entity_fk_field = "experiment_id" elif entity_type == BrokerEntityType.RUN: - model_cls = ReadSubmission - entity_fk_field = "read_id" + model_cls = QcReadSubmission + entity_fk_field = "qc_read_id" elif entity_type == BrokerEntityType.PROJECT: model_cls = ProjectSubmission entity_fk_field = "project_id" @@ -231,6 +233,8 @@ def _create_new_draft_submission_after_rejection( model_column_names = set(getattr(model_cls, "__table__").columns.keys()) if hasattr(row, "project_id") and "project_id" in model_column_names: kwargs["project_id"] = getattr(row, "project_id", None) + if hasattr(row, "experiment_id") and "experiment_id" in model_column_names: + kwargs["experiment_id"] = getattr(row, "experiment_id", None) if hasattr(row, "biosample_accession") and "biosample_accession" in model_column_names: kwargs["biosample_accession"] = getattr(row, "biosample_accession", None) @@ -479,18 +483,18 @@ def _query_ready_experiment_submissions(db: Session, tax_id: int) -> List[Experi ) -def _query_ready_run_submissions(db: Session, tax_id: int) -> List[ReadSubmission]: +def _query_ready_run_submissions(db: Session, tax_id: int) -> List[QcReadSubmission]: return ( - db.query(ReadSubmission) - .join(Read, ReadSubmission.read_id == Read.id) - .join(Experiment, Read.experiment_id == Experiment.id) + db.query(QcReadSubmission) + .join(QcRead, QcReadSubmission.qc_read_id == QcRead.id) + .join(Experiment, QcRead.experiment_id == Experiment.id) .join(Sample, Experiment.sample_id == Sample.id) .join(Organism, Sample.organism_key == Organism.grouping_key) .filter( Organism.tax_id == tax_id, - ReadSubmission.status.in_(CLAIMABLE_SUBMISSION_STATES), + QcReadSubmission.status.in_(CLAIMABLE_SUBMISSION_STATES), ) - .order_by(ReadSubmission.created_at.desc()) + .order_by(QcReadSubmission.created_at.desc()) .with_for_update(skip_locked=True) .all() ) @@ -546,12 +550,12 @@ def _find_latest_claimable_entity_submission( .first() ) return ( - db.query(ReadSubmission) + db.query(QcReadSubmission) .filter( - ReadSubmission.read_id == entity_id, - ReadSubmission.status.in_(CLAIMABLE_SUBMISSION_STATES), + QcReadSubmission.qc_read_id == entity_id, + QcReadSubmission.status.in_(CLAIMABLE_SUBMISSION_STATES), ) - .order_by(ReadSubmission.created_at.desc()) + .order_by(QcReadSubmission.created_at.desc()) .with_for_update(skip_locked=True) .first() ) @@ -583,9 +587,9 @@ def _find_latest_submission_for_validation( .first() ) return ( - db.query(ReadSubmission) - .filter(ReadSubmission.read_id == entity_id) - .order_by(ReadSubmission.created_at.desc()) + db.query(QcReadSubmission) + .filter(QcReadSubmission.qc_read_id == entity_id) + .order_by(QcReadSubmission.created_at.desc()) .first() ) @@ -617,12 +621,13 @@ def _lookup_taxonomy_for_entity( .first() ) else: + # RUN entity_id is a qc_read_id; traverse qc_read → experiment → sample → organism row = ( db.query(Sample.organism_key, Organism.tax_id) .join(Experiment, Experiment.sample_id == Sample.id) - .join(Read, Read.experiment_id == Experiment.id) + .join(QcRead, QcRead.experiment_id == Experiment.id) .join(Organism, Sample.organism_key == Organism.grouping_key) - .filter(Read.id == entity_id) + .filter(QcRead.id == entity_id) .first() ) @@ -745,9 +750,12 @@ def _find_submission_for_attempt( return result if entity_type == BrokerEntityType.RUN: result = ( - db.query(ReadSubmission) - .filter(ReadSubmission.read_id == entity_id, ReadSubmission.attempt_id == attempt_id) - .order_by(ReadSubmission.created_at.desc()) + db.query(QcReadSubmission) + .filter( + QcReadSubmission.qc_read_id == entity_id, + QcReadSubmission.attempt_id == attempt_id, + ) + .order_by(QcReadSubmission.created_at.desc()) .first() ) return result @@ -774,8 +782,8 @@ def _register_submission_accession( registry_entity_type = "experiment" registry_entity_id = row.experiment_id else: - registry_entity_type = "read" - registry_entity_id = row.read_id + registry_entity_type = "qc_read" + registry_entity_id = row.qc_read_id if registry_entity_id is None: return @@ -813,7 +821,7 @@ def claim_ready_entities( experiment_rows = _latest_per_entity( _query_ready_experiment_submissions(db, tax_id), "experiment_id" ) - run_rows = _latest_per_entity(_query_ready_run_submissions(db, tax_id), "read_id") + run_rows = _latest_per_entity(_query_ready_run_submissions(db, tax_id), "qc_read_id") if not any([project_rows, sample_rows, experiment_rows, run_rows]): # Return empty response when organism exists but has no claimable entities @@ -844,8 +852,10 @@ def claim_ready_entities( _build_contract_entity(db, BrokerEntityType.EXPERIMENT, row.experiment_id, tax_id, row) ) for row in run_rows: - _claim_submission_row(db, attempt=attempt, row=row, entity_type="read", now=now) - entities.append(_build_contract_entity(db, BrokerEntityType.RUN, row.read_id, tax_id, row)) + _claim_submission_row(db, attempt=attempt, row=row, entity_type="qc_read", now=now) + entities.append( + _build_contract_entity(db, BrokerEntityType.RUN, row.qc_read_id, tax_id, row) + ) db.commit() @@ -880,7 +890,7 @@ def claim_specific_entity( db, attempt=attempt, row=row, - entity_type="read" + entity_type="qc_read" if payload.entity_type == BrokerEntityType.RUN else payload.entity_type.value, now=datetime.now(timezone.utc), @@ -961,7 +971,7 @@ def claim_batch_entities( db, attempt=attempt, row=row, - entity_type="read" if entity_type == BrokerEntityType.RUN else entity_type.value, + entity_type="qc_read" if entity_type == BrokerEntityType.RUN else entity_type.value, now=now, ) entities.append(_build_contract_entity(db, entity_type, entity_id, tax_id, row)) @@ -1045,7 +1055,7 @@ def report_submission_outcomes( db.add( SubmissionEvent( attempt_id=attempt_id, - entity_type="read" + entity_type="qc_read" if _coerce_entity_type(result.entity_type) == BrokerEntityType.RUN else _coerce_entity_type(result.entity_type).value, submission_id=row.id, @@ -1108,7 +1118,9 @@ def claim_by_entity_ids( # Expire any stale leases before claiming expire_stale_leases(db) - if not any([payload.sample_ids, payload.experiment_ids, payload.read_ids, payload.project_ids]): + if not any( + [payload.sample_ids, payload.experiment_ids, payload.qc_read_ids, payload.project_ids] + ): raise HTTPException( status_code=400, detail="At least one of sample_ids, experiment_ids, read_ids, or project_ids must be provided", @@ -1320,27 +1332,32 @@ def claim_by_entity_ids( ) ) - # Claim reads by entity IDs - if payload.read_ids: - read_rank_subq = ( + # Claim qc_reads by entity IDs + if payload.qc_read_ids: + qc_read_rank_subq = ( db.query( - ReadSubmission.id.label("id"), - ReadSubmission.read_id.label("read_id"), + QcReadSubmission.id.label("id"), + QcReadSubmission.qc_read_id.label("qc_read_id"), func.row_number() .over( - partition_by=ReadSubmission.read_id, - order_by=ReadSubmission.created_at.desc(), + partition_by=QcReadSubmission.qc_read_id, + order_by=QcReadSubmission.created_at.desc(), ) .label("rn"), - ).filter(ReadSubmission.read_id.in_(payload.read_ids), ReadSubmission.status == "draft") + ).filter( + QcReadSubmission.qc_read_id.in_(payload.qc_read_ids), + QcReadSubmission.status == "draft", + ) ).subquery() - read_ids_subq = (db.query(read_rank_subq.c.id).filter(read_rank_subq.c.rn == 1)).subquery() + qc_read_ids_subq = ( + db.query(qc_read_rank_subq.c.id).filter(qc_read_rank_subq.c.rn == 1) + ).subquery() read_rows = ( - db.query(ReadSubmission) - .filter(ReadSubmission.id.in_(db.query(read_ids_subq.c.id))) - .filter(ReadSubmission.status == "draft") + db.query(QcReadSubmission) + .filter(QcReadSubmission.id.in_(db.query(qc_read_ids_subq.c.id))) + .filter(QcReadSubmission.status == "draft") .with_for_update(skip_locked=True) .all() ) @@ -1353,17 +1370,17 @@ def claim_by_entity_ids( db.add( SubmissionEvent( attempt_id=attempt_id, - entity_type="read", + entity_type="qc_read", submission_id=row.id, action="claimed", ) ) db.commit() - # Build relationships for reads -> experiment/experiment_submission + # Build relationships for qc_reads -> experiment/experiment_submission read_exp_ids = [r.experiment_id for r in read_rows] - # Get organism keys from reads via experiments and samples + # Get organism keys from qc_reads via experiments and samples if read_exp_ids: organism_keys = ( db.query(Sample.organism_key) @@ -1379,7 +1396,7 @@ def claim_by_entity_ids( r.experiment_id: r for r in exp_rows } - # For reads whose experiments weren't claimed, fall back to latest accepted experiment submission + # For qc_reads whose experiments weren't claimed, fall back to latest accepted missing_exp_ids = [ eid for eid in set(read_exp_ids) if eid not in claimed_exp_by_experiment_id ] @@ -1412,7 +1429,7 @@ def claim_by_entity_ids( } claimed_reads.append( ClaimedEntity( - id=row.read_id, + id=row.qc_read_id, submission_id=row.id, status=row.status, prepared_payload=row.prepared_payload, @@ -1729,40 +1746,42 @@ def claim_drafts_for_organism( ) ) - # Choose read rows by explicit IDs (if provided) else by organism/limit - if payload and payload.read_submission_ids: + # Choose qc_read rows by explicit IDs (if provided) else by organism/limit + if payload and payload.qc_read_submission_ids: read_rows = ( - db.query(ReadSubmission) - .filter(ReadSubmission.id.in_(payload.read_submission_ids)) - .filter(ReadSubmission.status == "draft") + db.query(QcReadSubmission) + .filter(QcReadSubmission.id.in_(payload.qc_read_submission_ids)) + .filter(QcReadSubmission.status == "draft") .with_for_update(skip_locked=True) .all() ) else: - read_rank_subq = ( + qc_read_rank_subq = ( db.query( - ReadSubmission.id.label("id"), + QcReadSubmission.id.label("id"), func.row_number() .over( - partition_by=ReadSubmission.read_id, - order_by=ReadSubmission.created_at.desc(), + partition_by=QcReadSubmission.qc_read_id, + order_by=QcReadSubmission.created_at.desc(), ) .label("rn"), ) - .join(Read, ReadSubmission.read_id == Read.id) - .join(Experiment, Read.experiment_id == Experiment.id) + .join(QcRead, QcReadSubmission.qc_read_id == QcRead.id) + .join(Experiment, QcRead.experiment_id == Experiment.id) .join(Sample, Experiment.sample_id == Sample.id) - .filter(Sample.organism_key == organism_key, ReadSubmission.status == "draft") + .filter(Sample.organism_key == organism_key, QcReadSubmission.status == "draft") ).subquery() - read_ids_subq = ( - db.query(read_rank_subq.c.id).filter(read_rank_subq.c.rn == 1).limit(per_type_limit) + qc_read_ids_subq = ( + db.query(qc_read_rank_subq.c.id) + .filter(qc_read_rank_subq.c.rn == 1) + .limit(per_type_limit) ).subquery() read_rows = ( - db.query(ReadSubmission) - .filter(ReadSubmission.id.in_(db.query(read_ids_subq.c.id))) - .filter(ReadSubmission.status == "draft") + db.query(QcReadSubmission) + .filter(QcReadSubmission.id.in_(db.query(qc_read_ids_subq.c.id))) + .filter(QcReadSubmission.status == "draft") .with_for_update(skip_locked=True) .all() ) @@ -1773,19 +1792,19 @@ def claim_drafts_for_organism( row.lock_expires_at = attempt.lock_expires_at db.add( SubmissionEvent( - attempt_id=attempt_id, entity_type="read", submission_id=row.id, action="claimed" + attempt_id=attempt_id, entity_type="qc_read", submission_id=row.id, action="claimed" ) ) db.commit() - # Build relationships for reads -> experiment/experiment_submission + # Build relationships for qc_reads -> experiment/experiment_submission read_exp_ids = [r.experiment_id for r in read_rows] # Map of claimed experiment submissions: experiment_id -> ExperimentSubmission row claimed_exp_by_experiment_id: Dict[UUID, ExperimentSubmission] = { r.experiment_id: r for r in exp_rows } - # For reads whose experiments weren't claimed, fall back to latest accepted experiment submission + # For qc_reads whose experiments weren't claimed, fall back to latest accepted missing_exp_ids = [eid for eid in set(read_exp_ids) if eid not in claimed_exp_by_experiment_id] accepted_exp_by_experiment_id: Dict[UUID, ExperimentSubmission] = {} if missing_exp_ids: @@ -1816,7 +1835,7 @@ def claim_drafts_for_organism( } claimed_reads.append( ClaimedEntity( - id=row.read_id, + id=row.qc_read_id, submission_id=row.id, status=row.status, prepared_payload=row.prepared_payload, @@ -1998,8 +2017,8 @@ def renew_attempt_lease( ): sub.lock_expires_at = new_exp for sub in ( - db.query(ReadSubmission) - .filter(ReadSubmission.attempt_id == attempt_id, ReadSubmission.status == "submitting") + db.query(QcReadSubmission) + .filter(QcReadSubmission.attempt_id == attempt_id, QcReadSubmission.status == "submitting") .all() ): sub.lock_expires_at = new_exp @@ -2071,10 +2090,10 @@ def finalise_attempt( ) released["experiments"] += 1 - # Reads + # QC reads read_rows = ( - db.query(ReadSubmission) - .filter(ReadSubmission.attempt_id == attempt_id, ReadSubmission.status == "submitting") + db.query(QcReadSubmission) + .filter(QcReadSubmission.attempt_id == attempt_id, QcReadSubmission.status == "submitting") .all() ) for sub in read_rows: @@ -2084,7 +2103,10 @@ def finalise_attempt( sub.lock_expires_at = None db.add( SubmissionEvent( - attempt_id=attempt_id, entity_type="read", submission_id=sub.id, action="released" + attempt_id=attempt_id, + entity_type="qc_read", + submission_id=sub.id, + action="released", ) ) released["reads"] += 1 @@ -2298,26 +2320,29 @@ def report_results( updated_experiments += 1 - # Process ReadSubmission updates + # Process QcReadSubmission updates for item in payload.reads: submission_id = item.submission_id or item.id - sub = db.query(ReadSubmission).filter(ReadSubmission.id == submission_id).first() + sub = db.query(QcReadSubmission).filter(QcReadSubmission.id == submission_id).first() if not sub: - raise HTTPException(status_code=404, detail=f"ReadSubmission {submission_id} not found") - if item.submission_id and sub.read_id != item.id: + raise HTTPException( + status_code=404, detail=f"QcReadSubmission {submission_id} not found" + ) + if item.submission_id and sub.qc_read_id != item.id: raise HTTPException( status_code=409, - detail=f"ReadSubmission {submission_id} does not match read entity id {item.id}", + detail=f"QcReadSubmission {submission_id} does not match qc_read entity id {item.id}", ) if sub.status != "submitting": raise HTTPException( - status_code=409, detail=f"ReadSubmission {submission_id} not in 'submitting' state" + status_code=409, + detail=f"QcReadSubmission {submission_id} not in 'submitting' state", ) if sub.attempt_id != provided_attempt_id: raise HTTPException( status_code=409, - detail=f"ReadSubmission {submission_id} belongs to different attempt", + detail=f"QcReadSubmission {submission_id} belongs to different attempt", ) sub.status = item.status @@ -2335,17 +2360,16 @@ def report_results( ) stmt = stmt.on_conflict_do_nothing(index_elements=[AccessionRegistry.accession]) db.execute(stmt) - sub.experiment_accession = item.experiment_accession db.add(sub) db.flush() - if item.accession and sub.read_id is not None: + if item.accession and sub.qc_read_id is not None: stmt = insert(AccessionRegistry).values( authority=sub.authority or "ENA", accession=item.accession, - entity_type="read", - entity_id=sub.read_id, + entity_type="qc_read", + entity_id=sub.qc_read_id, accepted_at=item.submitted_at or datetime.now(timezone.utc), ) stmt = stmt.on_conflict_do_nothing(index_elements=[AccessionRegistry.accession]) @@ -2359,7 +2383,7 @@ def report_results( db.add( SubmissionEvent( attempt_id=attempt_id, - entity_type="read", + entity_type="qc_read", submission_id=sub.id, action=( "accepted" @@ -2370,6 +2394,12 @@ def report_results( details=item.response_payload, ) ) + if item.status == "rejected": + _create_new_draft_submission_after_rejection( + db, + entity_type=BrokerEntityType.RUN, + row=sub, + ) updated_reads += 1 @@ -2516,13 +2546,13 @@ def expire_stale_leases(db: Session) -> Dict[str, int]: ) expired_counts["experiments"] += 1 - # Expire read submissions + # Expire qc_read submissions read_rows = ( - db.query(ReadSubmission) + db.query(QcReadSubmission) .filter( - ReadSubmission.status == "submitting", - ReadSubmission.lock_expires_at.isnot(None), - ReadSubmission.lock_expires_at <= now, + QcReadSubmission.status == "submitting", + QcReadSubmission.lock_expires_at.isnot(None), + QcReadSubmission.lock_expires_at <= now, ) .all() ) @@ -2536,7 +2566,7 @@ def expire_stale_leases(db: Session) -> Dict[str, int]: db.add( SubmissionEvent( attempt_id=attempt_id, - entity_type="read", + entity_type="qc_read", submission_id=sub.id, action="expired", ) @@ -2628,11 +2658,11 @@ def _counts_by_entity_for_attempt(db: Session, attempt_id: UUID) -> Dict[str, Di ), "reads": _counts_for_model( db, - ReadSubmission, + QcReadSubmission, [ or_( - ReadSubmission.attempt_id == attempt_id, - ReadSubmission.finalised_attempt_id == attempt_id, + QcReadSubmission.attempt_id == attempt_id, + QcReadSubmission.finalised_attempt_id == attempt_id, ) ], ), @@ -2698,11 +2728,11 @@ def _get_attempt_items_with_relationships( ) read_ids: set[UUID] = set( x[0] - for x in db.query(ReadSubmission.id) + for x in db.query(QcReadSubmission.id) .filter( or_( - ReadSubmission.attempt_id == attempt_id, - ReadSubmission.finalised_attempt_id == attempt_id, + QcReadSubmission.attempt_id == attempt_id, + QcReadSubmission.finalised_attempt_id == attempt_id, ) ) .all() @@ -2740,7 +2770,7 @@ def _get_attempt_items_with_relationships( db.query(SubmissionEvent.submission_id) .filter( SubmissionEvent.attempt_id == attempt_id, - SubmissionEvent.entity_type == "read", + SubmissionEvent.entity_type == "qc_read", ) .all() ) @@ -2769,7 +2799,9 @@ def _get_attempt_items_with_relationships( else [] ) reads = ( - db.query(ReadSubmission).filter(ReadSubmission.id.in_(read_ids)).all() if read_ids else [] + db.query(QcReadSubmission).filter(QcReadSubmission.id.in_(read_ids)).all() + if read_ids + else [] ) projects = ( db.query(ProjectSubmission).filter(ProjectSubmission.id.in_(project_ids)).all() @@ -2878,7 +2910,7 @@ def _get_attempt_items_with_relationships( } read_entity_list.append( ClaimedEntity( - id=row.read_id, + id=row.qc_read_id, submission_id=row.id, status=row.status, prepared_payload=row.prepared_payload, @@ -3053,14 +3085,14 @@ def organism_summary( for st, cnt in e_rows: e_counts[st] = cnt - # Reads: ReadSubmission -> Read -> Experiment -> Sample + # QC reads: QcReadSubmission -> QcRead -> Experiment -> Sample r_rows = ( - db.query(ReadSubmission.status, func.count()) - .join(Read, ReadSubmission.read_id == Read.id) - .join(Experiment, Read.experiment_id == Experiment.id) + db.query(QcReadSubmission.status, func.count()) + .join(QcRead, QcReadSubmission.qc_read_id == QcRead.id) + .join(Experiment, QcRead.experiment_id == Experiment.id) .join(Sample, Experiment.sample_id == Sample.id) .filter(Sample.organism_key == organism_key) - .group_by(ReadSubmission.status) + .group_by(QcReadSubmission.status) .all() ) r_counts: Dict[str, int] = {"draft": 0, "submitting": 0, "accepted": 0, "rejected": 0} diff --git a/app/api/v1/endpoints/qc_reads.py b/app/api/v1/endpoints/qc_reads.py new file mode 100644 index 0000000..1103ee3 --- /dev/null +++ b/app/api/v1/endpoints/qc_reads.py @@ -0,0 +1,159 @@ +"""QC read endpoints — CRUD and genome launcher report callback.""" + +from typing import Any, List, Optional +from uuid import UUID + +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy.orm import Session + +from app.core.dependencies import get_current_active_user, get_db +from app.core.pagination import Pagination, apply_pagination, pagination_params +from app.core.policy import policy +from app.models.experiment import Experiment +from app.models.qc_read import QcRead, QcReadFile, QcReadSubmission +from app.models.user import User +from app.schemas.qc_read import QcCallbackRequest, QcReadOut + +router = APIRouter() + + +def _build_prepared_payload(qc_read: QcRead, files: list[QcReadFile]) -> dict: + """Build the ENA submission payload stored on QcReadSubmission. + + The broker's to_run_xml() expects a ``files`` list where each entry has: + filename, filetype, checksum (MD5), checksum_method. + """ + return { + "files": [ + { + "filename": f.path_to_file, + "filetype": f.file_type.replace("_r1", "").replace("_r2", ""), + "checksum": f.md5_checksum, + "checksum_method": "MD5", + } + for f in files + ] + } + + +# --------------------------------------------------------------------------- +# CRUD +# --------------------------------------------------------------------------- + + +@router.get("/", response_model=List[QcReadOut]) +def list_qc_reads( + db: Session = Depends(get_db), + pagination: Pagination = Depends(pagination_params), + experiment_id: Optional[UUID] = Query(None, description="Filter by experiment ID"), + current_user: User = Depends(get_current_active_user), +) -> Any: + """List QC reads, optionally filtered by experiment.""" + query = db.query(QcRead) + if experiment_id: + query = query.filter(QcRead.experiment_id == experiment_id) + return apply_pagination(query, pagination).all() + + +@router.get("/{qc_read_id}", response_model=QcReadOut) +def get_qc_read( + *, + db: Session = Depends(get_db), + qc_read_id: UUID, + current_user: User = Depends(get_current_active_user), +) -> Any: + """Get a single QC read by ID, including its files and submission records.""" + qc_read = db.query(QcRead).filter(QcRead.id == qc_read_id).first() + if not qc_read: + raise HTTPException(status_code=404, detail="QC read not found") + return qc_read + + +@router.delete("/{qc_read_id}", response_model=QcReadOut) +@policy("qc_reads:delete") +def delete_qc_read( + *, + db: Session = Depends(get_db), + qc_read_id: UUID, + current_user: User = Depends(get_current_active_user), +) -> Any: + """Delete a QC read and its associated files and submission records.""" + qc_read = db.query(QcRead).filter(QcRead.id == qc_read_id).first() + if not qc_read: + raise HTTPException(status_code=404, detail="QC read not found") + db.delete(qc_read) + db.commit() + return qc_read + + +# --------------------------------------------------------------------------- +# Genome launcher callback +# --------------------------------------------------------------------------- + + +@router.post("/report", response_model=QcReadOut, status_code=201) +@policy("qc_reads:report") +def report_qc_result( + *, + payload: QcCallbackRequest, + current_user: User = Depends(get_current_active_user), + db: Session = Depends(get_db), +) -> QcRead: + """Accept a QC result from the genome launcher and create submission records. + + The genome launcher identifies the target experiment by ``bpa_package_id``. + On success, ``qc_read``, ``qc_read_file``, and ``qc_read_submission`` rows + are created atomically and the ``qc_read`` is returned. + """ + experiment = ( + db.query(Experiment).filter(Experiment.bpa_package_id == payload.bpa_package_id).first() + ) + if not experiment: + raise HTTPException( + status_code=404, + detail=f"Experiment with bpa_package_id '{payload.bpa_package_id}' not found", + ) + + qc_read = QcRead( + experiment_id=experiment.id, + base_count=payload.base_count, + read_count=payload.read_count, + qc_bases_removed=payload.qc_bases_removed, + qc_reads_removed=payload.qc_reads_removed, + mean_gc_content=payload.mean_gc_content, + n50_length=payload.n50_length, + ) + db.add(qc_read) + db.flush() + + qc_files = [ + QcReadFile( + qc_read_id=qc_read.id, + file_type=f.file_type, + storage_backend=f.storage_backend, + storage_profile=f.storage_profile, + bucket_name=f.bucket_name, + path_to_file=f.path_to_file, + md5_checksum=f.md5_checksum, + sha256_checksum=f.sha256_checksum, + ) + for f in payload.files + ] + for qf in qc_files: + db.add(qf) + db.flush() + + prepared_payload = _build_prepared_payload(qc_read, qc_files) + + submission = QcReadSubmission( + qc_read_id=qc_read.id, + experiment_id=experiment.id, + authority="ENA", + status="draft", + prepared_payload=prepared_payload, + entity_type_const="qc_read", + ) + db.add(submission) + db.commit() + db.refresh(qc_read) + return qc_read diff --git a/app/api/v1/endpoints/reads.py b/app/api/v1/endpoints/reads.py index 996fbe6..0cbcff5 100644 --- a/app/api/v1/endpoints/reads.py +++ b/app/api/v1/endpoints/reads.py @@ -1,19 +1,15 @@ -import json -import os import uuid -from typing import Any, Dict, List, Optional +from typing import Any, List, Optional from uuid import UUID -from fastapi import APIRouter, Depends, HTTPException, Query, status +from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session -from sqlalchemy.orm.attributes import flag_modified from app.core.dependencies import get_current_active_user, get_db from app.core.pagination import Pagination, apply_pagination, pagination_params from app.core.policy import policy -from app.models.read import Read, ReadSubmission +from app.models.read import Read from app.models.user import User -from app.schemas.common import SubmissionJsonResponse, SubmissionStatus from app.schemas.read import ( Read as ReadSchema, ) @@ -35,7 +31,6 @@ def read_reads( """ Retrieve reads. """ - # All users can read reads query = db.query(Read) if experiment_id: query = query.filter(Read.experiment_id == experiment_id) @@ -57,78 +52,22 @@ def create_read( """ read_id = uuid.uuid4() - # Auto-map from Pydantic schema to Read columns read_data = read_in.model_dump(exclude_unset=True) allowed_cols = {c.name for c in Read.__table__.columns} exclude_keys = {"id", "created_at", "updated_at", "bpa_json"} read_kwargs = {k: v for k, v in read_data.items() if k in (allowed_cols - exclude_keys)} - # Normalize optional_file if present if "optional_file" in read_kwargs and read_kwargs["optional_file"] is not None: val = read_kwargs["optional_file"] if not isinstance(val, bool): read_kwargs["optional_file"] = str(val).lower() in ("true", "1", "yes") - read = Read( - id=read_id, - # bpa_json=read_data, - **read_kwargs, - ) + read = Read(id=read_id, **read_kwargs) db.add(read) - - read_data = read_in.dict(exclude_unset=True) - # Load the ENA-ATOL mapping file - ena_atol_map_path = os.path.join( - os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__)))), - "config", - "ena-atol-map.json", - ) - with open(ena_atol_map_path, "r") as f: - ena_atol_map = json.load(f) - # Generate ENA-mapped data for submission to ENA - prepared_payload = {} - for ena_key, atol_key in ena_atol_map["run"].items(): - if atol_key in read_data: - prepared_payload[ena_key] = read_data[atol_key] - - read_submission = ReadSubmission( - read_id=read_id, - experiment_id=read_in.experiment_id, - project_id=read_in.project_id, - entity_type_const="read", - prepared_payload=prepared_payload, - status=SubmissionStatus.DRAFT, - ) - db.add(read_submission) db.commit() db.refresh(read) - db.refresh(read_submission) return read -@router.get("/{read_id}/prepared-payload", response_model=SubmissionJsonResponse) -def get_read_prepared_payload( - *, - db: Session = Depends(get_db), - read_id: UUID, - current_user: User = Depends(get_current_active_user), -) -> Any: - """ - Get prepared_payload for a specific read submission. - """ - read_submission = db.query(ReadSubmission).filter(ReadSubmission.read_id == read_id).first() - if not read_submission: - raise HTTPException( - status_code=404, - detail="Read submission not found", - ) - if not read_submission.prepared_payload: - raise HTTPException( - status_code=404, - detail="Prepared payload not found for this read submission", - ) - return {"prepared_payload": read_submission.prepared_payload} - - @router.get("/{read_id}", response_model=ReadSchema) def read_read( *, @@ -139,7 +78,6 @@ def read_read( """ Get read by ID. """ - # All users can read read details read = db.query(Read).filter(Read.id == read_id).first() if not read: raise HTTPException(status_code=404, detail="Read not found") @@ -158,106 +96,14 @@ def update_read( """ Update a read. """ - read = db.query(Read).filter(Read.id == read_id).first() if not read: raise HTTPException(status_code=404, detail="Read not found") read_data = read_in.dict(exclude_unset=True) - # Load the ENA-ATOL mapping file - ena_atol_map_path = os.path.join( - os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__)))), - "config", - "ena-atol-map.json", - ) - with open(ena_atol_map_path, "r") as f: - ena_atol_map = json.load(f) - # Generate ENA-mapped data for submission to ENA - prepared_payload = {} - for ena_key, atol_key in ena_atol_map["run"].items(): - if atol_key in read_data: - prepared_payload[ena_key] = read_data[atol_key] - - read_submission = ( - db.query(ReadSubmission) - .filter(ReadSubmission.read_id == read_id) - .order_by(ReadSubmission.updated_at.desc()) - .first() - ) - new_read_submission = None - latest_read_submission = None - if not read_submission: - new_read_submission = ReadSubmission( - read_id=read_id, - experiment_id=read_in.experiment_id, - project_id=read_in.project_id, - entity_type_const="read", - prepared_payload=prepared_payload, - status=SubmissionStatus.DRAFT, - ) - db.add(new_read_submission) - else: - latest_read_submission = read_submission - if latest_read_submission.status == "submitting": - raise HTTPException( - status_code=404, - detail=f"Read with id: {read_id} is currently being submitted to ENA and could not be updated. Please try again later.", - ) - elif latest_read_submission.status == "rejected" or read_submission.status == "replaced": - # leave the old record for logs and create a new record - # retain accessions if they exist (accessions may not exist if status is 'rejected' and the sample has not successfully been submitted in the past) - new_read_submission = ReadSubmission( - read_id=read_id, - experiment_id=read_in.experiment_id, - project_id=read_in.project_id, - authority=read_submission.authority, - entity_type_const="read", - prepared_payload=prepared_payload, - response_payload=None, - accession=experiment_submission.accession, - biosample_accession=experiment_submission.biosample_accession, - status="draft", - ) - db.add(new_read_submission) - - elif latest_read_submission.status == "accepted": - # change old record's status to "replaced" and create a new record - # retain accessions - setattr(latest_read_submission, "status", "replaced") - db.add(latest_read_submission) - new_read_submission = ReadSubmission( - read_id=read_id, - experiment_id=read_in.experiment_id, - project_id=read_in.project_id, - authority=read_submission.authority, - entity_type_const="read", - prepared_payload=prepared_payload, - response_payload=None, - accession=experiment_submission.accession, - biosample_accession=experiment_submission.biosample_accession, - status="draft", - ) - db.add(new_read_submission) - elif latest_read_submission.status == "draft" or latest_read_submission.status == "ready": - # update the existing record, since it has not yet been submitted to ENA (set status = 'draft') - setattr(latest_read_submission, "prepared_payload", prepared_payload) - setattr(latest_read_submission, "status", "draft") - db.add(latest_read_submission) - # update the experiment_submission object - - setattr(read, "bpa_resource_id", read_in.bpa_resource_id) - setattr(read, "experiment_id", read_in.experiment_id) - setattr(read, "project_id", read_in.project_id) - # initiate new bpa_json object to the previous bpa_json object - """ - new_bpa_json = read.bpa_json - for field, value in read_data.items(): - if field != "experiment_id" and field != "project_id": - new_bpa_json[field] = value - read.bpa_json = new_bpa_json - flag_modified(read, "bpa_json") - """ + setattr(read, field, value) + db.add(read) db.commit() db.refresh(read) diff --git a/app/core/policy.py b/app/core/policy.py index ac12a29..131094c 100644 --- a/app/core/policy.py +++ b/app/core/policy.py @@ -38,9 +38,9 @@ "reads:create": ["curator", "admin"], "reads:update": ["curator", "admin"], "reads:delete": ["admin", "superuser"], - # Read submissions - "read_submissions:read": ["admin", "curator", "broker", "genome_launcher"], - "read_submissions:write": ["curator", "admin"], + # QC reads + "qc_reads:report": ["genome_launcher", "admin"], + "qc_reads:delete": ["admin", "superuser"], # Projects "projects:create": ["curator", "admin"], "projects:update": ["curator", "admin"], diff --git a/app/models/__init__.py b/app/models/__init__.py index ec648a3..93dad50 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -6,7 +6,8 @@ from app.models.genome_note import GenomeNote from app.models.organism import Organism from app.models.project import Project -from app.models.read import Read, ReadSubmission +from app.models.qc_read import QcRead, QcReadFile, QcReadSubmission +from app.models.read import Read from app.models.sample import Sample, SampleSubmission from app.models.token import RefreshToken from app.models.user import User diff --git a/app/models/experiment.py b/app/models/experiment.py index deca0ec..ad7ba3a 100644 --- a/app/models/experiment.py +++ b/app/models/experiment.py @@ -25,6 +25,7 @@ class Experiment(Base): UUID(as_uuid=True), ForeignKey("project.id", ondelete="CASCADE"), nullable=False ) bpa_package_id = Column(Text, unique=True, nullable=False) + base_url = Column(Text, nullable=True) design_description = Column(Text, nullable=True) bpa_library_id = Column(Text, nullable=True) library_strategy = Column(Text, nullable=True) diff --git a/app/models/qc_read.py b/app/models/qc_read.py new file mode 100644 index 0000000..87d544a --- /dev/null +++ b/app/models/qc_read.py @@ -0,0 +1,155 @@ +import uuid + +from sqlalchemy import ( + BigInteger, + CheckConstraint, + Column, + DateTime, + Float, + ForeignKey, + ForeignKeyConstraint, + Text, + func, +) +from sqlalchemy import Enum as SQLAlchemyEnum +from sqlalchemy.dialects.postgresql import JSONB, UUID +from sqlalchemy.orm import backref, relationship + +from app.db.session import Base + + +class QcRead(Base): + """Aggregate QC metrics for one QC output set linked to an experiment.""" + + __tablename__ = "qc_read" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + experiment_id = Column( + UUID(as_uuid=True), ForeignKey("experiment.id", ondelete="CASCADE"), nullable=False + ) + base_count = Column(BigInteger, nullable=False) + read_count = Column(BigInteger, nullable=False) + qc_bases_removed = Column(BigInteger, nullable=False) + qc_reads_removed = Column(BigInteger, nullable=False) + mean_gc_content = Column(Float, nullable=False) + n50_length = Column(BigInteger, nullable=True) + created_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now()) + updated_at = Column( + DateTime(timezone=True), + nullable=False, + server_default=func.now(), + onupdate=func.now(), + ) + + experiment = relationship( + "Experiment", backref=backref("qc_reads", cascade="all, delete-orphan") + ) + files = relationship("QcReadFile", back_populates="qc_read", cascade="all, delete-orphan") + submission_records = relationship( + "QcReadSubmission", back_populates="qc_read", cascade="all, delete-orphan" + ) + + +class QcReadFile(Base): + """One physical file belonging to a QcRead output set (CRAM, FASTQ R1, or FASTQ R2).""" + + __tablename__ = "qc_read_file" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + qc_read_id = Column( + UUID(as_uuid=True), ForeignKey("qc_read.id", ondelete="CASCADE"), nullable=False + ) + # 'cram', 'fastq_r1', or 'fastq_r2' + file_type = Column(Text, nullable=False) + storage_backend = Column(Text, nullable=False) + storage_profile = Column(Text, nullable=False) + bucket_name = Column(Text, nullable=False) + path_to_file = Column(Text, nullable=False) + md5_checksum = Column(Text, nullable=False) + sha256_checksum = Column(Text, nullable=False) + created_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now()) + updated_at = Column( + DateTime(timezone=True), + nullable=False, + server_default=func.now(), + onupdate=func.now(), + ) + + qc_read = relationship("QcRead", back_populates="files") + + __table_args__ = ( + CheckConstraint( + "file_type IN ('cram', 'fastq_r1', 'fastq_r2')", name="ck_qc_read_file_type" + ), + CheckConstraint("md5_checksum ~ '^[a-f0-9]{32}$'", name="ck_qc_read_file_md5"), + CheckConstraint("sha256_checksum ~ '^[a-f0-9]{64}$'", name="ck_qc_read_file_sha256"), + ) + + +class QcReadSubmission(Base): + """ENA submission record for a QcRead. Mirrors the lifecycle of ReadSubmission.""" + + __tablename__ = "qc_read_submission" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + qc_read_id = Column( + UUID(as_uuid=True), ForeignKey("qc_read.id", ondelete="CASCADE"), nullable=False + ) + experiment_id = Column( + UUID(as_uuid=True), ForeignKey("experiment.id", ondelete="CASCADE"), nullable=True + ) + authority = Column( + SQLAlchemyEnum("ENA", "NCBI", "DDBJ", name="authority_type"), nullable=False, default="ENA" + ) + status = Column( + SQLAlchemyEnum( + "draft", + "ready", + "submitting", + "accepted", + "rejected", + "replaced", + name="submission_status", + ), + nullable=False, + default="draft", + ) + prepared_payload = Column(JSONB, nullable=False) + response_payload = Column(JSONB, nullable=True) + accession = Column(Text, nullable=True) + entity_type_const = Column(Text, nullable=False, default="qc_read", server_default="qc_read") + + created_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now()) + updated_at = Column( + DateTime(timezone=True), + nullable=False, + server_default=func.now(), + onupdate=func.now(), + ) + + # Broker lease/claim fields + attempt_id = Column(UUID(as_uuid=True), nullable=True) + finalised_attempt_id = Column(UUID(as_uuid=True), nullable=True) + lock_acquired_at = Column(DateTime(timezone=True), nullable=True) + lock_expires_at = Column(DateTime(timezone=True), nullable=True) + + qc_read = relationship("QcRead", back_populates="submission_records") + experiment = relationship( + "Experiment", + backref=backref("qc_read_submission_records", cascade="all, delete-orphan"), + ) + + __table_args__ = ( + ForeignKeyConstraint( + ["accession", "authority", "entity_type_const", "qc_read_id"], + [ + "accession_registry.accession", + "accession_registry.authority", + "accession_registry.entity_type", + "accession_registry.entity_id", + ], + name="fk_qc_read_submission_accession", + deferrable=True, + initially="DEFERRED", + ), + ) diff --git a/app/models/read.py b/app/models/read.py index 0a6bb90..e296753 100644 --- a/app/models/read.py +++ b/app/models/read.py @@ -1,17 +1,14 @@ import uuid from sqlalchemy import ( - BigInteger, Boolean, Column, DateTime, ForeignKey, - ForeignKeyConstraint, Text, func, ) -from sqlalchemy import Enum as SQLAlchemyEnum -from sqlalchemy.dialects.postgresql import JSONB, UUID +from sqlalchemy.dialects.postgresql import UUID from sqlalchemy.orm import backref, relationship from app.db.session import Base @@ -50,103 +47,3 @@ class Read(Base): # Relationships experiment = relationship("Experiment", backref=backref("reads", cascade="all, delete-orphan")) - - -""" -file_name = Column(Text, nullable=False) - file_checksum = Column(Text, nullable=False) - file_format = Column(Text, nullable=False) - file_submission_date = Column(Text, nullable=True) - optional_file = Column(Text, nullable=True) - bioplatforms_url = Column(Text, nullable=True) - reads_access_date = Column(Text, nullable=True) - read_number = Column(Text, nullable=True) - lane_number = Column(Text, nullable=True) - sra_run_accession = Column(Text, nullable=True) - run_read_count = Column(Text, nullable=True) - run_base_count = Column(Text, nullable=True) -""" - - -class ReadSubmission(Base): - """ - ReadSubmission model for storing read data staged for submission to ENA. - - This model corresponds to the 'read_submission' table in the database. - """ - - __tablename__ = "read_submission" - - id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) - read_id = Column(UUID(as_uuid=True), ForeignKey("read.id", ondelete="CASCADE"), nullable=False) - authority = Column( - SQLAlchemyEnum("ENA", "NCBI", "DDBJ", name="authority_type"), nullable=False, default="ENA" - ) - status = Column( - SQLAlchemyEnum( - "draft", - "ready", - "submitting", - "accepted", - "rejected", - "replaced", - name="submission_status", - ), - nullable=False, - default="draft", - ) - - prepared_payload = Column(JSONB, nullable=False) - response_payload = Column(JSONB, nullable=True) - - # TODO determine whether these relations are needed based on query requirements - experiment_id = Column( - UUID(as_uuid=True), ForeignKey("experiment.id", ondelete="CASCADE"), nullable=True - ) - - accession = Column(Text, nullable=True) - - # Constant to help the composite FK - entity_type_const = Column(Text, nullable=False, default="read", server_default="read") - - created_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now()) - updated_at = Column( - DateTime(timezone=True), - nullable=False, - server_default=func.now(), - onupdate=func.now(), - ) - - # Relationships - read = relationship( - "Read", backref=backref("read_submission_records", cascade="all, delete-orphan") - ) - experiment = relationship( - "Experiment", backref=backref("read_exp_submission_records", cascade="all, delete-orphan") - ) - - # Broker lease/claim fields - attempt_id = Column(UUID(as_uuid=True), nullable=True) - finalised_attempt_id = Column(UUID(as_uuid=True), nullable=True) - lock_acquired_at = Column(DateTime(timezone=True), nullable=True) - lock_expires_at = Column(DateTime(timezone=True), nullable=True) - - # Table constraints - __table_args__ = ( - # Foreign key constraint for accession registry (self) - ForeignKeyConstraint( - ["accession", "authority", "entity_type_const", "read_id"], - [ - "accession_registry.accession", - "accession_registry.authority", - "accession_registry.entity_type", - "accession_registry.entity_id", - ], - name="fk_self_accession", - deferrable=True, - initially="DEFERRED", - ), - # This is a simplified version of the SQL constraint: - # UNIQUE (read_id, authority) WHERE (status = 'accepted' AND accession IS NOT NULL) - # SQLAlchemy doesn't directly support WHERE clauses in constraints, so this would need custom SQL - ) diff --git a/app/schemas/aggregate.py b/app/schemas/aggregate.py index b781f09..cf01e87 100644 --- a/app/schemas/aggregate.py +++ b/app/schemas/aggregate.py @@ -48,19 +48,16 @@ class ExperimentSubmissionJson(BaseModel): lock_expires_at: Optional[datetime] = None -class ReadSubmissionJson(BaseModel): - """Schema for read prepared_payload data with read ID""" +class QcReadSubmissionJson(BaseModel): + """Schema for QC read submission data""" - read_id: UUID - experiment_id: UUID - file_name: Optional[str] = None + qc_read_id: UUID + experiment_id: Optional[UUID] = None prepared_payload: Optional[Dict[str, Any]] = None status: Optional[str] = None accession: Optional[str] = None - experiment_accession: Optional[str] = None authority: Optional[str] = None response_payload: Optional[Dict[str, Any]] = None - submitted_at: Optional[datetime] = None created_at: Optional[datetime] = None updated_at: Optional[datetime] = None @@ -69,6 +66,8 @@ class ReadSubmissionJson(BaseModel): lock_acquired_at: Optional[datetime] = None lock_expires_at: Optional[datetime] = None + model_config = {"from_attributes": True} + class OrganismSubmissionJsonResponse(BaseModel): """Schema for returning all prepared_payload data related to an organism""" @@ -78,4 +77,4 @@ class OrganismSubmissionJsonResponse(BaseModel): scientific_name: Optional[str] = None samples: List[SampleSubmissionJson] = [] experiments: List[ExperimentSubmissionJson] = [] - reads: List[ReadSubmissionJson] = [] + reads: List[QcReadSubmissionJson] = [] diff --git a/app/schemas/qc_read.py b/app/schemas/qc_read.py new file mode 100644 index 0000000..c1868dc --- /dev/null +++ b/app/schemas/qc_read.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +import re +from datetime import datetime +from typing import Any, Dict, List, Literal, Optional +from uuid import UUID + +from pydantic import BaseModel, field_validator, model_validator + +_MD5_RE = re.compile(r"^[a-f0-9]{32}$") +_SHA256_RE = re.compile(r"^[a-f0-9]{64}$") + +FileType = Literal["cram", "fastq_r1", "fastq_r2"] + + +# --------------------------------------------------------------------------- +# QC callback payload (genome launcher → Canopy) +# --------------------------------------------------------------------------- + + +class QcFileInput(BaseModel): + file_type: FileType + storage_backend: str + storage_profile: str + bucket_name: str + path_to_file: str + md5_checksum: str + sha256_checksum: str + + @field_validator("md5_checksum") + @classmethod + def validate_md5(cls, v: str) -> str: + if not _MD5_RE.match(v): + raise ValueError("md5_checksum must be 32 lowercase hex characters") + return v + + @field_validator("sha256_checksum") + @classmethod + def validate_sha256(cls, v: str) -> str: + if not _SHA256_RE.match(v): + raise ValueError("sha256_checksum must be 64 lowercase hex characters") + return v + + +class QcCallbackRequest(BaseModel): + """Payload posted by the genome launcher after QC completes.""" + + bpa_package_id: str + base_count: int + read_count: int + qc_bases_removed: int + qc_reads_removed: int + mean_gc_content: float + n50_length: Optional[int] = None + files: List[QcFileInput] + + @model_validator(mode="after") + def validate_file_set(self) -> "QcCallbackRequest": + types = {f.file_type for f in self.files} + is_cram = types == {"cram"} and len(self.files) == 1 + is_fastq_pair = types == {"fastq_r1", "fastq_r2"} and len(self.files) == 2 + if not (is_cram or is_fastq_pair): + raise ValueError( + "files must be either exactly one CRAM file or exactly one fastq_r1 + one fastq_r2" + ) + return self + + +# --------------------------------------------------------------------------- +# Response schemas +# --------------------------------------------------------------------------- + + +class QcReadFileOut(BaseModel): + id: UUID + qc_read_id: UUID + file_type: str + storage_backend: str + storage_profile: str + bucket_name: str + path_to_file: str + md5_checksum: str + sha256_checksum: str + created_at: datetime + + model_config = {"from_attributes": True} + + +class QcReadSubmissionOut(BaseModel): + id: UUID + qc_read_id: UUID + experiment_id: Optional[UUID] + authority: str + status: str + accession: Optional[str] + created_at: datetime + updated_at: datetime + + model_config = {"from_attributes": True} + + +class QcReadOut(BaseModel): + id: UUID + experiment_id: UUID + base_count: int + read_count: int + qc_bases_removed: int + qc_reads_removed: int + mean_gc_content: float + n50_length: Optional[int] + created_at: datetime + updated_at: datetime + files: List[QcReadFileOut] = [] + submission_records: List[QcReadSubmissionOut] = [] + + model_config = {"from_attributes": True} diff --git a/app/services/assembly_helper.py b/app/services/assembly_helper.py index 66aa19e..512567a 100644 --- a/app/services/assembly_helper.py +++ b/app/services/assembly_helper.py @@ -95,6 +95,18 @@ def get_detected_platforms(experiments: List[Experiment]) -> dict: } +def _normalize_read_number(read_number: str | None) -> str | None: + """Normalize read_number to 'r1' or 'r2'.""" + if read_number is None: + return None + normalized = read_number.strip().lower().lstrip("r") + if normalized == "1": + return "r1" + if normalized == "2": + return "r2" + return None + + def generate_assembly_manifest( organism: Organism, reads: List[Read], @@ -105,11 +117,11 @@ def generate_assembly_manifest( ) -> str: """Generate assembly manifest YAML from organism and reads data. - Groups reads by platform type (PACBIO_SMRT, Hi-C) and formats as YAML. + Groups reads by bpa_package_id (from Experiment), then by platform type. Rules: - PACBIO_SMRT: Only include files ending in .ccs.bam or hifi_reads.bam - - Hi-C: Include read_number and lane_number fields + - Hi-C: Split reads into r1/r2 groups by read_number Args: organism: Organism object @@ -127,76 +139,89 @@ def generate_assembly_manifest( ) logger.info(f"Total experiments: {len(experiments)}, Total reads: {len(reads)}") - # Create experiment_id to platform/sample mapping - exp_platform_map = {} - exp_sample_map = {} + # Build experiment info map: experiment.id → metadata + exp_info_map = {} for exp in experiments: logger.info( f"Experiment {exp.id}: platform={exp.platform}, library_strategy={exp.library_strategy}" ) - if exp.platform: - exp_platform_map[exp.id] = exp.platform.upper() - if exp.library_strategy: - exp_platform_map[f"{exp.id}_strategy"] = exp.library_strategy.upper() - if getattr(exp, "sample_id", None): - exp_sample_map[str(exp.id)] = str(exp.sample_id) - - # Group reads by platform - pacbio_reads = [] - hic_reads = [] + exp_info_map[exp.id] = { + "platform": exp.platform.upper() if exp.platform else "", + "library_strategy": exp.library_strategy.upper() if exp.library_strategy else "", + "bpa_package_id": exp.bpa_package_id, + "base_url": getattr(exp, "base_url", None), + "sample_id": str(exp.sample_id) if getattr(exp, "sample_id", None) else None, + } + + # Group reads by bpa_package_id per platform + pacbio_by_package: Dict[str, Any] = {} + hic_by_package: Dict[str, Any] = {} for read in reads: if not read.experiment_id: logger.debug(f"Read {read.id} has no experiment_id, skipping") continue - platform = exp_platform_map.get(read.experiment_id, "") - library_strategy = exp_platform_map.get(f"{read.experiment_id}_strategy", "") + exp_info = exp_info_map.get(read.experiment_id) + if not exp_info: + logger.debug(f"Read {read.id} has no matching experiment, skipping") + continue + + platform = exp_info["platform"] + library_strategy = exp_info["library_strategy"] + bpa_package_id = exp_info["bpa_package_id"] + sample_id = exp_info["sample_id"] + sample_meta = (sample_metadata_by_id or {}).get(sample_id, {}) if sample_id else {} logger.debug( f"Read {read.id} (file: {read.file_name}): platform={platform}, library_strategy={library_strategy}" ) - # Check for PacBio SMRT reads (only .ccs.bam or hifi_reads.bam) + # PacBio SMRT reads (only .ccs.bam or hifi_reads.bam) if platform == "PACBIO_SMRT" and read.file_name: if read.file_name.endswith(".ccs.bam") or read.file_name.endswith("hifi_reads.bam"): - # TODO remove logging logger.info(f"Adding PacBio read: {read.file_name}") - sample_id = exp_sample_map.get(str(read.experiment_id)) - sample_meta = (sample_metadata_by_id or {}).get(sample_id, {}) if sample_id else {} - pacbio_reads.append( - { - "file_name": read.file_name, - "file_checksum": read.file_checksum, - "url": read.bioplatforms_url, + if bpa_package_id not in pacbio_by_package: + entry: Dict[str, Any] = { "sample_id": sample_id, "bpa_sample_id": sample_meta.get("bpa_sample_id"), "specimen_id": sample_meta.get("specimen_id"), + "resources": [], } + if exp_info["base_url"]: + entry["base_url"] = exp_info["base_url"] + pacbio_by_package[bpa_package_id] = entry + pacbio_by_package[bpa_package_id]["resources"].append( + {"md5sum": read.file_checksum, "url": read.bioplatforms_url} ) else: logger.debug( f"Skipping PacBio read {read.file_name} - doesn't match .ccs.bam or hifi_reads.bam" ) - # Check for Hi-C reads (Illumina + Hi-C or WGS library strategy) + # Hi-C reads (Illumina + Hi-C or WGS library strategy) elif platform == "ILLUMINA" and library_strategy in ("HI-C", "WGS"): - # TODO remove logging logger.info(f"Adding Hi-C read: {read.file_name} (library_strategy={library_strategy})") - sample_id = exp_sample_map.get(str(read.experiment_id)) - sample_meta = (sample_metadata_by_id or {}).get(sample_id, {}) if sample_id else {} - hic_reads.append( - { - "file_name": read.file_name, - "file_checksum": read.file_checksum, - "url": read.bioplatforms_url, - "read_number": read.read_number, - "lane_number": read.lane_number, + if bpa_package_id not in hic_by_package: + hic_by_package[bpa_package_id] = { "sample_id": sample_id, "bpa_sample_id": sample_meta.get("bpa_sample_id"), "specimen_id": sample_meta.get("specimen_id"), + "resources": {"r1": [], "r2": []}, } - ) + rkey = _normalize_read_number(read.read_number) + if rkey in ("r1", "r2"): + hic_by_package[bpa_package_id]["resources"][rkey].append( + { + "url": read.bioplatforms_url, + "md5sum": read.file_checksum, + "lane_number": read.lane_number, + } + ) + else: + logger.debug( + f"Hi-C read {read.file_name} has unrecognized read_number={read.read_number}, skipping" + ) else: logger.debug( f"Read {read.file_name} doesn't match criteria: platform={platform}, library_strategy={library_strategy}" @@ -211,15 +236,15 @@ def generate_assembly_manifest( "reads": {}, } - if pacbio_reads: - manifest["reads"]["PACBIO_SMRT"] = pacbio_reads - logger.info(f"Added {len(pacbio_reads)} PacBio reads to manifest") + if pacbio_by_package: + manifest["reads"]["PACBIO_SMRT"] = pacbio_by_package + logger.info(f"Added {len(pacbio_by_package)} PacBio packages to manifest") - if hic_reads: - manifest["reads"]["Hi-C"] = hic_reads - logger.info(f"Added {len(hic_reads)} Hi-C reads to manifest") + if hic_by_package: + manifest["reads"]["Hi-C"] = hic_by_package + logger.info(f"Added {len(hic_by_package)} Hi-C packages to manifest") - if not pacbio_reads and not hic_reads: + if not pacbio_by_package and not hic_by_package: logger.warning("No reads matched the filtering criteria!") # Convert to YAML diff --git a/app/services/broker_service.py b/app/services/broker_service.py index 851d32d..866c210 100644 --- a/app/services/broker_service.py +++ b/app/services/broker_service.py @@ -9,7 +9,7 @@ from app.models.broker import SubmissionAttempt from app.models.experiment import ExperimentSubmission from app.models.project import ProjectSubmission -from app.models.read import ReadSubmission +from app.models.qc_read import QcReadSubmission from app.models.sample import SampleSubmission @@ -47,6 +47,6 @@ def _expire_attempts() -> int: "project_submissions": _expire_submissions(ProjectSubmission), "sample_submissions": _expire_submissions(SampleSubmission), "experiment_submissions": _expire_submissions(ExperimentSubmission), - "read_submissions": _expire_submissions(ReadSubmission), + "qc_read_submissions": _expire_submissions(QcReadSubmission), "attempts": _expire_attempts(), } diff --git a/app/services/experiment_service.py b/app/services/experiment_service.py index 99a6d55..64d058e 100644 --- a/app/services/experiment_service.py +++ b/app/services/experiment_service.py @@ -8,7 +8,7 @@ from app.models.experiment import Experiment, ExperimentSubmission from app.models.project import Project -from app.models.read import Read, ReadSubmission +from app.models.read import Read from app.models.sample import Sample from app.schemas.bulk_import import BulkImportResponseExperiments from app.schemas.common import SubmissionStatus @@ -243,10 +243,8 @@ def bulk_import_experiments( ena_atol_map = json.load(f) experiment_mapping = ena_atol_map.get("experiment", {}) - run_mapping = ena_atol_map.get("run", {}) created_experiments_count = 0 - created_submission_count = 0 created_reads_count = 0 skipped_experiments_count = 0 skipped_reads_count = 0 @@ -313,21 +311,6 @@ def bulk_import_experiments( db.add(read) created_reads_count += 1 - run_prepared_payload: Dict[str, Any] = {} - for ena_key, atol_key in run_mapping.items(): - if atol_key in run: - run_prepared_payload[ena_key] = run[atol_key] - - read_submission = ReadSubmission( - id=uuid.uuid4(), - read_id=read.id, - experiment_id=experiment_id, - authority="ENA", - entity_type_const="read", - prepared_payload=run_prepared_payload, - ) - db.add(read_submission) - created_submission_count += 1 except Exception as e: # Try to get the most identifying information from the run run_identifier = ( @@ -466,22 +449,6 @@ def bulk_import_experiments( read = Read(**read_kwargs) db.add(read) created_reads_count += 1 - - run_prepared_payload: Dict[str, Any] = {} - for ena_key, atol_key in run_mapping.items(): - if atol_key in run: - run_prepared_payload[ena_key] = run[atol_key] - - read_submission = ReadSubmission( - id=uuid.uuid4(), - read_id=read.id, - experiment_id=experiment_id, - authority="ENA", - entity_type_const="read", - prepared_payload=run_prepared_payload, - ) - db.add(read_submission) - created_submission_count += 1 except Exception as e: # Try to get the most identifying information from the run run_identifier = ( @@ -496,7 +463,6 @@ def bulk_import_experiments( db.commit() created_experiments_count += 1 - created_submission_count += 1 except Exception as e: errors.append(f"{package_id}: {str(e)}") db.rollback() @@ -509,8 +475,7 @@ def bulk_import_experiments( skipped_reads_count=skipped_reads_count, message=( f"Experiments: {created_experiments_count} created, {skipped_experiments_count} skipped. " - f"Reads: {created_reads_count} created, {skipped_reads_count} skipped. " - f"Submission records: {created_submission_count} created." + f"Reads: {created_reads_count} created, {skipped_reads_count} skipped." ), errors=errors if errors else None, debug={ diff --git a/app/services/organism_service.py b/app/services/organism_service.py index 89ac17c..dc9b671 100644 --- a/app/services/organism_service.py +++ b/app/services/organism_service.py @@ -6,7 +6,8 @@ from app.models.experiment import Experiment, ExperimentSubmission from app.models.organism import Organism from app.models.project import Project, ProjectSubmission -from app.models.read import Read, ReadSubmission +from app.models.qc_read import QcRead, QcReadSubmission +from app.models.read import Read from app.models.sample import Sample, SampleSubmission from app.schemas.aggregate import OrganismSubmissionJsonResponse from app.schemas.bulk_import import BulkImportResponse @@ -158,14 +159,16 @@ def get_organism_prepared_payload( ) response.experiments = experiment_submission_records - reads = db.query(Read).filter(Read.experiment_id.in_(experiment_ids)).all() - read_ids = [read.id for read in reads] + qc_reads = db.query(QcRead).filter(QcRead.experiment_id.in_(experiment_ids)).all() + qc_read_ids = [qr.id for qr in qc_reads] - if read_ids: - read_submission_records = ( - db.query(ReadSubmission).filter(ReadSubmission.read_id.in_(read_ids)).all() + if qc_read_ids: + qc_read_submission_records = ( + db.query(QcReadSubmission) + .filter(QcReadSubmission.qc_read_id.in_(qc_read_ids)) + .all() ) - response.reads = read_submission_records + response.reads = qc_read_submission_records return response diff --git a/app/services/qc_read_service.py b/app/services/qc_read_service.py new file mode 100644 index 0000000..1cc564d --- /dev/null +++ b/app/services/qc_read_service.py @@ -0,0 +1,34 @@ +from typing import List, Optional +from uuid import UUID + +from sqlalchemy.orm import Session + +from app.models.experiment import Experiment +from app.models.qc_read import QcRead, QcReadSubmission + + +class QcReadSubmissionService: + """Query helpers for QcReadSubmission records.""" + + def get_by_qc_read_id(self, db: Session, qc_read_id: UUID) -> List[QcReadSubmission]: + return db.query(QcReadSubmission).filter(QcReadSubmission.qc_read_id == qc_read_id).all() + + def get_by_experiment_id(self, db: Session, experiment_id: UUID) -> List[QcReadSubmission]: + return ( + db.query(QcReadSubmission).filter(QcReadSubmission.experiment_id == experiment_id).all() + ) + + def get_by_project_id(self, db: Session, project_id: UUID) -> List[QcReadSubmission]: + return ( + db.query(QcReadSubmission) + .join(QcRead, QcRead.id == QcReadSubmission.qc_read_id) + .join(Experiment, Experiment.id == QcRead.experiment_id) + .filter(Experiment.project_id == project_id) + .all() + ) + + def get_by_accession(self, db: Session, accession: str) -> Optional[QcReadSubmission]: + return db.query(QcReadSubmission).filter(QcReadSubmission.accession == accession).first() + + +qc_read_submission_service = QcReadSubmissionService() diff --git a/app/services/read_service.py b/app/services/read_service.py index a63c8a7..ceabaf6 100644 --- a/app/services/read_service.py +++ b/app/services/read_service.py @@ -3,8 +3,7 @@ from sqlalchemy.orm import Session -from app.models.experiment import Experiment -from app.models.read import Read, ReadSubmission +from app.models.read import Read from app.schemas.read import ReadCreate, ReadUpdate from app.services.base_service import BaseService @@ -13,11 +12,9 @@ class ReadService(BaseService[Read, ReadCreate, ReadUpdate]): """Service for Read operations.""" def get_by_experiment_id(self, db: Session, experiment_id: UUID) -> List[Read]: - """Get reads by experiment ID.""" return db.query(Read).filter(Read.experiment_id == experiment_id).all() def get_by_bpa_resource_id(self, db: Session, bpa_resource_id: str) -> Optional[Read]: - """Get read by BPA resource ID.""" return db.query(Read).filter(Read.bpa_resource_id == bpa_resource_id).first() def get_multi_with_filters( @@ -29,7 +26,6 @@ def get_multi_with_filters( experiment_id: Optional[UUID] = None, bpa_resource_id: Optional[str] = None, ) -> List[Read]: - """Get reads with filters.""" query = db.query(Read) if experiment_id: query = query.filter(Read.experiment_id == experiment_id) @@ -38,30 +34,4 @@ def get_multi_with_filters( return query.offset(skip).limit(limit).all() -class ReadSubmissionService(BaseService[ReadSubmission, ReadCreate, ReadUpdate]): - """Service for ReadSubmission operations.""" - - def get_by_read_id(self, db: Session, read_id: UUID) -> List[ReadSubmission]: - """Get submission reads by read ID.""" - return db.query(ReadSubmission).filter(ReadSubmission.read_id == read_id).all() - - def get_by_experiment_id(self, db: Session, experiment_id: UUID) -> List[ReadSubmission]: - """Get submission reads by experiment ID.""" - return db.query(ReadSubmission).filter(ReadSubmission.experiment_id == experiment_id).all() - - def get_by_project_id(self, db: Session, project_id: UUID) -> List[ReadSubmission]: - """Get submission reads by project ID.""" - return ( - db.query(ReadSubmission) - .join(Experiment, Experiment.id == ReadSubmission.experiment_id) - .filter(Experiment.project_id == project_id) - .all() - ) - - def get_by_accession(self, db: Session, accession: str) -> Optional[ReadSubmission]: - """Get submission read by accession.""" - return db.query(ReadSubmission).filter(ReadSubmission.accession == accession).first() - - read_service = ReadService(Read) -read_submission_service = ReadSubmissionService(ReadSubmission) diff --git a/tests/unit/endpoints/test_broker_lease_expiration.py b/tests/unit/endpoints/test_broker_lease_expiration.py index c08d5c4..fec38fd 100644 --- a/tests/unit/endpoints/test_broker_lease_expiration.py +++ b/tests/unit/endpoints/test_broker_lease_expiration.py @@ -12,7 +12,7 @@ from app.models.experiment import ExperimentSubmission from app.models.organism import Organism from app.models.project import ProjectSubmission -from app.models.read import ReadSubmission +from app.models.qc_read import QcReadSubmission from app.models.sample import Sample, SampleSubmission @@ -226,7 +226,7 @@ def test_expire_stale_leases_resets_all_entity_types(): { SampleSubmission: [s1], ExperimentSubmission: [e1], - ReadSubmission: [r1], + QcReadSubmission: [r1], ProjectSubmission: [p1], } ) diff --git a/tests/unit/endpoints/test_endpoints_assemblies.py b/tests/unit/endpoints/test_endpoints_assemblies.py index b2e94fa..ffdb6a6 100644 --- a/tests/unit/endpoints/test_endpoints_assemblies.py +++ b/tests/unit/endpoints/test_endpoints_assemblies.py @@ -204,6 +204,8 @@ def test_get_assembly_manifest_success(monkeypatch): sample_id="550e8400-e29b-41d4-a716-446655440000", platform="PACBIO_SMRT", library_strategy="WGS", + bpa_package_id="pkg-exp-1", + base_url=None, ) read = SimpleNamespace( id="read-1", @@ -374,6 +376,8 @@ def test_create_assembly_intent_allows_empty_body(monkeypatch): sample_id=selected_sample.id, platform="PACBIO_SMRT", library_strategy="WGS", + bpa_package_id="pkg-e1", + base_url=None, ) ] @@ -423,11 +427,13 @@ def refresh(self, obj): resp = client.post("/api/v1/assemblies/intent/172942") assert resp.status_code == 200 - body = resp.json() + import yaml as _yaml + + body = _yaml.safe_load(resp.content) assert body["assembly_run_id"] == str(run_id) assert body["version"] == 1 assert body["status"] == "reserved" - assert "manifest_yaml" in body + assert "manifest" in body def test_cancel_assembly_intent_success(monkeypatch): diff --git a/tests/unit/endpoints/test_endpoints_broker.py b/tests/unit/endpoints/test_endpoints_broker.py index a4658c1..2a7e383 100644 --- a/tests/unit/endpoints/test_endpoints_broker.py +++ b/tests/unit/endpoints/test_endpoints_broker.py @@ -10,7 +10,7 @@ from app.models.experiment import ExperimentSubmission from app.models.organism import Organism from app.models.project import ProjectSubmission -from app.models.read import ReadSubmission +from app.models.qc_read import QcReadSubmission from app.models.sample import SampleSubmission @@ -99,7 +99,7 @@ def test_broker_claim_explicit_ids_empty_lists_returns_empty_response(): payload = broker.ClaimRequest( sample_submission_ids=[uuid4()], experiment_submission_ids=[uuid4()], - read_submission_ids=[uuid4()], + qc_read_submission_ids=[uuid4()], project_submission_ids=[uuid4()], lease_duration_minutes=5, ) @@ -130,7 +130,7 @@ def test_broker_renew_attempt_lease_updates_items(): SubmissionAttempt: [attempt], SampleSubmission: [s1], ExperimentSubmission: [e1], - ReadSubmission: [r1], + QcReadSubmission: [r1], ProjectSubmission: [p1], } ) @@ -164,7 +164,7 @@ def mk(kind): SubmissionAttempt: [attempt], SampleSubmission: [s1], ExperimentSubmission: [e1], - ReadSubmission: [r1], + QcReadSubmission: [r1], ProjectSubmission: [p1], } ) @@ -445,17 +445,17 @@ def test_broker_report_results_experiment_registry_inserts_and_accept(): def test_broker_report_results_read_registry_inserts_and_accept(): att_id = uuid4() sub_id = uuid4() - read_id = uuid4() + qc_read_id = uuid4() exp_id = uuid4() sub = SimpleNamespace( id=sub_id, - read_id=read_id, + qc_read_id=qc_read_id, experiment_id=exp_id, status="submitting", attempt_id=att_id, authority="ENA", ) - db = FakeSession({ReadSubmission: [sub]}) + db = FakeSession({QcReadSubmission: [sub]}) payload = broker.ReportRequest( attempt_id=att_id, samples=[], diff --git a/tests/unit/endpoints/test_endpoints_broker_contract.py b/tests/unit/endpoints/test_endpoints_broker_contract.py index 9aeadf0..8927e10 100644 --- a/tests/unit/endpoints/test_endpoints_broker_contract.py +++ b/tests/unit/endpoints/test_endpoints_broker_contract.py @@ -123,14 +123,19 @@ def test_claims_ready_returns_flat_entity_contract(monkeypatch): ) run_row = SimpleNamespace( id=uuid4(), - read_id=run_id, + qc_read_id=run_id, experiment_id=experiment_id, # FK to experiment - project_id=project_id, # FK to project status="ready", prepared_payload={ "alias": "run-1", - "file_name": "reads_1.fastq.gz", - "file_format": "fastq", + "files": [ + { + "filename": "reads_1.fastq.gz", + "filetype": "fastq", + "checksum": "abc123", + "checksum_method": "MD5", + } + ], "expected_experiment_accession": "ERX000001", # Required but not submitted }, accession=None, diff --git a/tests/unit/services/test_assembly_helper.py b/tests/unit/services/test_assembly_helper.py index 0154b57..29765b0 100644 --- a/tests/unit/services/test_assembly_helper.py +++ b/tests/unit/services/test_assembly_helper.py @@ -3,6 +3,7 @@ from unittest.mock import Mock import pytest +import yaml from app.models.experiment import Experiment from app.models.organism import Organism @@ -160,13 +161,39 @@ def test_empty_experiments(self): assert result["experiment_count"] == 0 +def _make_pacbio_experiment( + exp_id="exp1", bpa_package_id="pkg-001", sample_id="sample-uuid-1", base_url=None +): + return Mock( + id=exp_id, + platform="PACBIO_SMRT", + library_strategy="WGS", + bpa_package_id=bpa_package_id, + base_url=base_url, + sample_id=sample_id, + ) + + +def _make_hic_experiment( + exp_id="exp2", bpa_package_id="pkg-002", sample_id="sample-uuid-2", base_url=None +): + return Mock( + id=exp_id, + platform="ILLUMINA", + library_strategy="Hi-C", + bpa_package_id=bpa_package_id, + base_url=base_url, + sample_id=sample_id, + ) + + class TestGenerateAssemblyManifest: """Tests for generate_assembly_manifest function.""" def test_pacbio_reads_filtered_by_extension(self): """Test that only .ccs.bam and hifi_reads.bam files are included for PacBio.""" organism = Mock(scientific_name="Test Species", tax_id=12345) - experiments = [Mock(id="exp1", platform="PACBIO_SMRT", library_strategy="WGS")] + experiments = [_make_pacbio_experiment()] reads = [ Mock( id="r1", @@ -198,20 +225,26 @@ def test_pacbio_reads_filtered_by_extension(self): ] result = generate_assembly_manifest(organism, reads, experiments, "tol1", 1) - - assert "PACBIO_SMRT:" in result - assert "sample.ccs.bam" in result - assert "sample.hifi_reads.bam" in result - assert "sample.subreads.bam" not in result + data = yaml.safe_load(result) + + pacbio = data["reads"]["PACBIO_SMRT"] + assert "pkg-001" in pacbio + resources = pacbio["pkg-001"]["resources"] + urls = [r["url"] for r in resources] + assert "https://example.com/1" in urls + assert "https://example.com/2" in urls + assert "https://example.com/3" not in urls + assert all("md5sum" in r for r in resources) + assert all("file_name" not in r for r in resources) def test_hic_reads_include_metadata(self): - """Test that Hi-C reads include read_number and lane_number.""" + """Test that Hi-C reads include lane_number and are split by r1/r2.""" organism = Mock(scientific_name="Test Species", tax_id=12345) - experiments = [Mock(id="exp1", platform="ILLUMINA", library_strategy="Hi-C")] + experiments = [_make_hic_experiment()] reads = [ Mock( id="r1", - experiment_id="exp1", + experiment_id="exp2", file_name="hic_R1.fastq.gz", file_checksum="abc123", bioplatforms_url="https://example.com/1", @@ -221,16 +254,62 @@ def test_hic_reads_include_metadata(self): ] result = generate_assembly_manifest(organism, reads, experiments, "tol1", 1) + data = yaml.safe_load(result) + + hic = data["reads"]["Hi-C"] + assert "pkg-002" in hic + r1_resources = hic["pkg-002"]["resources"]["r1"] + assert len(r1_resources) == 1 + assert r1_resources[0]["lane_number"] == "001" + assert r1_resources[0]["md5sum"] == "abc123" + + def test_hic_reads_split_into_r1_r2(self): + """Test that Hi-C reads with read_number 1 and 2 are split into r1 and r2.""" + organism = Mock(scientific_name="Test Species", tax_id=12345) + experiments = [_make_hic_experiment()] + reads = [ + Mock( + id="r1", + experiment_id="exp2", + file_name="hic_R1.fastq.gz", + file_checksum="md5-r1", + bioplatforms_url="https://example.com/r1", + read_number="1", + lane_number="L001", + ), + Mock( + id="r2", + experiment_id="exp2", + file_name="hic_R2.fastq.gz", + file_checksum="md5-r2", + bioplatforms_url="https://example.com/r2", + read_number="2", + lane_number="L001", + ), + ] + + result = generate_assembly_manifest(organism, reads, experiments, "tol1", 1) + data = yaml.safe_load(result) - assert "Hi-C:" in result - assert "hic_R1.fastq.gz" in result - assert "read_number: '1'" in result - assert "lane_number: '001'" in result + resources = data["reads"]["Hi-C"]["pkg-002"]["resources"] + assert len(resources["r1"]) == 1 + assert len(resources["r2"]) == 1 + assert resources["r1"][0]["url"] == "https://example.com/r1" + assert resources["r2"][0]["url"] == "https://example.com/r2" def test_wgs_treated_as_hic(self): """Test that ILLUMINA + WGS is treated as Hi-C.""" organism = Mock(scientific_name="Test Species", tax_id=12345) - experiments = [Mock(id="exp1", platform="ILLUMINA", library_strategy="WGS")] + experiments = [ + Mock( + id="exp1", + platform="ILLUMINA", + library_strategy="WGS", + bpa_package_id="pkg-wgs", + base_url=None, + sample_id="s1", + ) + ] reads = [ Mock( id="r1", @@ -244,14 +323,24 @@ def test_wgs_treated_as_hic(self): ] result = generate_assembly_manifest(organism, reads, experiments, "tol1", 1) + data = yaml.safe_load(result) - assert "Hi-C:" in result - assert "sample_R1.fastq.gz" in result + assert "Hi-C" in data["reads"] + assert "pkg-wgs" in data["reads"]["Hi-C"] def test_empty_reads_dict(self): """Test that empty reads result in empty reads dict.""" organism = Mock(scientific_name="Test Species", tax_id=12345) - experiments = [Mock(id="exp1", platform="UNKNOWN", library_strategy="WGS")] + experiments = [ + Mock( + id="exp1", + platform="UNKNOWN", + library_strategy="WGS", + bpa_package_id="pkg-x", + base_url=None, + sample_id="s1", + ) + ] reads = [] result = generate_assembly_manifest(organism, reads, experiments, "tol1", 1) @@ -271,8 +360,8 @@ def test_organism_metadata_included(self): assert "tolid: tol123" in result assert "version: 2" in result - def test_sample_metadata_included_per_read(self): - """Test that related sample metadata is included on each read entry.""" + def test_sample_metadata_included_at_package_level(self): + """Test that sample metadata appears at the bpa_package_id level.""" organism = Mock(scientific_name="Saiphos equalis", tax_id=172942) experiments = [ Mock( @@ -280,6 +369,8 @@ def test_sample_metadata_included_per_read(self): sample_id="550e8400-e29b-41d4-a716-446655440000", platform="PACBIO_SMRT", library_strategy="WGS", + bpa_package_id="pkg-001", + base_url=None, ) ] reads = [ @@ -303,15 +394,61 @@ def test_sample_metadata_included_per_read(self): result = generate_assembly_manifest( organism, reads, experiments, "tol123", 2, sample_metadata_by_id ) + data = yaml.safe_load(result) - assert "sample_id: 550e8400-e29b-41d4-a716-446655440000" in result - assert "bpa_sample_id: 102.100.100/9000" in result - assert "specimen_id: SPEC-001" in result + pkg = data["reads"]["PACBIO_SMRT"]["pkg-001"] + assert pkg["sample_id"] == "550e8400-e29b-41d4-a716-446655440000" + assert pkg["bpa_sample_id"] == "102.100.100/9000" + assert pkg["specimen_id"] == "SPEC-001" + + def test_base_url_included_when_set(self): + """Test that base_url appears in the manifest when set on the experiment.""" + organism = Mock(scientific_name="Test Species", tax_id=12345) + experiments = [_make_pacbio_experiment(base_url="https://base.example.com/pkg-001")] + reads = [ + Mock( + id="r1", + experiment_id="exp1", + file_name="sample.ccs.bam", + file_checksum="abc123", + bioplatforms_url="https://example.com/1", + read_number=None, + lane_number=None, + ), + ] + + result = generate_assembly_manifest(organism, reads, experiments, "tol1", 1) + data = yaml.safe_load(result) + + pkg = data["reads"]["PACBIO_SMRT"]["pkg-001"] + assert pkg["base_url"] == "https://base.example.com/pkg-001" + + def test_base_url_omitted_when_none(self): + """Test that base_url is absent from the manifest when not set.""" + organism = Mock(scientific_name="Test Species", tax_id=12345) + experiments = [_make_pacbio_experiment(base_url=None)] + reads = [ + Mock( + id="r1", + experiment_id="exp1", + file_name="sample.ccs.bam", + file_checksum="abc123", + bioplatforms_url="https://example.com/1", + read_number=None, + lane_number=None, + ), + ] + + result = generate_assembly_manifest(organism, reads, experiments, "tol1", 1) + data = yaml.safe_load(result) + + pkg = data["reads"]["PACBIO_SMRT"]["pkg-001"] + assert "base_url" not in pkg def test_reads_without_experiment_id_skipped(self): """Test that reads without experiment_id are skipped.""" organism = Mock(scientific_name="Test Species", tax_id=12345) - experiments = [Mock(id="exp1", platform="PACBIO_SMRT", library_strategy="WGS")] + experiments = [_make_pacbio_experiment()] reads = [ Mock( id="r1", @@ -326,15 +463,14 @@ def test_reads_without_experiment_id_skipped(self): result = generate_assembly_manifest(organism, reads, experiments, "tol1", 1) - assert "sample.ccs.bam" not in result assert "reads: {}" in result def test_multiple_platform_types(self): - """Test manifest with both PacBio and Hi-C reads.""" + """Test manifest with both PacBio and Hi-C packages.""" organism = Mock(scientific_name="Test Species", tax_id=12345) experiments = [ - Mock(id="exp1", platform="PACBIO_SMRT", library_strategy="WGS"), - Mock(id="exp2", platform="ILLUMINA", library_strategy="Hi-C"), + _make_pacbio_experiment(exp_id="exp1", bpa_package_id="pkg-pacbio"), + _make_hic_experiment(exp_id="exp2", bpa_package_id="pkg-hic"), ] reads = [ Mock( @@ -358,8 +494,40 @@ def test_multiple_platform_types(self): ] result = generate_assembly_manifest(organism, reads, experiments, "tol1", 1) + data = yaml.safe_load(result) + + assert "PACBIO_SMRT" in data["reads"] + assert "Hi-C" in data["reads"] + assert "pkg-pacbio" in data["reads"]["PACBIO_SMRT"] + assert "pkg-hic" in data["reads"]["Hi-C"] + + def test_multiple_reads_same_package(self): + """Test that multiple reads under the same experiment are grouped together.""" + organism = Mock(scientific_name="Test Species", tax_id=12345) + experiments = [_make_pacbio_experiment()] + reads = [ + Mock( + id="r1", + experiment_id="exp1", + file_name="sample1.ccs.bam", + file_checksum="md5-1", + bioplatforms_url="https://example.com/1", + read_number=None, + lane_number=None, + ), + Mock( + id="r2", + experiment_id="exp1", + file_name="sample2.ccs.bam", + file_checksum="md5-2", + bioplatforms_url="https://example.com/2", + read_number=None, + lane_number=None, + ), + ] + + result = generate_assembly_manifest(organism, reads, experiments, "tol1", 1) + data = yaml.safe_load(result) - assert "PACBIO_SMRT:" in result - assert "Hi-C:" in result - assert "sample.ccs.bam" in result - assert "hic_R1.fastq.gz" in result + resources = data["reads"]["PACBIO_SMRT"]["pkg-001"]["resources"] + assert len(resources) == 2