Stores¶

In-memory track and genome stores with UUID keys and provenance-based result caching. When two requests produce the same computation (identical operation, input IDs, and parameters), the store returns the cached result instead of recomputing.

Provenance `dataclass` ¶

Provenance(operation: str, input_ids: tuple[str, ...], params: tuple[tuple[str, Any], ...])

Records how a track was produced.

Parameters:

Name	Type	Description	Default
`operation`	`str`	Name of the operation (e.g. `"transform.log"`).	required
`input_ids`	`tuple[str, ...]`	Sorted tuple of input track UUIDs.	required
`params`	`tuple[tuple[str, Any], ...]`	Sorted tuple of (key, value) parameter pairs.	required

Examples:

>>> p = Provenance("transform.log", ("abc",), (("base", 2),))
>>> p.cache_key()
('transform.log', ('abc',), (('base', 2),))

cache_key ¶

cache_key() -> tuple

Return a hashable key for result caching.

Returns:

Type	Description
`tuple`	Tuple of (operation, input_ids, params).

Examples:

>>> Provenance("op", (), ()).cache_key()
('op', (), ())

Source code in src/seqchain/api/store.py

def cache_key(self) -> tuple:
    """Return a hashable key for result caching.

    Returns:
        Tuple of (operation, input_ids, params).

    Examples:
        >>> Provenance("op", (), ()).cache_key()
        ('op', (), ())
    """
    return (self.operation, self.input_ids, self.params)

Track store¶

StoredTrack ¶

StoredTrack(id: str, track: Track, track_type: str, chroms: dict[str, int], created_at: datetime | None = None, provenance: Provenance | None = None, temp_file: Any = None)

A track stored in the TrackStore with metadata.

Parameters:

Name	Type	Description	Default
`id`	`str`	UUID string.	required
`track`	`Track`	The underlying Track object.	required
`track_type`	`str`	One of `"interval"`, `"table"`, `"signal"`.	required
`chroms`	`dict[str, int]`	Chromosome sizes for serialization iteration.	required
`created_at`	`datetime \| None`	Creation timestamp.	`None`
`provenance`	`Provenance \| None`	How the track was produced, or `None`.	`None`

Examples:

>>> st = StoredTrack("abc", IntervalTrack(TrackLabel("t"), []), "interval", {})
>>> st.id
'abc'

Source code in src/seqchain/api/store.py

def __init__(
    self,
    id: str,
    track: Track,
    track_type: str,
    chroms: dict[str, int],
    created_at: datetime | None = None,
    provenance: Provenance | None = None,
    temp_file: Any = None,
):
    self.id = id
    self.track = track
    self.track_type = track_type
    self.chroms = chroms
    self.created_at = created_at or datetime.now(timezone.utc)
    self.provenance = provenance
    self._temp_file = temp_file

TrackStore ¶

TrackStore()

In-memory store for tracks, with provenance-based caching.

Examples:

>>> store = TrackStore()
>>> tid = store.put(IntervalTrack(TrackLabel("t"), []))
>>> len(store.list_all())
1

Source code in src/seqchain/api/store.py

def __init__(self) -> None:
    self._tracks: dict[str, StoredTrack] = {}
    self._cache: dict[tuple, str] = {}

put ¶

put(track: Track, *, chroms: dict[str, int] | None = None, provenance: Provenance | None = None, temp_file: Any = None) -> str

Store a track and return its UUID.

If a track with identical provenance already exists, returns the existing track's ID instead of creating a duplicate.

Parameters:

Name	Type	Description	Default
`track`	`Track`	The Track to store.	required
`chroms`	`dict[str, int] \| None`	Chromosome sizes for serialization.	`None`
`provenance`	`Provenance \| None`	How the track was produced.	`None`
`temp_file`	`Any`	Reference to keep alive (e.g. BigWig handle).	`None`

Returns:

Type	Description
`str`	UUID string of the stored (or cached) track.

Examples:

>>> store = TrackStore()
>>> store.put(IntervalTrack(TrackLabel("t"), []))
'a1b2c3...'

Source code in src/seqchain/api/store.py

