API Reference¶

@dataclass
class Alignment(Asset):
    """BAM/CRAM alignment file asset.

    Carries reference_cid and r1_cid for provenance (PROV entity derivation).
    The format is the file name's extension. Fields are kept to what fits a
    new public file's 9 metadata keys with the asset type and owner.
    """

    _asset_key: ClassVar[str] = "alignment"
    sample_id: str = ""
    sorted: str = ""
    duplicates_marked: bool = False
    bqsr_applied: bool = False
    tool: str = ""
    reference_cid: str = ""
    r1_cid: str = ""

@dataclass
class AlignmentIndex(Asset):
    """BAI/CRAI alignment index file asset.

    A companion: `companion_of` is the CID of the Alignment it indexes.
    """

    _asset_key: ClassVar[str] = "alignment_index"
    sample_id: str = ""
    companion_of: str = ""

@dataclass
class BQSRReport(Asset):
    """BQSR recalibration table produced by GATK BaseRecalibrator.

    Carries alignment_cid linking to the Alignment it was produced from.
    """

    _asset_key: ClassVar[str] = "bqsr_report"
    sample_id: str = ""
    tool: str = ""
    alignment_cid: str = ""

@dataclass
class DuplicateMetrics(Asset):
    """Duplicate metrics text file produced by GATK MarkDuplicates.

    Carries alignment_cid linking to the Alignment it was produced from.
    """

    _asset_key: ClassVar[str] = "duplicate_metrics"
    sample_id: str = ""
    tool: str = ""
    alignment_cid: str = ""

