diff --git a/flowbio/cli/_accession_sheet.py b/flowbio/cli/_accession_sheet.py new file mode 100644 index 0000000..866f484 --- /dev/null +++ b/flowbio/cli/_accession_sheet.py @@ -0,0 +1,252 @@ +"""CSV accession-sheet parsing for ``samples import``. + +An accession sheet is a CSV with one row per accession to import: required +``accession`` and ``sample_type`` columns, plus optional ``name``/ +``organism`` and per-accession metadata columns. This mirrors ``_sheet.py``'s +reads-based sample sheet, but the reserved columns differ — there is nothing +to upload (no ``reads1``/``reads2``) and the import API has no project field. + +Domain rules (accession format, duplicates, sample type, organism, metadata) +are all checked server-side when the sheet is submitted — duplicating that +locally would just be a second, driftable copy of the same rules. What *is* +checked locally is structural: every row must have an accession and a +sample type to mean anything at all, every header column must have a +unique, non-empty name, and every row must have no fewer cells than the +header, and no more unless the extra ones are blank — any of these is a +case the parser can't resolve on the user's behalf, so it's rejected +rather than guessed at. See :func:`parse_accession_sheet`. +""" +from __future__ import annotations + +import csv +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import Path + +from flowbio.cli._exit_codes import CliUsageError +from flowbio.cli._files import existing_file +from flowbio.v2.samples import SampleImportSpec, SampleTypeId + +RESERVED_COLUMNS = ("accession", "name", "organism", "sample_type") + +ParsedRow = Mapping[str | None, str | list[str] | None] +"""One row as ``csv.DictReader`` yields it: a header's cell, or ``None`` if +the row was too short to reach it; the extra cells of a too-long row, as a +``list[str]``, under the key ``None``.""" + + +@dataclass(frozen=True) +class AccessionSheetRow: + """One data row of an accession sheet.""" + + row_number: int + accession: str + name: str | None + organism: str | None + sample_type: SampleTypeId + metadata: dict[str, str] + + def __post_init__(self) -> None: + if not self.accession: + raise ValueError("accession must not be empty") + if not self.sample_type: + raise ValueError("sample_type must not be empty") + + def to_spec(self) -> SampleImportSpec: + """Build the :class:`~flowbio.v2.samples.SampleImportSpec` for this row.""" + return SampleImportSpec( + accession=self.accession, + sample_type=self.sample_type, + name=self.name, + organism_id=self.organism, + metadata=self.metadata or None, + ) + + +@dataclass(frozen=True) +class AccessionSheet: + """A parsed accession sheet.""" + + path: Path + rows: list[AccessionSheetRow] + + +def parse_accession_sheet(path: Path) -> AccessionSheet: + """Parse a CSV accession sheet into an :class:`AccessionSheet`. + + :param path: The accession-sheet file. Must be a ``.csv`` — an ``.xlsx`` or + ``.tsv`` is a usage error directing the user to export to CSV. + :returns: The parsed sheet, with wholly-blank rows skipped, empty cells + dropped, and surrounding whitespace trimmed (including in header + names). Values are otherwise passed through unchanged, including + ``accession``, sent to the server as-entered. + :raises CliUsageError: If the file is not a readable ``.csv``, has no + header row, has an unnamed or duplicated column, has no rows, has + a row with fewer cells than the header or with more cells than the + header where the overflow isn't blank, or has a row with no + accession or no sample_type. + """ + if path.suffix.lower() != ".csv": + raise CliUsageError( + f"Accession sheet must be a .csv file: {path}. " + f"Export your spreadsheet to CSV first.", + ) + existing_file(path) + rows: list[AccessionSheetRow] = [] + short_rows: list[int] = [] + overflow_rows: list[int] = [] + missing_accession: list[int] = [] + missing_sample_type: list[int] = [] + # utf-8-sig transparently strips a leading BOM, which spreadsheet tools + # (notably Excel's "CSV UTF-8" export) prepend — otherwise the first header + # parses as "accession" and every row reports a missing accession. + with path.open(newline="", encoding="utf-8-sig") as handle: + reader = csv.DictReader(handle) + headers, metadata_columns = _read_header(reader) + for row_number, record in enumerate(reader, start=1): + overflow_has_value = any(cell.strip() for cell in _overflow_cells(record)) + if not overflow_has_value and _is_blank_row(record, headers): + continue + # A short row's missing cells could just as easily be trailing + # optional columns as data landing in the wrong place, so — + # unlike a too-wide row's blank overflow — it's rejected outright. + if any(record.get(header) is None for header in headers): + short_rows.append(row_number) + continue + if overflow_has_value: + overflow_rows.append(row_number) + continue + accession = _cell(record, "accession") + sample_type = _cell(record, "sample_type") + if accession is None: + missing_accession.append(row_number) + if sample_type is None: + missing_sample_type.append(row_number) + if accession is None or sample_type is None: + continue + rows.append(_build_row(record, row_number, metadata_columns, accession, sample_type)) + _check_rows(rows, short_rows, overflow_rows, missing_accession, missing_sample_type) + return AccessionSheet(path=path, rows=rows) + + +def _read_header(reader: csv.DictReader[str]) -> tuple[list[str], list[str]]: + """Validate the sheet's header row and return its columns and metadata columns.""" + # A blank first line parses as fieldnames == [] — distinct from an + # absent one, which is fieldnames is None. + if reader.fieldnames == []: + raise CliUsageError("Accession sheet has no header row.") + headers = [header.strip() for header in reader.fieldnames or []] + reader.fieldnames = headers + _check_headers(headers) + metadata_columns = [header for header in headers if header not in RESERVED_COLUMNS] + return headers, metadata_columns + + +def _check_rows( + rows: list[AccessionSheetRow], + short_rows: list[int], + overflow_rows: list[int], + missing_accession: list[int], + missing_sample_type: list[int], +) -> None: + """Raise if the sheet had no rows at all, or any row failed a structural check.""" + if not (rows or short_rows or overflow_rows or missing_accession or missing_sample_type): + raise CliUsageError("Accession sheet has no rows.") + if not (short_rows or overflow_rows or missing_accession or missing_sample_type): + return + clauses = [ + clause for clause in ( + _row_reason_clause("fewer cells than the header", short_rows), + _row_reason_clause("more cells than the header", overflow_rows), + _row_reason_clause("no accession", missing_accession), + _row_reason_clause("no sample_type", missing_sample_type), + ) if clause is not None + ] + message = f"Accession sheet {'; '.join(clauses)}." + remedies: list[str] = [] + if short_rows: + remedies.append( + "add the missing trailing comma(s), leaving the cell(s) blank if that " + "column doesn't apply to this row, or fill in a value", + ) + if overflow_rows: + remedies.append( + "if a value legitimately contains a comma, quote it; otherwise " + "remove the extra cell(s)", + ) + message += "".join(f" {remedy[0].upper()}{remedy[1:]}." for remedy in remedies) + raise CliUsageError(message) + + +def _check_headers(headers: list[str]) -> None: + unnamed = [position for position, header in enumerate(headers, start=1) if not header] + duplicates = sorted({header for header in headers if header and headers.count(header) > 1}) + if not unnamed and not duplicates: + return + clauses: list[str] = [] + remedies: list[str] = [] + if unnamed: + clauses.append(_unnamed_columns_clause(unnamed)) + remedies.append("remove the trailing comma(s) from the header row, or give the column(s) a name") + if duplicates: + clauses.append(_duplicate_columns_clause(duplicates)) + remedies.append("rename the repeated column(s) so each column is unique") + remedy = ". ".join(f"{r[0].upper()}{r[1:]}" for r in remedies) + raise CliUsageError(f"Accession sheet {'; '.join(clauses)}. {remedy}.") + + +def _unnamed_columns_clause(unnamed: list[int]) -> str: + positions = ", ".join(str(position) for position in unnamed) + verb = "is" if len(unnamed) == 1 else "are" + return f"column(s) {positions} {verb} unnamed" + + +def _duplicate_columns_clause(duplicates: list[str]) -> str: + names = ", ".join(f"'{name}'" for name in duplicates) + verb = "is" if len(duplicates) == 1 else "are" + return f"column name(s) {names} {verb} duplicated" + + +def _is_blank_row(record: ParsedRow, headers: list[str]) -> bool: + return not any(_cell(record, header) for header in headers) + + +def _overflow_cells(record: ParsedRow) -> list[str]: + overflow = record.get(None) + return overflow if isinstance(overflow, list) else [] + + +def _row_reason_clause(reason: str, rows: list[int]) -> str | None: + if not rows: + return None + numbers = ", ".join(str(number) for number in rows) + verb = "has" if len(rows) == 1 else "have" + return f"data row(s) {numbers} {verb} {reason}" + + +def _cell(record: ParsedRow, column: str) -> str | None: + raw = record.get(column) + value = (raw if isinstance(raw, str) else "").strip() + return value or None + + +def _build_row( + record: ParsedRow, + row_number: int, + metadata_columns: list[str], + accession: str, + sample_type: str, +) -> AccessionSheetRow: + metadata = { + column: value + for column in metadata_columns + if (value := _cell(record, column)) is not None + } + return AccessionSheetRow( + row_number=row_number, + accession=accession, + name=_cell(record, "name"), + organism=_cell(record, "organism"), + sample_type=SampleTypeId(sample_type), + metadata=metadata, + ) diff --git a/flowbio/cli/_samples.py b/flowbio/cli/_samples.py index abd1896..e681d32 100644 --- a/flowbio/cli/_samples.py +++ b/flowbio/cli/_samples.py @@ -11,9 +11,11 @@ import argparse import json from dataclasses import dataclass +from datetime import datetime from pathlib import Path from typing import Literal +from flowbio.cli._accession_sheet import parse_accession_sheet from flowbio.cli._exit_codes import CliUsageError, ExitCode from flowbio.cli._files import existing_file from flowbio.cli._output import Output, format_issue @@ -26,7 +28,12 @@ from flowbio.cli._types import JsonValue from flowbio.v2.client import Client from flowbio.v2.exceptions import FlowApiError -from flowbio.v2.samples import MetadataAttribute, SampleTypeId +from flowbio.v2.samples import ( + MetadataAttribute, + SampleImportJob, + SampleImportJobId, + SampleTypeId, +) def register( @@ -76,6 +83,22 @@ def register( "valid rows sequentially, reporting each row's outcome." ), )) + _configure_import(verbs.add_parser( + "import", + parents=[global_parent], + help="Import samples from public-repository accessions.", + description=( + "Kick off a batch import job from a CSV accession sheet and report " + "its id. Does not wait for the job to finish — check its progress " + "with 'samples import-status'." + ), + )) + _configure_import_status(verbs.add_parser( + "import-status", + parents=[global_parent], + help="Check the status of an import job.", + description="Fetch and report the current state of a 'samples import' job.", + )) def _configure_upload(upload: argparse.ArgumentParser) -> None: @@ -233,6 +256,38 @@ def _configure_upload_batch(upload_batch: argparse.ArgumentParser) -> None: ) +def _configure_import(import_parser: argparse.ArgumentParser) -> None: + import_parser.set_defaults(command_parser=import_parser, handler=_import_command) + import_parser.add_argument( + "--sheet", + required=True, + metavar="PATH", + type=Path, + help=( + "CSV accession sheet (required accession/sample_type columns, " + "optional name/organism, plus metadata columns)." + ), + ) + + +def _job_id(value: str) -> SampleImportJobId: + try: + return SampleImportJobId(int(value)) + except ValueError: + raise argparse.ArgumentTypeError(f"job id must be an integer, got {value!r}") from None + + +def _configure_import_status(import_status: argparse.ArgumentParser) -> None: + import_status.set_defaults(command_parser=import_status, handler=_import_status_command) + import_status.add_argument( + "--job-id", + required=True, + metavar="ID", + type=_job_id, + help="Import job id, as reported by 'samples import'.", + ) + + def _upload_command(args: argparse.Namespace, client: Client, output: Output) -> ExitCode: """Upload a single sample and report its identifier. @@ -558,6 +613,75 @@ def _invalid_line(row: SheetRow, reasons: list[str]) -> str: return f"Row {row.row_number} ({row.name}): {'; '.join(reasons)}" +def _import_command( + args: argparse.Namespace, client: Client, output: Output, +) -> ExitCode: + """Kick off a batch import job from an accession sheet and report its id. + + Every row is submitted as-is: the accession format, sample type, + organism, and metadata rules are all validated server-side, so a + malformed sheet surfaces as a normal :class:`FlowApiError` rather than a + local pre-flight rejection. This command does not wait for the job to + finish — poll it yourself with ``samples import-status``. + + :param args: Parsed command-line arguments. + :param client: The authenticated Flow client. + :param output: The result/error renderer. + :returns: :attr:`ExitCode.SUCCESS` once the job has been kicked off. + :raises CliUsageError: If the accession sheet is structurally invalid + (see :func:`~flowbio.cli._accession_sheet.parse_accession_sheet`). + """ + sheet = parse_accession_sheet(args.sheet) + specs = [row.to_spec() for row in sheet.rows] + job = client.samples.import_samples(specs) + output.emit_result( + f"Started import job {job.id} for {len(specs)} accession(s) " + f"(status: {job.status}). Check progress with " + f"'flowbio samples import-status --job-id {job.id}'.", + job.model_dump(mode="json"), + ) + return ExitCode.SUCCESS + + +def _import_status_command( + args: argparse.Namespace, client: Client, output: Output, +) -> ExitCode: + """Fetch and report the current state of an import job. + + :param args: Parsed command-line arguments. + :param client: The authenticated Flow client. + :param output: The result/error renderer. + :returns: :attr:`ExitCode.SUCCESS` once the job's state has been fetched, + unless the job itself is ``"FAILED"``, in which case + :attr:`ExitCode.RUNTIME` — so a caller polling this command can + branch on its exit code alone, without parsing ``--json`` output. + """ + job = client.samples.get_import(args.job_id) + if job.status == "FAILED": + output.emit_advisory(f"Job {job.id} failed: {job.error or 'no error message returned'}") + output.emit_result(_job_summary(job), job.model_dump(mode="json")) + return ExitCode.RUNTIME if job.status == "FAILED" else ExitCode.SUCCESS + + +def _job_summary(job: SampleImportJob) -> str: + if job.status == "COMPLETED": + ids = ", ".join(str(sample_id) for sample_id in job.sample_ids) or "none" + suffix = _timestamp_suffix("finished", job.finished) + return f"Job {job.id}: COMPLETED{suffix}. Sample ids: {ids}." + if job.status == "FAILED": + # A human sees the error as a separate advisory on stderr, so it + # isn't repeated here. + return f"Job {job.id}: FAILED{_timestamp_suffix('finished', job.finished)}." + label = "started" if job.started else "created" + return f"Job {job.id}: {job.status}{_timestamp_suffix(label, job.started or job.created)}." + + +def _timestamp_suffix(label: str, timestamp: datetime | None) -> str: + if timestamp is None: + return "" + return f" ({label} {timestamp.strftime('%Y-%m-%d %H:%M:%S UTC')})" + + def _merge_metadata( pairs: list[str] | None, json_text: str | None, ) -> dict[str, str]: diff --git a/flowbio/v2/__init__.py b/flowbio/v2/__init__.py index 1b239ee..d0d5f1f 100644 --- a/flowbio/v2/__init__.py +++ b/flowbio/v2/__init__.py @@ -45,6 +45,10 @@ Organism, Project, Sample, + SampleImportJob, + SampleImportJobId, + SampleImportSpec, + SampleImportStatus, SampleType, SampleTypeId, ) @@ -59,6 +63,10 @@ "Organism", "Project", "Sample", + "SampleImportJob", + "SampleImportJobId", + "SampleImportSpec", + "SampleImportStatus", "SampleType", "SampleTypeId", "TokenCredentials", diff --git a/flowbio/v2/samples.py b/flowbio/v2/samples.py index fd888ed..e18923e 100644 --- a/flowbio/v2/samples.py +++ b/flowbio/v2/samples.py @@ -27,10 +27,12 @@ from __future__ import annotations from collections.abc import Sequence +from dataclasses import asdict, dataclass +from datetime import datetime, timezone from pathlib import Path -from typing import TYPE_CHECKING, NewType +from typing import TYPE_CHECKING, Literal, NewType -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, field_validator from flowbio.v2._pagination import PageIterator from flowbio.v2.exceptions import ( @@ -139,6 +141,94 @@ class MultiplexedUpload(BaseModel, frozen=True): ) +SampleImportJobId = NewType("SampleImportJobId", int) +"""The identifier of a sample-import job, as returned by +:meth:`SampleResource.import_samples`.""" + + +SampleImportStatus = Literal["RUNNING", "COMPLETED", "FAILED"] +"""The lifecycle state of a :class:`SampleImportJob`.""" + + +@dataclass(frozen=True) +class SampleImportSpec: + """One accession to import, with its per-accession identity and metadata. + + Example:: + + specs = [ + SampleImportSpec(accession="ERR1160845", sample_type="RNA-Seq"), + SampleImportSpec( + accession="ERR10677146", + sample_type="RNA-Seq", + organism_id="Hs", + metadata={"strandedness": "reverse"}, + ), + ] + + :param accession: The public-repository run or experiment accession + (e.g. ``"ERR1160845"``), validated server-side. + :param sample_type: The sample type identifier (e.g. ``"RNA-Seq"``), + validated server-side. See :meth:`SampleResource.get_types`. + :param name: Optional sample name. Defaults to ``accession`` server-side + when omitted. + :param organism_id: Optional organism id (e.g. ``"Hs"``) to associate + with the sample, sent as ``organism``. + :param metadata: Optional metadata key-value pairs. See + :ref:`metadata-attributes` for details on required attributes. + """ + + accession: str + sample_type: SampleTypeId + name: str | None = None + organism_id: str | None = None + metadata: dict[str, str] | None = None + + +class SampleImportJob(BaseModel, frozen=True): + """A batch job that imports one or more accessions into samples. + + All accessions submitted in one :meth:`SampleResource.import_samples` call + share a single job: ``status`` and ``error`` describe the whole batch, and + ``accessions``/``sample_ids`` correspond positionally once ``status`` is + ``"COMPLETED"``. + """ + + id: SampleImportJobId = Field(description="Unique identifier for this import job.") + status: SampleImportStatus = Field(description="The job's current lifecycle state.") + created: datetime | None = Field(default=None, description="When the job was created.") + started: datetime | None = Field(default=None, description="When the job started running, if it has.") + finished: datetime | None = Field( + default=None, description="When the job finished (completed or failed), if it has.", + ) + accessions: list[str] = Field( + default_factory=list, description="The accessions submitted with this job, in submission order.", + ) + sample_ids: list[int] = Field( + default_factory=list, + description="The created samples' ids, corresponding to ``accessions`` once the job has completed.", + ) + execution_id: int | None = Field( + default=None, description="The pipeline execution backing this job, if one was created.", + ) + error: str | None = Field( + default=None, description='The failure reason, set only when status is "FAILED".', + ) + + @field_validator("created", "started", "finished", mode="after") + @classmethod + def _normalize_to_utc(cls, value: datetime | None) -> datetime | None: + if value is None: + return None + # A naive value (no tzinfo) is treated as already UTC, matching what + # the server always means by these timestamps, rather than left + # ambiguous for every consumer (CLI, --json, library) to decide on + # its own. + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + class SampleResource: """Provides access to sample-related API endpoints. @@ -399,6 +489,56 @@ def get_metadata_attributes(self) -> list[MetadataAttribute]: """ return [self._create_metadata_attribute(item) for item in (self._transport.get("/samples/metadata"))] + def import_samples(self, imports: Sequence[SampleImportSpec]) -> SampleImportJob: + """Kick off a batch import of samples from public-repository accessions. + + Every accession is submitted together and tracked as a single job — poll + it with :meth:`get_import` until its status leaves ``"RUNNING"``. + + Requires authentication. + + Example:: + + job = client.samples.import_samples([ + SampleImportSpec(accession="ERR1160845", sample_type="RNA-Seq"), + ]) + print(f"Import job: {job.id}") + + :param imports: The accessions to import, one :class:`SampleImportSpec` each. + :raises FlowApiError: If any entry is invalid, e.g. an unsupported + accession format, unknown sample type, or missing required metadata. + """ + payload = {"imports": [self._import_spec_fields(spec) for spec in imports]} + return SampleImportJob(**self._transport.post("/v2/sample-imports", json=payload)) + + def get_import(self, job_id: SampleImportJobId) -> SampleImportJob: + """Fetch the current state of an import job. + + Example:: + + job = client.samples.get_import(job.id) + if job.status == "COMPLETED": + print(f"Imported samples: {job.sample_ids}") + + :param job_id: The job id returned by :meth:`import_samples`. + :raises NotFoundError: If no import job with that id exists. + """ + return SampleImportJob(**self._transport.get(f"/v2/sample-imports/{job_id}")) + + @staticmethod + def _import_spec_fields(spec: SampleImportSpec) -> dict[str, str | dict[str, str]]: + """Build the wire payload for one accession. + + Every field is sent under its dataclass name as-is; only ``name``, + ``organism`` (renamed from ``organism_id``), and ``metadata`` are + omitted when empty. A field added to :class:`SampleImportSpec` is + sent even when unset unless it's also added to ``optional`` here. + """ + fields = asdict(spec) + fields["organism"] = fields.pop("organism_id") + optional = ("name", "organism", "metadata") + return {key: value for key, value in fields.items() if key not in optional or value} + def _create_metadata_attribute(self, item: dict) -> MetadataAttribute: item["required_for_sample_types"] = [ SampleTypeId(link["sample_type_identifier"]) diff --git a/setup.py b/setup.py index 0316e01..28962a7 100644 --- a/setup.py +++ b/setup.py @@ -5,7 +5,7 @@ setup( name="flowbio", - version="0.9.0", + version="0.10.0", description="A client for the Flow API.", long_description=long_description, long_description_content_type="text/markdown", diff --git a/source/cli.rst b/source/cli.rst index 029c86a..7d7911b 100644 --- a/source/cli.rst +++ b/source/cli.rst @@ -367,6 +367,157 @@ authentication failure; otherwise the standard mapping above. $ flowbio samples upload-batch --sheet ./samples.csv --sample-type RNA-Seq --json {"uploaded": [{"row_number": 1, "name": "liver_r1", "sample_id": "samp_1"}], "failed": [], "skipped": [], "counts": {"uploaded": 1, "failed": 0, "skipped": 0}} +``samples import`` +~~~~~~~~~~~~~~~~~~ + +Kick off a batch import of samples from public-repository accessions (SRR/ +ERR/DRR run or SRX/ERX/DRX experiment accessions) — no files to upload +yourself. + +:: + + flowbio samples import --sheet PATH + +Run ``flowbio samples import --help`` for the full option list. The sheet is +a CSV with required ``accession``/``sample_type`` columns, plus optional +``name``/``organism`` and metadata columns (there is no ``batch-template`` +equivalent for it, since it has no reads files or project field). ``name`` +defaults to the accession when omitted. There is deliberately no +``--sample-type`` flag: the sheet's own column is the only way to supply a +sample type, so a mixed-type sheet needs no special handling and a +single-type sheet just repeats the same value down the column. + +Every value is sent as-is, with surrounding whitespace trimmed; header +names are trimmed the same way. The accession format, sample type, and +metadata rules are all validated **server-side**. This command only +checks what's structural — anything it can't resolve on your behalf, it +rejects up front rather than guessing: + +- The sheet must be a readable ``.csv``, with at least one data row. +- Every header column must have a name, and no two columns may share one — + a trailing comma in the header row (or a copy-pasted column) is rejected + rather than becoming an empty-named metadata attribute, or one column + silently overwriting another. +- A data row with data in it must have as many cells as the header — a + *short* row (e.g. trailing optional columns omitted entirely, rather + than left as empty cells) is rejected, not treated as having blank + values. A row with *more* cells than the header is rejected only if the + overflow carries a value (e.g. an unquoted comma inside a value shifting + a real value into the extra cell); a purely blank overflow (e.g. one + stray trailing comma) is ignored. +- Every row must have an accession and a sample type; a blank cell in + either column rejects the whole sheet. + +A row with every cell blank — whatever its width, including one with fewer +cells than the header (e.g. a trailing comma-only line some spreadsheet +exports leave below the data, however many commas it happens to have) — is +skipped rather than treated as a row missing values. Rows are counted from +``1`` for the first data row, after the header (the same convention as +``upload-batch``'s ``row_number``) — except a line with no commas at all +(as opposed to one with only blank cells), which doesn't consume a number. + +Unlike ``upload-batch``, a metadata column named ``__annotation`` +is **not** given any special handling here — it is forwarded as an ordinary +metadata key, which the server does not recognise, so it is silently +ignored rather than attached as an annotation or rejected. + +Every row is submitted **together as one server-side job**. This command +does **not wait for it to finish** — it reports the job's id and initial +status (almost always ``"RUNNING"``) and returns immediately. Check on it +with ``samples import-status --job-id ID``; polling (if you want it) is up +to you, e.g. in a shell loop. Building the same thing directly against the +library instead of the CLI? See :ref:`sample-imports`. + +**Output** — human: a confirmation line with the job id and a pointer to +``import-status``. ``--json``: the created job as a single document — +``id``, ``status``, ``created``/``started``/``finished`` (ISO 8601 +timestamps, ``null`` if not yet reached), ``accessions``, ``sample_ids`` +(empty until the job completes), ``execution_id``, ``error``. + +**Exit codes** — ``0`` the job was created (regardless of its eventual +outcome — check that with ``import-status``); ``2`` the sheet is +structurally invalid (see the checks above), including not being a +readable ``.csv`` or having no rows; ``1`` the API rejected the batch +(e.g. unknown sample type, missing required metadata, an unsupported +accession format — these come back as an HTTP ``422``; ``5`` in the +unlikely case it answers ``400`` instead); ``3`` authentication failure; +otherwise the standard mapping above. + +**Example** + +``accessions.csv``: + +.. code-block:: text + + accession,sample_type,name,organism + ERR1160845,RNA-Seq,liver_r1,Hs + ERR10677146,RNA-Seq,, + +.. code-block:: bash + + $ flowbio samples import --sheet ./accessions.csv + Started import job 42 for 2 accession(s) (status: RUNNING). Check progress with 'flowbio samples import-status --job-id 42'. + + $ flowbio samples import --sheet ./accessions.csv --json + {"id": 42, "status": "RUNNING", "created": "2024-04-05T19:34:38Z", "started": null, "finished": null, "accessions": ["ERR1160845", "ERR10677146"], "sample_ids": [], "execution_id": null, "error": null} + +``samples import-status`` +~~~~~~~~~~~~~~~~~~~~~~~~~ + +Fetch and report the current state of a ``samples import`` job. + +:: + + flowbio samples import-status --job-id ID + +Read-only — checking a job's status never changes it. There is no built-in +polling; run this again (or wrap it in your own loop) until ``status`` leaves +``"RUNNING"``. Gate the loop on ``flowbio``'s own exit code (assigning the +document, not a value piped through ``jq``, in the ``while`` condition) so a +transient failure (auth, network) breaks the loop instead of being read as +``"RUNNING"``: + +.. code-block:: bash + + while out=$(flowbio samples import-status --job-id 42 --json); do + [ "$(printf '%s' "$out" | jq -r .status)" = RUNNING ] || break + sleep 30 + done + printf '%s' "$out" | jq -r .status # COMPLETED / FAILED; empty if the command errored + +**Output** — human: a one-line summary including the sample ids and when it +finished on ``"COMPLETED"``, when it started — or, if it hasn't yet, when it +was created — on ``"RUNNING"``, or — on ``"FAILED"`` — when it finished plus +the error as a separate advisory on stderr (so it isn't printed twice). +``--json``: the job as a single document — ``id``, ``status``, +``created``/``started``/``finished`` (ISO 8601 timestamps, useful for judging +how long a job has been running when you've resumed polling one from +elsewhere), ``accessions``, ``sample_ids``, ``execution_id``, ``error``. +``--json`` never prints prose to stderr (or anywhere but that one stdout +document); the failure reason on a ``"FAILED"`` job is the document's +``error`` field, not a separate message. + +**Exit codes** — ``0`` the job was fetched and is ``"RUNNING"`` or +``"COMPLETED"``; ``1`` either the job was fetched but is ``"FAILED"``, or the +request itself failed (e.g. a transient server error) — human mode +distinguishes them (an ``Error:`` line means the request failed; a +``Job N: FAILED.`` line means the job did), and under ``--json`` a request +failure's document is on stderr while a ``FAILED`` job's is on stdout like +any other successful fetch. The loop above still needs ``--json``/``jq`` to +tell ``"RUNNING"`` from ``"COMPLETED"``, since both exit ``0``; ``4`` no job +with that id exists; ``3`` authentication failure; otherwise the standard +mapping above. + +**Example** + +.. code-block:: bash + + $ flowbio samples import-status --job-id 42 + Job 42: COMPLETED (finished 2024-04-05 19:38:20 UTC). Sample ids: 101, 102. + + $ flowbio samples import-status --job-id 42 --json + {"id": 42, "status": "COMPLETED", "created": "2024-04-05T19:34:38Z", "started": "2024-04-05T19:34:40Z", "finished": "2024-04-05T19:38:20Z", "accessions": ["ERR1160845", "ERR10677146"], "sample_ids": [101, 102], "execution_id": 7, "error": null} + ``api get`` ~~~~~~~~~~~ diff --git a/source/v2/samples.rst b/source/v2/samples.rst index 6fb5dd2..3ebeafd 100644 --- a/source/v2/samples.rst +++ b/source/v2/samples.rst @@ -166,6 +166,48 @@ To reject the upload when warnings are present, set This raises :class:`~flowbio.v2.exceptions.AnnotationValidationError` if the annotation has any warnings. +.. _sample-imports: + +Importing samples from public repositories +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Use :meth:`~flowbio.v2.samples.SampleResource.import_samples` to create +samples directly from public-repository run or experiment accessions +(SRR/ERR/DRR or SRX/ERX/DRX), instead of uploading files yourself. Every +accession submitted together is tracked as a single job:: + + from flowbio.v2 import SampleImportSpec + + job = client.samples.import_samples([ + SampleImportSpec(accession="ERR1160845", sample_type="RNA-Seq"), + SampleImportSpec( + accession="ERR10677146", + sample_type="RNA-Seq", + metadata={"strandedness": "reverse"}, + ), + ]) + +The job starts out ``"RUNNING"``. Poll it with +:meth:`~flowbio.v2.samples.SampleResource.get_import` until its status +leaves ``"RUNNING"`` — bound the wait so a stuck job doesn't loop forever:: + + import time + + deadline = time.monotonic() + 1800 # 30 minutes + while job.status == "RUNNING" and time.monotonic() < deadline: + time.sleep(5) + job = client.samples.get_import(job.id) + + if job.status == "COMPLETED": + print(f"Imported samples: {job.sample_ids}") + elif job.status == "FAILED": + print(f"Import failed: {job.error}") + else: + print(f"Import did not finish within the deadline (status: {job.status})") + +``job.accessions`` and ``job.sample_ids`` correspond positionally once the +job has completed. + API Reference ------------- @@ -185,4 +227,13 @@ Models .. autopydantic_model:: flowbio.v2.samples.Organism -.. autopydantic_model:: flowbio.v2.samples.MultiplexedUpload \ No newline at end of file +.. autopydantic_model:: flowbio.v2.samples.MultiplexedUpload + +.. autoclass:: flowbio.v2.samples.SampleImportSpec + :members: + +.. autopydantic_model:: flowbio.v2.samples.SampleImportJob + +.. autodata:: flowbio.v2.samples.SampleImportJobId + +.. autodata:: flowbio.v2.samples.SampleImportStatus diff --git a/tests/unit/cli/test_accession_sheet.py b/tests/unit/cli/test_accession_sheet.py new file mode 100644 index 0000000..51ac7a4 --- /dev/null +++ b/tests/unit/cli/test_accession_sheet.py @@ -0,0 +1,441 @@ +import csv +from pathlib import Path + +import pytest + +from flowbio.cli._accession_sheet import AccessionSheetRow, parse_accession_sheet +from flowbio.cli._exit_codes import CliUsageError +from flowbio.v2.samples import SampleImportSpec + +HEADERS = ["accession", "name", "organism", "sample_type", "cell_type", "source", "source__annotation"] + + +def _write_sheet( + directory: Path, *records: dict[str, str], headers: list[str] = HEADERS, +) -> Path: + path = directory / "sheet.csv" + with path.open("w", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=headers) + writer.writeheader() + writer.writerows(records) + return path + + +def _record(**overrides: str) -> dict[str, str]: + record = {"accession": "ERR1160845", "sample_type": "rna_seq"} + record.update(overrides) + return record + + +class TestParseAccessionSheet: + + def test_accession_is_passed_through_unchanged(self, tmp_path: Path) -> None: + sheet = parse_accession_sheet( + _write_sheet(tmp_path, _record(accession="err1160845")), + ) + + assert sheet.rows[0].accession == "err1160845" + + def test_empty_cells_omitted_from_metadata(self, tmp_path: Path) -> None: + sheet = parse_accession_sheet(_write_sheet( + tmp_path, + _record(cell_type="", source="blood"), + )) + + assert sheet.rows[0].metadata == {"source": "blood"} + + def test_name_and_organism_are_optional(self, tmp_path: Path) -> None: + sheet = parse_accession_sheet( + _write_sheet(tmp_path, _record()), + ) + + assert sheet.rows[0].name is None + assert sheet.rows[0].organism is None + + def test_row_shorter_than_the_header_is_usage_error( + self, tmp_path: Path, + ) -> None: + path = tmp_path / "sheet.csv" + path.write_text("accession,sample_type,name,organism\nERR1,rna_seq\n") + + with pytest.raises( + CliUsageError, match=r"data row\(s\) 1 has fewer cells than the header", + ): + parse_accession_sheet(path) + + def test_name_and_organism_are_parsed(self, tmp_path: Path) -> None: + sheet = parse_accession_sheet(_write_sheet( + tmp_path, + _record(name="liver_r1", organism="Hs"), + )) + + assert sheet.rows[0].name == "liver_r1" + assert sheet.rows[0].organism == "Hs" + + def test_sample_type_is_parsed(self, tmp_path: Path) -> None: + sheet = parse_accession_sheet(_write_sheet( + tmp_path, _record(sample_type="chip_seq"), + )) + + assert sheet.rows[0].sample_type == "chip_seq" + + def test_utf8_bom_is_stripped_from_first_header(self, tmp_path: Path) -> None: + path = tmp_path / "sheet.csv" + with path.open("w", newline="", encoding="utf-8-sig") as handle: + writer = csv.DictWriter(handle, fieldnames=HEADERS) + writer.writeheader() + writer.writerow(_record()) + + sheet = parse_accession_sheet(path) + + assert sheet.rows[0].accession == "ERR1160845" + + def test_row_numbers_are_one_based(self, tmp_path: Path) -> None: + sheet = parse_accession_sheet(_write_sheet( + tmp_path, + _record(accession="ERR1160845"), + _record(accession="ERR10677146"), + )) + + assert [row.row_number for row in sheet.rows] == [1, 2] + + def test_header_with_spaces_after_commas_is_still_recognised(self, tmp_path: Path) -> None: + path = tmp_path / "sheet.csv" + path.write_text("accession, sample_type, name\nERR1160845, rna_seq, liver_r1\n") + + sheet = parse_accession_sheet(path) + + assert sheet.rows[0].accession == "ERR1160845" + assert sheet.rows[0].sample_type == "rna_seq" + assert sheet.rows[0].name == "liver_r1" + assert sheet.rows[0].metadata == {} + + def test_trailing_empty_header_column_is_usage_error(self, tmp_path: Path) -> None: + # A trailing comma in the header row (e.g. from a spreadsheet export) + # must not become a metadata attribute with an empty name. + path = tmp_path / "sheet.csv" + path.write_text("accession,sample_type,name,\nERR1,rna_seq,liver_r1,note\n") + + with pytest.raises(CliUsageError, match=r"column\(s\) 4 is unnamed"): + parse_accession_sheet(path) + + def test_multiple_unnamed_columns_are_all_named_in_the_error(self, tmp_path: Path) -> None: + path = tmp_path / "sheet.csv" + path.write_text("accession,sample_type,,\nERR1,rna_seq,,\n") + + with pytest.raises(CliUsageError, match=r"column\(s\) 3, 4 are unnamed"): + parse_accession_sheet(path) + + def test_duplicated_metadata_column_is_usage_error(self, tmp_path: Path) -> None: + # Otherwise the earlier column's value is silently discarded, since + # csv.DictReader zips duplicate field names last-wins. + path = tmp_path / "sheet.csv" + path.write_text("accession,sample_type,treatment,treatment\nERR1,rna_seq,drugA,drugB\n") + + with pytest.raises(CliUsageError, match=r"column name\(s\) 'treatment' is duplicated"): + parse_accession_sheet(path) + + def test_duplicated_accession_column_is_usage_error(self, tmp_path: Path) -> None: + path = tmp_path / "sheet.csv" + path.write_text("accession,sample_type,accession\nERR1,rna_seq,ERR2\n") + + with pytest.raises(CliUsageError, match=r"column name\(s\) 'accession' is duplicated"): + parse_accession_sheet(path) + + def test_unnamed_and_duplicated_columns_reported_in_one_error(self, tmp_path: Path) -> None: + path = tmp_path / "sheet.csv" + path.write_text( + "accession,sample_type,treatment,treatment,\nERR1,rna_seq,drugA,drugB,note\n", + ) + + with pytest.raises(CliUsageError) as excinfo: + parse_accession_sheet(path) + + assert "column(s) 5 is unnamed" in str(excinfo.value) + assert "column name(s) 'treatment' is duplicated" in str(excinfo.value) + + def test_row_with_more_cells_than_the_header_is_usage_error(self, tmp_path: Path) -> None: + path = tmp_path / "sheet.csv" + path.write_text("accession,sample_type,name\nERR1,rna_seq,liver_r1,note\n") + + with pytest.raises( + CliUsageError, match=r"data row\(s\) 1 has more cells than the header", + ): + parse_accession_sheet(path) + + def test_shifted_right_row_with_a_real_value_is_usage_error_not_skipped( + self, tmp_path: Path, + ) -> None: + # Every *named* cell is blank, but the accession landed in the + # overflow column — this must not be treated as a wholly-blank row. + path = tmp_path / "sheet.csv" + path.write_text("accession,sample_type,name\n,,,ERR1\n") + + with pytest.raises( + CliUsageError, match=r"data row\(s\) 1 has more cells than the header", + ): + parse_accession_sheet(path) + + def test_wholly_blank_row_with_extra_trailing_comma_is_still_skipped( + self, tmp_path: Path, + ) -> None: + # A comma-only line one cell wider than the header (the overflow + # carries no value) is spreadsheet noise, not a row with data in + # the wrong place — must not be confused with the shifted-right case. + path = tmp_path / "sheet.csv" + path.write_text("accession,sample_type,name\nERR1,rna_seq,liver_r1\n,,,\n") + + sheet = parse_accession_sheet(path) + + assert [row.accession for row in sheet.rows] == ["ERR1"] + + def test_row_with_a_blank_overflow_cell_is_still_accepted( + self, tmp_path: Path, + ) -> None: + # One cell wider than the header, but the overflow is blank and + # every named cell has its normal value — this is harmless trailing + # noise, not the shifted-right case above, where the row's *data* + # spilled into the overflow instead. + path = tmp_path / "sheet.csv" + path.write_text("accession,sample_type,name\nERR1,rna_seq,liver_r1,\n") + + sheet = parse_accession_sheet(path) + + assert sheet.rows[0].accession == "ERR1" + assert sheet.rows[0].name == "liver_r1" + + def test_quoted_value_containing_a_comma_is_not_treated_as_overflow( + self, tmp_path: Path, + ) -> None: + # The over-wide-row error's remedy tells the user to quote a value + # that legitimately contains a comma — confirm that actually works. + path = tmp_path / "sheet.csv" + path.write_text('accession,sample_type,name\nERR1,rna_seq,"liver, left lobe"\n') + + sheet = parse_accession_sheet(path) + + assert sheet.rows[0].name == "liver, left lobe" + + def test_wholly_blank_row_shorter_than_the_header_is_still_skipped( + self, tmp_path: Path, + ) -> None: + # A blank trailing line can have fewer commas than the header width + # too (some spreadsheet exports trim trailing empty cells) — this + # must be treated the same as the exact-width blank-row case above, + # not as a short row with data that can't be attributed to a column. + path = tmp_path / "sheet.csv" + path.write_text("accession,sample_type,name\nERR1,rna_seq,liver_r1\n,\n") + + sheet = parse_accession_sheet(path) + + assert [row.accession for row in sheet.rows] == ["ERR1"] + + def test_sheet_with_both_a_short_and_an_over_wide_row_reports_both( + self, tmp_path: Path, + ) -> None: + path = tmp_path / "sheet.csv" + path.write_text( + "accession,sample_type,name\nERR1,rna_seq\nERR2,rna_seq,liver_r1,note\n", + ) + + with pytest.raises(CliUsageError) as excinfo: + parse_accession_sheet(path) + + assert "data row(s) 1 has fewer cells than the header" in str(excinfo.value) + assert "data row(s) 2 has more cells than the header" in str(excinfo.value) + assert "Add the missing trailing comma(s)" in str(excinfo.value) + assert "quote it; otherwise remove the extra cell(s)" in str(excinfo.value) + + def test_non_csv_xlsx_rejected_with_export_message(self, tmp_path: Path) -> None: + xlsx = tmp_path / "sheet.xlsx" + xlsx.write_bytes(b"PK") + + with pytest.raises(CliUsageError, match="CSV"): + parse_accession_sheet(xlsx) + + def test_tsv_sheet_rejected(self, tmp_path: Path) -> None: + tsv = tmp_path / "sheet.tsv" + tsv.write_text("accession\nERR1160845\n") + + with pytest.raises(CliUsageError, match="CSV"): + parse_accession_sheet(tsv) + + def test_missing_file_is_usage_error(self, tmp_path: Path) -> None: + with pytest.raises(CliUsageError): + parse_accession_sheet(tmp_path / "absent.csv") + + def test_header_only_sheet_is_usage_error(self, tmp_path: Path) -> None: + with pytest.raises(CliUsageError, match="no rows"): + parse_accession_sheet(_write_sheet(tmp_path)) + + def test_wholly_empty_file_is_usage_error(self, tmp_path: Path) -> None: + path = tmp_path / "sheet.csv" + path.write_text("") + + with pytest.raises(CliUsageError, match="no rows"): + parse_accession_sheet(path) + + def test_blank_first_line_is_usage_error(self, tmp_path: Path) -> None: + # A blank line where the header belongs must not be read as a + # present, zero-column header — that would make every data row's + # cells overflow it instead of reporting the real problem. + path = tmp_path / "sheet.csv" + path.write_text("\nERR1,rna_seq\n") + + with pytest.raises(CliUsageError, match="no header row"): + parse_accession_sheet(path) + + def test_row_with_blank_accession_is_usage_error(self, tmp_path: Path) -> None: + with pytest.raises(CliUsageError, match=r"data row\(s\) 2 has no accession"): + parse_accession_sheet(_write_sheet( + tmp_path, + _record(accession="ERR1160845"), + _record(accession=""), + )) + + def test_row_with_no_accession_column_is_usage_error(self, tmp_path: Path) -> None: + path = tmp_path / "sheet.csv" + with path.open("w", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=["run", "name", "sample_type"]) + writer.writeheader() + writer.writerow({"run": "ERR1160845", "name": "liver_r1", "sample_type": "rna_seq"}) + + with pytest.raises(CliUsageError): + parse_accession_sheet(path) + + def test_usage_error_names_every_row_missing_an_accession(self, tmp_path: Path) -> None: + with pytest.raises(CliUsageError, match=r"data row\(s\) 1, 3 have no accession"): + parse_accession_sheet(_write_sheet( + tmp_path, + _record(accession=""), + _record(accession="ERR1"), + _record(accession=""), + )) + + def test_row_with_blank_sample_type_is_usage_error(self, tmp_path: Path) -> None: + with pytest.raises(CliUsageError, match=r"data row\(s\) 1 has no sample_type"): + parse_accession_sheet(_write_sheet(tmp_path, _record(sample_type=""))) + + def test_row_with_no_sample_type_column_is_usage_error(self, tmp_path: Path) -> None: + path = tmp_path / "sheet.csv" + with path.open("w", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=["accession", "name"]) + writer.writeheader() + writer.writerow({"accession": "ERR1160845", "name": "liver_r1"}) + + with pytest.raises(CliUsageError): + parse_accession_sheet(path) + + def test_sheet_broken_in_both_columns_reports_both_in_one_error( + self, tmp_path: Path, + ) -> None: + with pytest.raises(CliUsageError) as excinfo: + parse_accession_sheet(_write_sheet( + tmp_path, + _record(accession=""), + _record(accession="ERR2", sample_type=""), + )) + + assert "data row(s) 1 has no accession" in str(excinfo.value) + assert "data row(s) 2 has no sample_type" in str(excinfo.value) + + def test_trailing_blank_row_is_skipped(self, tmp_path: Path) -> None: + # A comma-only line, e.g. one a spreadsheet export leaves below the + # data — not a row with data but no accession, so it isn't an error. + sheet = parse_accession_sheet(_write_sheet( + tmp_path, + _record(accession="ERR1"), + {}, + )) + + assert [row.accession for row in sheet.rows] == ["ERR1"] + + def test_sheet_of_only_blank_rows_is_usage_error(self, tmp_path: Path) -> None: + with pytest.raises(CliUsageError, match="no rows"): + parse_accession_sheet(_write_sheet(tmp_path, {}, {})) + + def test_blank_row_in_the_middle_does_not_shift_later_row_numbers( + self, tmp_path: Path, + ) -> None: + with pytest.raises(CliUsageError, match=r"data row\(s\) 3 has no accession"): + parse_accession_sheet(_write_sheet( + tmp_path, + _record(accession="ERR1"), + {}, + _record(accession=""), + )) + + def test_a_line_with_no_commas_at_all_does_not_consume_a_row_number( + self, tmp_path: Path, + ) -> None: + # Unlike a comma-only blank row (above), a line with nothing on it + # at all isn't yielded as a row by csv.DictReader in the first + # place, so it doesn't take a row number — documented in cli.rst. + path = tmp_path / "sheet.csv" + path.write_text("accession,sample_type\nERR1,rna_seq\n\n,rna_seq\n") + + with pytest.raises(CliUsageError, match=r"data row\(s\) 2 has no accession"): + parse_accession_sheet(path) + + +def test_row_rejects_empty_accession_by_construction() -> None: + with pytest.raises(ValueError, match="accession"): + AccessionSheetRow( + row_number=1, + accession="", + name=None, + organism=None, + sample_type="rna_seq", + metadata={}, + ) + + +def test_row_rejects_empty_sample_type_by_construction() -> None: + with pytest.raises(ValueError, match="sample_type"): + AccessionSheetRow( + row_number=1, + accession="ERR1160845", + name=None, + organism=None, + sample_type="", + metadata={}, + ) + + +class TestAccessionSheetRowToSpec: + + def _row(self, tmp_path: Path, **overrides: str): + sheet = parse_accession_sheet(_write_sheet(tmp_path, _record(**overrides))) + return sheet.rows[0] + + def test_uses_the_row_sample_type(self, tmp_path: Path) -> None: + row = self._row(tmp_path, sample_type="chip_seq") + + spec = row.to_spec() + + assert spec == SampleImportSpec(accession="ERR1160845", sample_type="chip_seq") + + def test_carries_name_organism_and_metadata(self, tmp_path: Path) -> None: + row = self._row(tmp_path, name="liver_r1", organism="Hs", cell_type="Neuron") + + spec = row.to_spec() + + assert spec == SampleImportSpec( + accession="ERR1160845", + sample_type="rna_seq", + name="liver_r1", + organism_id="Hs", + metadata={"cell_type": "Neuron"}, + ) + + def test_annotation_suffixed_column_is_forwarded_as_a_plain_metadata_key( + self, tmp_path: Path, + ) -> None: + # Unlike upload-batch, there's no special handling of `__annotation` + # here — it's just another metadata column, and the server does not + # recognise it as one (see cli.rst). + row = self._row(tmp_path, source="blood", source__annotation="left lobe") + + spec = row.to_spec() + + assert spec.metadata == {"source": "blood", "source__annotation": "left lobe"} diff --git a/tests/unit/cli/test_samples.py b/tests/unit/cli/test_samples.py index 5c461c1..bdd8fb5 100644 --- a/tests/unit/cli/test_samples.py +++ b/tests/unit/cli/test_samples.py @@ -13,6 +13,7 @@ ANNOTATION_TEMPLATE_URL = f"{DEFAULT_BASE_URL}/annotation" ANNOTATION_UPLOAD_URL = f"{DEFAULT_BASE_URL}/upload/annotation" MULTIPLEXED_UPLOAD_URL = f"{DEFAULT_BASE_URL}/upload/multiplexed" +SAMPLE_IMPORTS_URL = f"{DEFAULT_BASE_URL}/v2/sample-imports" TOKEN = "test.token" @@ -923,3 +924,687 @@ def test_non_csv_sheet_is_usage_error(self, run_cli, tmp_path: Path) -> None: assert result.exit_code == 2 assert upload.call_count == 0 assert "CSV" in result.stderr + + +IMPORT_HEADERS = ["accession", "name", "organism", "sample_type", "cell_type", "source", "source__annotation"] + + +def _write_import_sheet(directory: Path, *records: dict[str, str]) -> Path: + path = directory / "accessions.csv" + with path.open("w", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=IMPORT_HEADERS) + writer.writeheader() + writer.writerows(records) + return path + + +def _import_record(**overrides: str) -> dict[str, str]: + record = {"accession": "ERR1", "sample_type": "rna_seq"} + record.update(overrides) + return record + + +def _job_json( + job_id: int, + status: str, + accessions: list[str], + sample_ids: list[int] | None = None, + error: str | None = None, + execution_id: int | None = 7, +) -> dict: + return { + "id": job_id, + "status": status, + "created": 1700000000, + "started": 1700000001 if status != "RUNNING" else None, + "finished": 1700000002 if status in ("COMPLETED", "FAILED") else None, + "accessions": accessions, + "sample_ids": sample_ids or [], + "execution_id": execution_id, + "error": error, + } + + +class TestSamplesImport: + + @respx.mock + def test_kicks_off_job_and_reports_id_without_waiting( + self, run_cli, tmp_path: Path, + ) -> None: + route = respx.post(SAMPLE_IMPORTS_URL).mock( + return_value=httpx.Response(HTTPStatus.CREATED, json=_job_json( + 42, "RUNNING", ["ERR1", "ERR2"], + )), + ) + sheet = _write_import_sheet( + tmp_path, + _import_record(accession="ERR1", cell_type="Neuron"), + _import_record(accession="ERR2", cell_type="Fibroblast"), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import", "--sheet", str(sheet), + ) + + assert result.exit_code == 0 + assert route.call_count == 1 + assert "42" in result.stdout + assert "import-status" in result.stdout + + @respx.mock + def test_json_document_reports_job_fields( + self, run_cli, tmp_path: Path, + ) -> None: + respx.post(SAMPLE_IMPORTS_URL).mock( + return_value=httpx.Response(HTTPStatus.CREATED, json=_job_json( + 42, "RUNNING", ["ERR1"], + )), + ) + sheet = _write_import_sheet(tmp_path, _import_record(cell_type="Neuron")) + + result = run_cli( + "--token", TOKEN, "samples", "import", "--sheet", str(sheet), "--json", + ) + + assert result.exit_code == 0 + document = json.loads(result.stdout) + assert result.stdout.count("\n") == 1 + assert document == { + "id": 42, + "status": "RUNNING", + "created": "2023-11-14T22:13:20Z", + "started": None, + "finished": None, + "accessions": ["ERR1"], + "sample_ids": [], + "execution_id": 7, + "error": None, + } + + @respx.mock + def test_sends_every_row_without_local_validation( + self, run_cli, tmp_path: Path, + ) -> None: + # No metadata/sample-type endpoints are mocked: if the command called + # client.samples.get_metadata_attributes() or otherwise validated + # locally, respx would fail the test for an unmocked request. + route = respx.post(SAMPLE_IMPORTS_URL).mock( + return_value=httpx.Response(HTTPStatus.CREATED, json=_job_json( + 1, "RUNNING", ["not-an-accession", "ERR1", "ERR1"], + )), + ) + sheet = _write_import_sheet( + tmp_path, + _import_record(accession="not-an-accession", sample_type="bogus"), + _import_record(accession="ERR1"), + _import_record(accession="ERR1"), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import", "--sheet", str(sheet), + ) + + assert result.exit_code == 0 + payload = json.loads(route.calls[0].request.content) + assert [entry["accession"] for entry in payload["imports"]] == [ + "not-an-accession", "ERR1", "ERR1", + ] + + @respx.mock + def test_sends_name_organism_and_metadata_in_payload( + self, run_cli, tmp_path: Path, + ) -> None: + route = respx.post(SAMPLE_IMPORTS_URL).mock( + return_value=httpx.Response(HTTPStatus.CREATED, json=_job_json( + 1, "RUNNING", ["ERR1"], + )), + ) + sheet = _write_import_sheet(tmp_path, _import_record( + name="liver_r1", organism="Hs", cell_type="Neuron", + )) + + run_cli( + "--token", TOKEN, "samples", "import", "--sheet", str(sheet), + ) + + payload = json.loads(route.calls[0].request.content) + assert payload == { + "imports": [{ + "accession": "ERR1", + "sample_type": "rna_seq", + "name": "liver_r1", + "organism": "Hs", + "metadata": {"cell_type": "Neuron"}, + }], + } + + @respx.mock + def test_each_row_uses_its_own_sample_type( + self, run_cli, tmp_path: Path, + ) -> None: + route = respx.post(SAMPLE_IMPORTS_URL).mock( + return_value=httpx.Response(HTTPStatus.CREATED, json=_job_json( + 1, "RUNNING", ["ERR1", "ERR2"], + )), + ) + sheet = _write_import_sheet( + tmp_path, + _import_record(accession="ERR1", sample_type="chip_seq"), + _import_record(accession="ERR2", sample_type="atac_seq"), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import", "--sheet", str(sheet), + ) + + assert result.exit_code == 0 + payload = json.loads(route.calls[0].request.content) + sample_types = [entry["sample_type"] for entry in payload["imports"]] + assert sample_types == ["chip_seq", "atac_seq"] + + @respx.mock + def test_trailing_blank_row_is_skipped( + self, run_cli, tmp_path: Path, + ) -> None: + route = respx.post(SAMPLE_IMPORTS_URL).mock( + return_value=httpx.Response(HTTPStatus.CREATED, json=_job_json( + 1, "RUNNING", ["ERR1"], + )), + ) + sheet = _write_import_sheet( + tmp_path, _import_record(accession="ERR1"), {}, + ) + + result = run_cli( + "--token", TOKEN, "samples", "import", "--sheet", str(sheet), + ) + + assert result.exit_code == 0 + payload = json.loads(route.calls[0].request.content) + assert [entry["accession"] for entry in payload["imports"]] == ["ERR1"] + + @respx.mock + def test_api_rejection_propagates_as_error( + self, run_cli, tmp_path: Path, + ) -> None: + route = respx.post(SAMPLE_IMPORTS_URL).mock( + return_value=httpx.Response( + HTTPStatus.UNPROCESSABLE_ENTITY, + json={"error": "sample type 'bogus' does not exist"}, + ), + ) + sheet = _write_import_sheet(tmp_path, _import_record(sample_type="bogus")) + + result = run_cli( + "--token", TOKEN, "samples", "import", "--sheet", str(sheet), + ) + + assert result.exit_code == 1 + assert route.call_count == 1 + assert "bogus" in result.stderr + + @respx.mock + def test_non_csv_sheet_is_usage_error(self, run_cli, tmp_path: Path) -> None: + route = respx.post(SAMPLE_IMPORTS_URL) + xlsx = tmp_path / "sheet.xlsx" + xlsx.write_bytes(b"PK\x03\x04") + + result = run_cli( + "--token", TOKEN, "samples", "import", "--sheet", str(xlsx), + ) + + assert result.exit_code == 2 + assert route.call_count == 0 + assert "CSV" in result.stderr + + def test_missing_sheet_is_usage_error(self, run_cli) -> None: + result = run_cli("--token", TOKEN, "samples", "import") + + assert result.exit_code == 2 + + @respx.mock + def test_header_only_sheet_is_usage_error(self, run_cli, tmp_path: Path) -> None: + route = respx.post(SAMPLE_IMPORTS_URL) + sheet = _write_import_sheet(tmp_path) + + result = run_cli( + "--token", TOKEN, "samples", "import", "--sheet", str(sheet), + ) + + assert result.exit_code == 2 + assert route.call_count == 0 + assert "no rows" in result.stderr + + @respx.mock + def test_row_missing_sample_type_is_usage_error( + self, run_cli, tmp_path: Path, + ) -> None: + route = respx.post(SAMPLE_IMPORTS_URL) + sheet = _write_import_sheet( + tmp_path, + _import_record(accession="ERR1"), + _import_record(accession="ERR2", sample_type=""), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import", "--sheet", str(sheet), + ) + + assert result.exit_code == 2 + assert route.call_count == 0 + assert "data row(s) 2 has no sample_type" in result.stderr + + @respx.mock + def test_sheet_with_only_blank_accessions_is_usage_error( + self, run_cli, tmp_path: Path, + ) -> None: + route = respx.post(SAMPLE_IMPORTS_URL) + sheet = _write_import_sheet( + tmp_path, _import_record(accession=""), _import_record(accession=""), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import", "--sheet", str(sheet), + ) + + assert result.exit_code == 2 + assert route.call_count == 0 + assert "data row(s) 1, 2 have no accession" in result.stderr + + @respx.mock + def test_mixed_valid_and_blank_accession_rows_is_usage_error( + self, run_cli, tmp_path: Path, + ) -> None: + # A blank accession is rejected outright rather than silently + # skipped, even when the rest of the sheet is otherwise fine. + route = respx.post(SAMPLE_IMPORTS_URL) + sheet = _write_import_sheet( + tmp_path, + _import_record(accession="ERR1"), + _import_record(accession=""), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import", "--sheet", str(sheet), + ) + + assert result.exit_code == 2 + assert route.call_count == 0 + assert "data row(s) 2 has no accession" in result.stderr + + @respx.mock + def test_sheet_broken_in_both_columns_reports_both_in_one_error( + self, run_cli, tmp_path: Path, + ) -> None: + route = respx.post(SAMPLE_IMPORTS_URL) + sheet = _write_import_sheet( + tmp_path, + _import_record(accession=""), + _import_record(accession="ERR2", sample_type=""), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import", "--sheet", str(sheet), + ) + + assert result.exit_code == 2 + assert route.call_count == 0 + assert "data row(s) 1 has no accession" in result.stderr + assert "data row(s) 2 has no sample_type" in result.stderr + + @respx.mock + def test_sheet_with_no_accession_column_is_usage_error( + self, run_cli, tmp_path: Path, + ) -> None: + route = respx.post(SAMPLE_IMPORTS_URL) + sheet = tmp_path / "accessions.csv" + with sheet.open("w", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=["run", "name", "sample_type"]) + writer.writeheader() + writer.writerow({"run": "ERR1160845", "name": "liver_r1", "sample_type": "rna_seq"}) + + result = run_cli( + "--token", TOKEN, "samples", "import", "--sheet", str(sheet), + ) + + assert result.exit_code == 2 + assert route.call_count == 0 + assert "data row(s) 1 has no accession" in result.stderr + + @respx.mock + def test_unnamed_header_column_is_usage_error( + self, run_cli, tmp_path: Path, + ) -> None: + route = respx.post(SAMPLE_IMPORTS_URL) + sheet = tmp_path / "accessions.csv" + sheet.write_text("accession,sample_type,name,\nERR1,rna_seq,liver_r1,note\n") + + result = run_cli( + "--token", TOKEN, "samples", "import", "--sheet", str(sheet), + ) + + assert result.exit_code == 2 + assert route.call_count == 0 + assert "column(s) 4 is unnamed" in result.stderr + + @respx.mock + def test_duplicated_header_column_is_usage_error( + self, run_cli, tmp_path: Path, + ) -> None: + route = respx.post(SAMPLE_IMPORTS_URL) + sheet = tmp_path / "accessions.csv" + sheet.write_text("accession,sample_type,treatment,treatment\nERR1,rna_seq,drugA,drugB\n") + + result = run_cli( + "--token", TOKEN, "samples", "import", "--sheet", str(sheet), + ) + + assert result.exit_code == 2 + assert route.call_count == 0 + assert "column name(s) 'treatment' is duplicated" in result.stderr + + @respx.mock + def test_row_with_more_cells_than_the_header_is_usage_error( + self, run_cli, tmp_path: Path, + ) -> None: + route = respx.post(SAMPLE_IMPORTS_URL) + sheet = tmp_path / "accessions.csv" + sheet.write_text("accession,sample_type,name\nERR1,rna_seq,liver_r1,note\n") + + result = run_cli( + "--token", TOKEN, "samples", "import", "--sheet", str(sheet), + ) + + assert result.exit_code == 2 + assert route.call_count == 0 + assert "data row(s) 1 has more cells than the header" in result.stderr + + @respx.mock + def test_row_with_fewer_cells_than_the_header_is_usage_error( + self, run_cli, tmp_path: Path, + ) -> None: + route = respx.post(SAMPLE_IMPORTS_URL) + sheet = tmp_path / "accessions.csv" + sheet.write_text("accession,sample_type,name,organism\nERR1,rna_seq\n") + + result = run_cli( + "--token", TOKEN, "samples", "import", "--sheet", str(sheet), + ) + + assert result.exit_code == 2 + assert route.call_count == 0 + assert "data row(s) 1 has fewer cells than the header" in result.stderr + + +class TestSamplesImportStatus: + + @respx.mock + def test_reports_running_job(self, run_cli) -> None: + respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock( + return_value=httpx.Response(HTTPStatus.OK, json=_job_json( + 42, "RUNNING", ["ERR1", "ERR2"], + )), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import-status", "--job-id", "42", + ) + + assert result.exit_code == 0 + assert "42" in result.stdout + assert "RUNNING" in result.stdout + + @respx.mock + def test_running_job_with_started_reports_when_it_started(self, run_cli) -> None: + respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock( + return_value=httpx.Response(HTTPStatus.OK, json={ + "id": 42, "status": "RUNNING", "created": 1700000000, + "started": 1700000001, "finished": None, + "accessions": ["ERR1"], "sample_ids": [], "execution_id": None, "error": None, + }), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import-status", "--job-id", "42", + ) + + assert result.exit_code == 0 + assert "started 2023-11-14" in result.stdout + + @respx.mock + def test_started_with_non_utc_offset_is_reported_in_utc(self, run_cli) -> None: + respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock( + return_value=httpx.Response(HTTPStatus.OK, json={ + "id": 42, "status": "RUNNING", "created": 1700000000, + "started": "2024-04-05T19:34:38+02:00", "finished": None, + "accessions": ["ERR1"], "sample_ids": [], "execution_id": None, "error": None, + }), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import-status", "--job-id", "42", + ) + + assert result.exit_code == 0 + assert "started 2024-04-05 17:34:38 UTC" in result.stdout + + @respx.mock + def test_started_with_no_offset_is_treated_as_utc(self, run_cli) -> None: + respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock( + return_value=httpx.Response(HTTPStatus.OK, json={ + "id": 42, "status": "RUNNING", "created": 1700000000, + "started": "2024-04-05T19:34:38", "finished": None, + "accessions": ["ERR1"], "sample_ids": [], "execution_id": None, "error": None, + }), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import-status", "--job-id", "42", + ) + + assert result.exit_code == 0 + assert "started 2024-04-05 19:34:38 UTC" in result.stdout + + @respx.mock + def test_naive_started_is_reported_as_utc_in_json_too(self, run_cli) -> None: + respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock( + return_value=httpx.Response(HTTPStatus.OK, json={ + "id": 42, "status": "RUNNING", "created": 1700000000, + "started": "2024-04-05T19:34:38", "finished": None, + "accessions": ["ERR1"], "sample_ids": [], "execution_id": None, "error": None, + }), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import-status", "--job-id", "42", "--json", + ) + + assert result.exit_code == 0 + document = json.loads(result.stdout) + assert document["started"] == "2024-04-05T19:34:38Z" + + @respx.mock + def test_non_utc_offset_started_is_reported_as_utc_in_json_too(self, run_cli) -> None: + respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock( + return_value=httpx.Response(HTTPStatus.OK, json={ + "id": 42, "status": "RUNNING", "created": 1700000000, + "started": "2024-04-05T19:34:38+02:00", "finished": None, + "accessions": ["ERR1"], "sample_ids": [], "execution_id": None, "error": None, + }), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import-status", "--job-id", "42", "--json", + ) + + assert result.exit_code == 0 + document = json.loads(result.stdout) + assert document["started"] == "2024-04-05T17:34:38Z" + + @respx.mock + def test_running_job_falls_back_to_created_when_not_started(self, run_cli) -> None: + respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock( + return_value=httpx.Response(HTTPStatus.OK, json={ + "id": 42, "status": "RUNNING", "created": 1700000000, + "started": None, "finished": None, + "accessions": ["ERR1"], "sample_ids": [], "execution_id": None, "error": None, + }), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import-status", "--job-id", "42", + ) + + assert result.exit_code == 0 + assert "created 2023-11-14" in result.stdout + + @respx.mock + def test_completed_job_reports_when_it_finished(self, run_cli) -> None: + respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock( + return_value=httpx.Response(HTTPStatus.OK, json=_job_json( + 42, "COMPLETED", ["ERR1"], [101], + )), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import-status", "--job-id", "42", + ) + + assert result.exit_code == 0 + assert "finished 2023-11-14" in result.stdout + + @respx.mock + def test_reports_completed_job_with_sample_ids(self, run_cli) -> None: + respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock( + return_value=httpx.Response(HTTPStatus.OK, json=_job_json( + 42, "COMPLETED", ["ERR1", "ERR2"], [101, 102], + )), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import-status", "--job-id", "42", + ) + + assert result.exit_code == 0 + assert "101" in result.stdout + assert "102" in result.stdout + + @respx.mock + def test_reports_failed_job_with_error(self, run_cli) -> None: + respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock( + return_value=httpx.Response(HTTPStatus.OK, json=_job_json( + 42, "FAILED", ["ERR1"], error="download failed", + )), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import-status", "--job-id", "42", + ) + + assert result.exit_code == 1 + assert "FAILED" in result.stdout + assert "download failed" not in result.stdout + assert "download failed" in result.stderr + + @respx.mock + def test_failed_job_json_mode_has_error_on_stdout_only(self, run_cli) -> None: + respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock( + return_value=httpx.Response(HTTPStatus.OK, json=_job_json( + 42, "FAILED", ["ERR1"], error="download failed", + )), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import-status", "--job-id", "42", "--json", + ) + + assert result.exit_code == 1 + assert result.stderr == "" + document = json.loads(result.stdout) + assert document["error"] == "download failed" + + def test_non_numeric_job_id_reports_clear_message(self, run_cli) -> None: + result = run_cli( + "--token", TOKEN, "samples", "import-status", "--job-id", "not-a-number", + ) + + assert result.exit_code == 2 + assert "_job_id" not in result.stderr + assert "job id" in result.stderr.lower() + + @respx.mock + def test_completed_job_with_no_sample_ids_reports_none(self, run_cli) -> None: + respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock( + return_value=httpx.Response(HTTPStatus.OK, json=_job_json( + 42, "COMPLETED", ["ERR1"], [], + )), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import-status", "--job-id", "42", + ) + + assert result.exit_code == 0 + assert "none" in result.stdout + + @respx.mock + def test_json_document_matches_job_shape(self, run_cli) -> None: + respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock( + return_value=httpx.Response(HTTPStatus.OK, json=_job_json( + 42, "COMPLETED", ["ERR1"], [101], + )), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import-status", "--job-id", "42", "--json", + ) + + assert result.exit_code == 0 + document = json.loads(result.stdout) + assert result.stdout.count("\n") == 1 + assert document == { + "id": 42, + "status": "COMPLETED", + "created": "2023-11-14T22:13:20Z", + "started": "2023-11-14T22:13:21Z", + "finished": "2023-11-14T22:13:22Z", + "accessions": ["ERR1"], + "sample_ids": [101], + "execution_id": 7, + "error": None, + } + + @respx.mock + def test_reports_job_with_no_execution_yet(self, run_cli) -> None: + respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock( + return_value=httpx.Response(HTTPStatus.OK, json=_job_json( + 42, "RUNNING", ["ERR1"], execution_id=None, + )), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import-status", "--job-id", "42", "--json", + ) + + assert result.exit_code == 0 + document = json.loads(result.stdout) + assert document["execution_id"] is None + + @respx.mock + def test_unknown_job_id_is_not_found(self, run_cli) -> None: + respx.get(f"{SAMPLE_IMPORTS_URL}/999").mock( + return_value=httpx.Response( + HTTPStatus.NOT_FOUND, json={"error": "sample import 999 does not exist"}, + ), + ) + + result = run_cli( + "--token", TOKEN, "samples", "import-status", "--job-id", "999", + ) + + assert result.exit_code == 4 + + def test_missing_job_id_is_usage_error(self, run_cli) -> None: + result = run_cli("--token", TOKEN, "samples", "import-status") + + assert result.exit_code == 2 diff --git a/tests/unit/v2/test_package_exports.py b/tests/unit/v2/test_package_exports.py index 2bb0a12..06bff5a 100644 --- a/tests/unit/v2/test_package_exports.py +++ b/tests/unit/v2/test_package_exports.py @@ -20,6 +20,25 @@ def test_username_password_credentials_is_exported(self) -> None: assert UsernamePasswordCredentials is DirectCredentials + def test_sample_import_types_are_exported(self) -> None: + from flowbio.v2 import ( + SampleImportJob, + SampleImportJobId, + SampleImportSpec, + SampleImportStatus, + ) + from flowbio.v2.samples import ( + SampleImportJob as DirectJob, + SampleImportJobId as DirectJobId, + SampleImportSpec as DirectSpec, + SampleImportStatus as DirectStatus, + ) + + assert SampleImportSpec is DirectSpec + assert SampleImportJob is DirectJob + assert SampleImportJobId is DirectJobId + assert SampleImportStatus is DirectStatus + class TestSubmoduleExports: diff --git a/tests/unit/v2/test_samples.py b/tests/unit/v2/test_samples.py index cc42ad7..9444c6c 100644 --- a/tests/unit/v2/test_samples.py +++ b/tests/unit/v2/test_samples.py @@ -1,3 +1,5 @@ +import json +from datetime import datetime, timezone from http import HTTPStatus from pathlib import Path from unittest.mock import ANY, patch @@ -10,6 +12,7 @@ from flowbio.v2.exceptions import ( AnnotationValidationError, BadRequestError, + FlowApiError, NotFoundError, ) from flowbio.v2.samples import ( @@ -17,6 +20,9 @@ MultiplexedUpload, Organism, Project, + SampleImportJob, + SampleImportJobId, + SampleImportSpec, SampleResource, SampleType, Sample, @@ -1049,3 +1055,248 @@ def test_chunked_multiplexed_upload(self, tmp_path: Path) -> None: assert mux_route.call_count == 3 assert result.data_ids == ["mux_1"] + + +class TestImportSamples: + + @respx.mock + def test_posts_imports_and_parses_job(self) -> None: + route = respx.post(f"{DEFAULT_BASE_URL}/v2/sample-imports").mock( + return_value=httpx.Response(HTTPStatus.CREATED, json={ + "id": 42, + "status": "RUNNING", + "created": 1700000000, + "started": None, + "finished": None, + "accessions": ["ERR1160845"], + "sample_ids": [], + "execution_id": None, + "error": None, + }), + ) + + client = Client() + result = client.samples.import_samples([ + SampleImportSpec(accession="ERR1160845", sample_type="rna_seq"), + ]) + + assert result == SampleImportJob( + id=SampleImportJobId(42), + status="RUNNING", + created=1700000000, + started=None, + finished=None, + accessions=["ERR1160845"], + sample_ids=[], + execution_id=None, + error=None, + ) + assert route.call_count == 1 + + @respx.mock + def test_sends_accession_and_sample_type(self) -> None: + route = respx.post(f"{DEFAULT_BASE_URL}/v2/sample-imports").mock( + return_value=httpx.Response(HTTPStatus.CREATED, json={ + "id": 1, "status": "RUNNING", "created": 1700000000, "accessions": ["ERR1"], + "sample_ids": [], "execution_id": None, "error": None, + }), + ) + + client = Client() + client.samples.import_samples([ + SampleImportSpec(accession="ERR1", sample_type="rna_seq"), + ]) + + payload = json.loads(route.calls[0].request.content) + assert payload == { + "imports": [{"accession": "ERR1", "sample_type": "rna_seq"}], + } + + @respx.mock + def test_sends_optional_fields_when_present(self) -> None: + route = respx.post(f"{DEFAULT_BASE_URL}/v2/sample-imports").mock( + return_value=httpx.Response(HTTPStatus.CREATED, json={ + "id": 1, "status": "RUNNING", "created": 1700000000, "accessions": ["ERR1"], + "sample_ids": [], "execution_id": None, "error": None, + }), + ) + + client = Client() + client.samples.import_samples([ + SampleImportSpec( + accession="ERR1", + sample_type="rna_seq", + name="my_sample", + organism_id="Hs", + metadata={"strandedness": "reverse"}, + ), + ]) + + payload = json.loads(route.calls[0].request.content) + assert payload == { + "imports": [{ + "accession": "ERR1", + "sample_type": "rna_seq", + "name": "my_sample", + "organism": "Hs", + "metadata": {"strandedness": "reverse"}, + }], + } + + @respx.mock + def test_empty_string_optional_fields_are_omitted_not_the_required_ones(self) -> None: + route = respx.post(f"{DEFAULT_BASE_URL}/v2/sample-imports").mock( + return_value=httpx.Response(HTTPStatus.CREATED, json={ + "id": 1, "status": "RUNNING", "created": 1700000000, "accessions": ["ERR1"], + "sample_ids": [], "execution_id": None, "error": None, + }), + ) + + client = Client() + client.samples.import_samples([ + SampleImportSpec(accession="ERR1", sample_type="rna_seq", name="", organism_id=""), + ]) + + payload = json.loads(route.calls[0].request.content) + assert payload == { + "imports": [{"accession": "ERR1", "sample_type": "rna_seq"}], + } + + @respx.mock + def test_sends_multiple_imports_in_one_request(self) -> None: + route = respx.post(f"{DEFAULT_BASE_URL}/v2/sample-imports").mock( + return_value=httpx.Response(HTTPStatus.CREATED, json={ + "id": 1, "status": "RUNNING", "created": 1700000000, "accessions": ["ERR1", "ERR2"], + "sample_ids": [], "execution_id": None, "error": None, + }), + ) + + client = Client() + client.samples.import_samples([ + SampleImportSpec(accession="ERR1", sample_type="rna_seq"), + SampleImportSpec(accession="ERR2", sample_type="rna_seq"), + ]) + + payload = json.loads(route.calls[0].request.content) + assert [entry["accession"] for entry in payload["imports"]] == ["ERR1", "ERR2"] + assert route.call_count == 1 + + @respx.mock + def test_raises_flow_api_error_on_validation_failure(self) -> None: + respx.post(f"{DEFAULT_BASE_URL}/v2/sample-imports").mock( + return_value=httpx.Response( + HTTPStatus.UNPROCESSABLE_ENTITY, + json={"error": "at least one accession is required"}, + ), + ) + + client = Client() + + with pytest.raises(FlowApiError) as exc_info: + client.samples.import_samples([]) + + assert exc_info.value.status_code == HTTPStatus.UNPROCESSABLE_ENTITY + + +class TestGetImport: + + @respx.mock + def test_parses_completed_job(self) -> None: + respx.get(f"{DEFAULT_BASE_URL}/v2/sample-imports/42").mock( + return_value=httpx.Response(HTTPStatus.OK, json={ + "id": 42, + "status": "COMPLETED", + "created": 1700000000, + "started": 1700000001, + "finished": 1700000002, + "accessions": ["ERR1160845", "ERR10677146"], + "sample_ids": [101, 102], + "execution_id": 7, + "error": None, + }), + ) + + client = Client() + result = client.samples.get_import(SampleImportJobId(42)) + + assert result == SampleImportJob( + id=SampleImportJobId(42), + status="COMPLETED", + created=1700000000, + started=1700000001, + finished=1700000002, + accessions=["ERR1160845", "ERR10677146"], + sample_ids=[101, 102], + execution_id=7, + error=None, + ) + + @respx.mock + def test_parses_failed_job_with_error(self) -> None: + respx.get(f"{DEFAULT_BASE_URL}/v2/sample-imports/42").mock( + return_value=httpx.Response(HTTPStatus.OK, json={ + "id": 42, + "status": "FAILED", + "created": 1700000000, + "started": 1700000001, + "finished": 1700000002, + "accessions": ["ERR1160845"], + "sample_ids": [], + "execution_id": 7, + "error": "download failed: connection reset", + }), + ) + + client = Client() + result = client.samples.get_import(SampleImportJobId(42)) + + assert result.status == "FAILED" + assert result.error == "download failed: connection reset" + + @respx.mock + def test_raises_not_found_for_unknown_job(self) -> None: + respx.get(f"{DEFAULT_BASE_URL}/v2/sample-imports/999").mock( + return_value=httpx.Response( + HTTPStatus.NOT_FOUND, json={"error": "sample import 999 does not exist"}, + ), + ) + + client = Client() + + with pytest.raises(NotFoundError): + client.samples.get_import(SampleImportJobId(999)) + + @respx.mock + def test_parses_job_with_only_id_and_status(self) -> None: + respx.get(f"{DEFAULT_BASE_URL}/v2/sample-imports/42").mock( + return_value=httpx.Response(HTTPStatus.OK, json={ + "id": 42, + "status": "RUNNING", + }), + ) + + client = Client() + result = client.samples.get_import(SampleImportJobId(42)) + + assert result.created is None + assert result.started is None + assert result.finished is None + assert result.accessions == [] + assert result.sample_ids == [] + assert result.execution_id is None + assert result.error is None + + @respx.mock + def test_naive_timestamp_is_treated_as_utc(self) -> None: + respx.get(f"{DEFAULT_BASE_URL}/v2/sample-imports/42").mock( + return_value=httpx.Response(HTTPStatus.OK, json={ + "id": 42, + "status": "RUNNING", + "started": "2024-04-05T19:34:38", + }), + ) + + client = Client() + result = client.samples.get_import(SampleImportJobId(42)) + + assert result.started == datetime(2024, 4, 5, 19, 34, 38, tzinfo=timezone.utc)