def put(
    self,
    track: Track,
    *,
    chroms: dict[str, int] | None = None,
    provenance: Provenance | None = None,
    temp_file: Any = None,
) -> str:
    """Store a track and return its UUID.

    If a track with identical provenance already exists, returns the
    existing track's ID instead of creating a duplicate.

    Args:
        track: The Track to store.
        chroms: Chromosome sizes for serialization.
        provenance: How the track was produced.
        temp_file: Reference to keep alive (e.g. BigWig handle).

    Returns:
        UUID string of the stored (or cached) track.

    Examples:
        >>> store = TrackStore()
        >>> store.put(IntervalTrack(TrackLabel("t"), []))  # doctest: +SKIP
        'a1b2c3...'
    """
    if provenance is not None:
        key = provenance.cache_key()
        if key in self._cache and self._cache[key] in self._tracks:
            return self._cache[key]

    track_id = str(uuid.uuid4())
    stored = StoredTrack(
        id=track_id,
        track=track,
        track_type=_track_type_name(track),
        chroms=chroms or {},
        provenance=provenance,
        temp_file=temp_file,
    )
    self._tracks[track_id] = stored

    if provenance is not None:
        self._cache[provenance.cache_key()] = track_id

    return track_id

get ¶

get(track_id: str) -> StoredTrack | None

Retrieve a stored track by ID.

Parameters:

Name	Type	Description	Default
`track_id`	`str`	UUID of the track.	required

Returns:

Type	Description
`StoredTrack \| None`	The StoredTrack, or `None` if not found.

Examples:

>>> store = TrackStore()
>>> store.get("nonexistent") is None
True

Source code in src/seqchain/api/store.py

def get(self, track_id: str) -> StoredTrack | None:
    """Retrieve a stored track by ID.

    Args:
        track_id: UUID of the track.

    Returns:
        The StoredTrack, or ``None`` if not found.

    Examples:
        >>> store = TrackStore()
        >>> store.get("nonexistent") is None
        True
    """
    return self._tracks.get(track_id)

delete ¶

delete(track_id: str) -> bool

Remove a track from the store.

Parameters:

Name	Type	Description	Default
`track_id`	`str`	UUID of the track.	required

Returns:

Type	Description
`bool`	`True` if the track was removed, `False` if not found.

Examples:

>>> store = TrackStore()
>>> store.delete("nonexistent")
False

Source code in src/seqchain/api/store.py

def delete(self, track_id: str) -> bool:
    """Remove a track from the store.

    Args:
        track_id: UUID of the track.

    Returns:
        ``True`` if the track was removed, ``False`` if not found.

    Examples:
        >>> store = TrackStore()
        >>> store.delete("nonexistent")
        False
    """
    if track_id in self._tracks:
        stored = self._tracks.pop(track_id)
        if stored.provenance is not None:
            self._cache.pop(stored.provenance.cache_key(), None)
        return True
    return False

list_all ¶

list_all() -> list[StoredTrack]

Return all stored tracks.

Returns:

Type	Description
`list[StoredTrack]`	List of StoredTrack objects, ordered by creation time.

Examples:

>>> TrackStore().list_all()
[]

Source code in src/seqchain/api/store.py

def list_all(self) -> list[StoredTrack]:
    """Return all stored tracks.

    Returns:
        List of StoredTrack objects, ordered by creation time.

    Examples:
        >>> TrackStore().list_all()
        []
    """
    result = []
    for st in self._tracks.values():
        result.append(st)
    result.sort(key=lambda st: st.created_at)
    return result

Genome store¶

StoredGenome ¶

StoredGenome(id: str, genome: Genome, created_at: datetime | None = None)

A genome stored in the GenomeStore with metadata.

Parameters:

Name	Type	Description	Default
`id`	`str`	UUID string.	required
`genome`	`Genome`	The underlying Genome object.	required
`created_at`	`datetime \| None`	Creation timestamp.	`None`

Examples:

>>> from seqchain.io.genome import Genome
>>> g = Genome("test", {}, {"chr1": "ATCG"}, (), {})
>>> sg = StoredGenome("abc", g)
>>> sg.chrom_lengths
{'chr1': 4}

Source code in src/seqchain/api/store.py

def __init__(
    self,
    id: str,
    genome: Genome,
    created_at: datetime | None = None,
):
    self.id = id
    self.genome = genome
    self.created_at = created_at or datetime.now(timezone.utc)

chrom_lengths `property` ¶

chrom_lengths: dict[str, int]