@dataclass
class Asset:
    """Base class for all typed file assets in Stargazer.

    Attributes:
        cid: IPFS CID of the stored file
        path: The stored file as a `flyte.io.File`. Assigning a local `Path`
            or a URI string wraps it in a File, so an asset can be built from
            a local file that was never uploaded; `fetch()` then uses that
            file in place.
        keyvalues: On a bare Asset, all of its metadata, serialized
            verbatim (the ``asset`` key lives inside the dict): the catchall
            for kinds of file with no registered class in this process. On
            a typed subclass, the extra keys its class doesn't declare,
            kept as strings and written beside the declared fields.

    Subclasses declare typed fields as normal dataclass attributes:

        @dataclass
        class Alignment(Asset):
            _asset_key: ClassVar[str] = "alignment"
            sample_id: str = ""
            duplicates_marked: bool = False

    Fields are plain Python attributes. ``to_keyvalues()`` serializes them to
    ``dict[str, str]`` at storage boundaries; ``from_keyvalues()`` reconstructs
    from storage. ``str`` fields pass through directly; all other types use
    ``json.dumps`` / ``json.loads``. ``declared_fields()`` lists a type's
    fields. Attributes are strict (assigning an undeclared one raises); extra
    keys go in ``keyvalues`` instead:

        aln.keyvalues["lane"] = "3"
    """

    _registry: ClassVar[dict[str, type[Asset]]] = {}
    _asset_key: ClassVar[str] = ""

    cid: str = ""
    path: File | None = None
    keyvalues: dict[str, str] = field(default_factory=dict)

    def __init_subclass__(cls, **kwargs):
        """Register subclass in the asset registry."""
        super().__init_subclass__(**kwargs)
        ak = cls.__dict__.get("_asset_key", "")
        if ak:
            Asset._registry[ak] = cls

    def __setattr__(self, name: str, value: Any) -> None:
        """Enforce declared fields on typed subclasses and coerce `path` to a File."""
        if name == "path":
            value = _as_file(value)
        elif self._asset_key and not name.startswith("_") and name not in _BASE_FIELDS:
            allowed = self.declared_fields()
            if name not in allowed:
                raise AttributeError(
                    f"{type(self).__name__} has no field '{name}'. "
                    f"Allowed: {sorted(allowed)}"
                )
        super().__setattr__(name, value)

    @classmethod
    def declared_fields(cls) -> tuple[str, ...]:
        """Names of the fields this asset type declares, in declaration order.

        These are the typed keyvalues it reads and writes: every dataclass
        field except the base ``cid``, ``path`` and ``keyvalues``. Bare Asset
        declares none.
        """
        return tuple(
            f.name for f in dataclasses.fields(cls) if f.name not in _BASE_FIELDS
        )

    def to_keyvalues(self) -> dict[str, str]:
        """Serialize to storage format.

        str fields pass through as-is; all other types are serialized with
        json.dumps. A typed asset's extra keys are written beside its
        declared fields. Bare Asset instances return their keyvalues
        verbatim. The result is always a copy, so callers (e.g. ``_owner``
        stamping at the storage layer) can mutate it without touching the
        Asset.

        Raises:
            ValueError: when a typed asset's ``keyvalues`` holds ``asset``
                or one of its declared fields, which are set as attributes
        """
        if not self._asset_key:
            return dict(self.keyvalues)
        declared = self.declared_fields()
        clash = sorted(set(self.keyvalues) & {"asset", *declared})
        if clash:
            raise ValueError(
                f"{type(self).__name__}.keyvalues holds {clash}, which "
                f"{type(self).__name__} sets itself: assign the attribute instead"
            )
        hints = get_type_hints(type(self))
        result: dict[str, str] = {"asset": self._asset_key}
        for name in declared:
            val = getattr(self, name)
            result[name] = val if hints.get(name) is str else json.dumps(val)
        result.update(self.keyvalues)
        return result

    @classmethod
    def from_keyvalues(
        cls, kv: dict[str, str], cid: str = "", path: File | Path | str | None = None
    ) -> Asset:
        """Reconstruct from a storage keyvalues dict.

        str fields are assigned directly; all other types are deserialized
        with json.loads (raising ``json.JSONDecodeError`` on values that
        don't parse — callers that must tolerate malformed records catch it,
        see ``specialize()``). On a typed class, every key that isn't
        ``asset`` or a declared field becomes an extra in ``keyvalues``.
        Bare Asset receives the keyvalues verbatim.
        """
        if not cls._asset_key:
            return cls(cid=cid, path=path, keyvalues=dict(kv))
        hints = get_type_hints(cls)
        declared = cls.declared_fields()
        kwargs = {
            name: value if hints.get(name) is str else json.loads(value)
            for name, value in kv.items()
            if name in declared
        }
        extras = {k: v for k, v in kv.items() if k not in declared and k != "asset"}
        return cls(cid=cid, path=path, keyvalues=extras, **kwargs)

    def to_dict(self) -> dict:
        """Serialize to a JSON-friendly dict: the file's location and name."""
        return {
            "cid": self.cid,
            "path": self.path.path if self.path else None,
            "name": self.path.name if self.path else None,
            "keyvalues": self.to_keyvalues(),
        }

    @classmethod
    def from_dict(cls, data: dict) -> Self:
        """Reconstruct from a serialized dict."""
        path = None
        if data.get("path"):
            path = _as_file(data["path"])
            if data.get("name"):
                path = File(path=path.path, name=data["name"])
        return cls.from_keyvalues(
            data.get("keyvalues", {}), cid=data.get("cid", ""), path=path
        )

    async def fetch(self) -> Path:
        """Make a local copy of this asset and its companions; return its path.

        The asset lands at `<STARGAZER_LOCAL>/<cid>/<name>`, and its
        companions (`StorageClient.companions`: an index, a sequence
        dictionary) land in the same directory, where tools look for them.
        Outputs that record it as an input under another key, such as an
        Alignment's ``reference_cid``, stay behind. An asset built from a
        local file that was never uploaded is returned in place. Copies
        already on disk are reused; a companion file left by an earlier fetch
        is replaced when a different one of that name is picked now. A
        companion named like the asset itself is skipped.
        """
        import stargazer.utils.storage as _storage
        from stargazer.assets import specialize

        client = _storage.default_client
        local = await client.download(self)
        if self.cid:
            for record in await client.companions(self.cid):
                if record["name"] != local.name:
                    await client.download(specialize(record), local.parent)
        return local

    async def update(self, path: Path, **kwargs) -> None:
        """Store a local file as this asset, setting fields from kwargs first.

        Sets `cid` and `path` (the stored File) once the file is uploaded and
        its index row has committed. Raises if either fails.
        """
        import stargazer.utils.storage as _storage

        for key, value in kwargs.items():
            if value is not None:
                setattr(self, key, value)
        await _storage.default_client.upload(self, Path(path))

@dataclass
class Notebook(Asset):
    """A marimo notebook published from a user's workspace to the public tier.

    A workspace notebook is a plain file until its owner publishes it from
    the dashboard, which is the only thing that creates one. `title` and
    `description` are the notebook's declared `main_name` and
    `main_description`, and `slug` is its workspace filename's stem, which
    ties the public record to the notebook it came from.
    """

    _asset_key: ClassVar[str] = "notebook"
    title: str = ""
    description: str = ""
    slug: str = ""

@dataclass
class R1(Asset):
    """R1 (forward) FASTQ read file asset.

    Carries mate_cid pointing to the paired R2 asset's CID (empty for single-end).
    """

    _asset_key: ClassVar[str] = "r1"
    sample_id: str = ""
    mate_cid: str = ""
    sequencing_platform: str = ""

@dataclass
class R2(Asset):
    """R2 (reverse) FASTQ read file asset.

    Carries mate_cid pointing to the paired R1 asset's CID (empty for single-end).
    """

    _asset_key: ClassVar[str] = "r2"
    sample_id: str = ""
    mate_cid: str = ""
    sequencing_platform: str = ""

@dataclass
class AlignerIndex(Asset):
    """Aligner index file asset (one file per index file for multi-file indices).

    A companion: `companion_of` is the CID of the Reference it indexes.
    """

    _asset_key: ClassVar[str] = "aligner_index"
    build: str = ""
    aligner: str = ""
    companion_of: str = ""