Chromosome name → length mapping.

Returns:

Type	Description
`dict[str, int]`	Dictionary of chromosome sizes.

Examples:

>>> from seqchain.io.genome import Genome
>>> g = Genome("test", {}, {"chr1": "ATCG"}, (), {})
>>> StoredGenome("abc", g).chrom_lengths
{'chr1': 4}

GenomeStore ¶

GenomeStore()

In-memory store for genomes.

Examples:

>>> store = GenomeStore()
>>> len(store.list_all())
0

Source code in src/seqchain/api/store.py

def __init__(self) -> None:
    self._genomes: dict[str, StoredGenome] = {}

put ¶

put(genome: Genome) -> str

Store a genome and return its UUID.

Parameters:

Name	Type	Description	Default
`genome`	`Genome`	The Genome to store.	required

Returns:

Type	Description
`str`	UUID string of the stored genome.

Examples:

>>> from seqchain.io.genome import Genome
>>> store = GenomeStore()
>>> store.put(Genome("test", {}, {"chr1": "ATCG"}, (), {}))
'a1b2c3...'

Source code in src/seqchain/api/store.py

def put(self, genome: Genome) -> str:
    """Store a genome and return its UUID.

    Args:
        genome: The Genome to store.

    Returns:
        UUID string of the stored genome.

    Examples:
        >>> from seqchain.io.genome import Genome
        >>> store = GenomeStore()
        >>> store.put(Genome("test", {}, {"chr1": "ATCG"}, (), {}))  # doctest: +SKIP
        'a1b2c3...'
    """
    genome_id = str(uuid.uuid4())
    self._genomes[genome_id] = StoredGenome(id=genome_id, genome=genome)
    return genome_id

get ¶

get(genome_id: str) -> StoredGenome | None

Retrieve a stored genome by ID.

Parameters:

Name	Type	Description	Default
`genome_id`	`str`	UUID of the genome.	required

Returns:

Type	Description
`StoredGenome \| None`	The StoredGenome, or `None` if not found.

Examples:

>>> GenomeStore().get("nonexistent") is None
True

Source code in src/seqchain/api/store.py

def get(self, genome_id: str) -> StoredGenome | None:
    """Retrieve a stored genome by ID.

    Args:
        genome_id: UUID of the genome.

    Returns:
        The StoredGenome, or ``None`` if not found.

    Examples:
        >>> GenomeStore().get("nonexistent") is None
        True
    """
    return self._genomes.get(genome_id)

delete ¶

delete(genome_id: str) -> bool

Remove a genome from the store.

Parameters:

Name	Type	Description	Default
`genome_id`	`str`	UUID of the genome.	required

Returns:

Type	Description
`bool`	`True` if the genome was removed, `False` if not found.

Examples:

>>> GenomeStore().delete("nonexistent")
False

Source code in src/seqchain/api/store.py

def delete(self, genome_id: str) -> bool:
    """Remove a genome from the store.

    Args:
        genome_id: UUID of the genome.

    Returns:
        ``True`` if the genome was removed, ``False`` if not found.

    Examples:
        >>> GenomeStore().delete("nonexistent")
        False
    """
    if genome_id in self._genomes:
        del self._genomes[genome_id]
        return True
    return False

list_all ¶

list_all() -> list[StoredGenome]

Return all stored genomes.

Returns:

Type	Description
`list[StoredGenome]`	List of StoredGenome objects, ordered by creation time.

Examples:

>>> GenomeStore().list_all()
[]

Source code in src/seqchain/api/store.py

def list_all(self) -> list[StoredGenome]:
    """Return all stored genomes.

    Returns:
        List of StoredGenome objects, ordered by creation time.

    Examples:
        >>> GenomeStore().list_all()
        []
    """
    result = []
    for sg in self._genomes.values():
        result.append(sg)
    result.sort(key=lambda sg: sg.created_at)
    return result

Stores¶

Provenance dataclass ¶

cache_key ¶

Track store¶

StoredTrack ¶

TrackStore ¶

put ¶

get ¶

delete ¶

list_all ¶

Genome store¶

StoredGenome ¶

chrom_lengths property ¶

GenomeStore ¶

put ¶

get ¶

delete ¶

list_all ¶

Provenance `dataclass` ¶

chrom_lengths `property` ¶