@dataclass
class Reference(Asset):
    """Reference FASTA file asset."""

    _asset_key: ClassVar[str] = "reference"
    build: str = ""

    async def contigs(self) -> list[str]:
        """Read contig names from the companion .fai index.

        Fetches the reference first, which brings the ReferenceIndex
        companion alongside the local copy, so it works in any pod.
        """
        path = await self.fetch()
        fai_path = path.with_name(path.name + ".fai")
        if not fai_path.exists():
            raise FileNotFoundError(
                f"Reference index not found at {fai_path}. Run samtools_faidx first."
            )
        async with aiofiles.open(fai_path) as f:
            lines = (await f.read()).splitlines()
        return [line.split("\t", 1)[0].strip() for line in lines if line.strip()]

@dataclass
class ReferenceIndex(Asset):
    """FASTA index (.fai) file asset.

    A companion: `companion_of` is the CID of the Reference it indexes.
    """

    _asset_key: ClassVar[str] = "reference_index"
    build: str = ""
    tool: str = ""
    companion_of: str = ""

@dataclass
class SequenceDict(Asset):
    """Sequence dictionary (.dict) file asset.

    A companion: `companion_of` is the CID of the Reference it describes.
    """

    _asset_key: ClassVar[str] = "sequence_dict"
    build: str = ""
    tool: str = ""
    companion_of: str = ""

@dataclass
class AnnData(Asset):
    """AnnData (.h5ad) file asset for single-cell RNA-seq data.

    Tracks pipeline stage, cell/gene counts, and provenance through
    the scRNA-seq processing steps.
    """

    _asset_key: ClassVar[str] = "anndata"
    sample_id: str = ""
    n_obs: int = 0
    n_vars: int = 0
    stage: str = ""
    organism: str = ""
    source_cid: str = ""

@dataclass
class KnownSites(Asset):
    """Known variant sites VCF: BQSR known sites, or a VQSR training resource.

    `resource_name`, `known`, `training`, `truth` and `prior` are the VQSR
    resource arguments (`"true"`/`"false"` strings and a numeric prior).
    """

    _asset_key: ClassVar[str] = "known_sites"
    build: str = ""
    resource_name: str = ""
    known: str = "false"
    training: str = "false"
    truth: str = "false"
    prior: str = "10"
    vqsr_mode: str = ""

@dataclass
class KnownSitesIndex(Asset):
    """VCF index (.idx) file for a KnownSites asset.

    A companion: `companion_of` is the CID of the KnownSites VCF it indexes.
    """

    _asset_key: ClassVar[str] = "known_sites_index"
    companion_of: str = ""

@dataclass
class VQSRModel(Asset):
    """VQSR recalibration model: the .recal file VariantRecalibrator writes.

    ApplyVQSR needs two more files from the same run, stored as its
    companions: the recal file's index (`VQSRModelIndex`) and the tranches
    file (`VQSRTranches`).
    """

    _asset_key: ClassVar[str] = "vqsr_model"
    sample_id: str = ""
    mode: str = "SNP"
    build: str = ""
    variants_cid: str = ""

@dataclass
class VQSRModelIndex(Asset):
    """VCF index (.idx) for a VQSRModel's recal file.

    A companion: `companion_of` is the CID of the VQSRModel it indexes.
    """

    _asset_key: ClassVar[str] = "vqsr_model_index"
    sample_id: str = ""
    companion_of: str = ""

@dataclass
class VQSRTranches(Asset):
    """VQSR tranches file: the VQSLOD cutoff for each truth-sensitivity level.

    A companion: `companion_of` is the CID of the VQSRModel it was built
    with.
    """

    _asset_key: ClassVar[str] = "vqsr_tranches"
    sample_id: str = ""
    companion_of: str = ""

@dataclass
class Variants(Asset):
    """VCF/GVCF variant call file asset."""

    _asset_key: ClassVar[str] = "variants"
    sample_id: str = ""
    caller: str = ""
    variant_type: str = ""
    build: str = ""
    vqsr_mode: str = ""
    sample_count: int = 0
    source_samples: list = None

@dataclass
class VariantsIndex(Asset):
    """VCF index file asset (.idx, or .tbi for a bgzipped VCF).

    A companion: `companion_of` is the CID of the Variants file it indexes.
    """

    _asset_key: ClassVar[str] = "variants_index"
    sample_id: str = ""
    companion_of: str = ""

@app.class_definition
@dataclass
class MyAsset(Asset):
    """TODO: describe your file type."""

    _asset_key: ClassVar[str] = "my_asset"  # TODO: rename

@app.class_definition
@dataclass
class SampleSheet(Asset):
    """CSV of per-sample metadata for a cohort."""

    _asset_key: ClassVar[str] = "sample_sheet"
    cohort_id: str = ""
    n_samples: int = 0
    organism: str = ""