Artifacts¶

Content-addressed artifacts: the specs that key them, the store that holds their files and the cache that serves them.

femforge.artifacts ¶

Content-addressed artifacts: a spec keys each one, a store holds its files, and the cache serves, pulls through or produces it.

Modules:

cache –

The typed front door: key a spec, serve it from the local tier, then the remote tier, else produce it under a heartbeat claim and publish to both.
keys –

Artifact identity: a typed content key, one canonical JSON serialization, one hash, HashExclude() honoured at any depth.
kind –

An artifact kind: how its spec keys, what a complete artifact contains, and optionally how to make one; passed by reference, never looked up by name.
local –

The directory-backed store: a staging dir plus one rename publishes atomically, and a sidecar written inside the staging dir is the commit marker.
lock –

In-flight claims: a heartbeat-refreshed marker on one artifact key, held as a directory (mkdir is atomic on NTFS, POSIX, and SMB).
records –

A kind whose artifact is one JSON file holding a typed result: the sweep-point memo, generalized.
sidecar –

The sidecar that commits an artifact (artifact.json: file list, hashes, spec, provenance) and the reference a consumer keeps.
store –

The store contract: a byte store addressed by (kind, key) with a commit-marker sidecar and heartbeat claims, so any backend (a directory, S3) can implement it.

Classes:

ArtifactCache –

Local tier always; an optional remote tier is pulled through on a hit and pushed to after a publish.
ArtifactClaim –

A held claim: refreshed by heartbeat while the holder works, dropped by release.
ArtifactCorrupt –

A stored artifact disagrees with its sidecar.
ArtifactError –

An artifact-store failure.
ArtifactHandle –

A served artifact: where its files are, where they came from, and the record behind them.
ArtifactKey –

A content key (hex SHA-256), phantom-typed by the spec model it was hashed from; name is its readable directory form.
ArtifactKind –

One class of artifact; a module-level constant callers pass by reference.
ArtifactMissing –

The requested artifact is in no tier and no producer was given.
ArtifactOrigin –

Where a served artifact came from.
ArtifactRecord –

The commit marker plus provenance of one stored artifact.
ArtifactRef –

A consumer's pointer to a store artifact: enough to re-fetch it and to know which of its files are re-fetchable.
ArtifactStore –

One storage tier: an artifact exists exactly when its sidecar can be read, and put is first-wins and atomic.
BuiltinKind –

Store path segments of the built-in kinds; the only place these strings are spelled.
DirLock –

A claim held as a directory with an owner file; stale (no heartbeat for stale_after_s) claims are broken by the next claimant.
FileEntry –

One file of an artifact: its store-relative name, size, and content hash.
LocalArtifactStore –

A directory store: <root>/<kind>/<adjective-noun-hash8>/ holds each committed artifact beside its artifact.json sidecar.
LockOwner –

Who holds a claim, since when, and when it last beat.
Produced –

What a producer reports: the files it wrote under its workdir (relative, /-separated) and the tool versions it knows.
Producer –

The cache-miss callback: populate workdir for spec and report the files written.
RecordKind –

Pairs an artifact kind with the result model its single JSON file holds.
RecordOutcome –

A served record: the typed value, where it came from, and its key.
StoredArtifact –

One committed artifact found on disk: its directory and the sidecar inside it.
ToolVersions –

Provenance of the tools that produced an artifact; never part of its key.

Functions:

artifact_dir_of –

The artifact directory path is or lies inside (the nearest ancestor holding a sidecar), or None.
artifact_key –

The content key of one artifact: the kind's own identity when it overrides one, else the hash of the kind name, its producer version and the spec payload.
json_record_kind –

Declare a record kind: name keys spec_type instances to one JSON result_type.
keep_alive –

Beat claim every interval_s from a daemon thread while the block runs, so a live holder never looks stale.
sha256_json –

Hex SHA-256 of json.dumps(payload, sort_keys=True) — the one serialization every artifact key shares.
spec_payload –

spec.model_dump(mode="json") with every HashExclude() field (any depth) plus exclude removed — the dict that is hashed and recorded.
store_root_of –

The store root two levels above the artifact directory path is or lies inside, or None.

Attributes:

ARTIFACT_FILE –

The sidecar filename; an artifact exists in a store exactly when this file does, so it is always written last.

ARTIFACT_FILE `module-attribute` ¶

ARTIFACT_FILE = 'artifact.json'

The sidecar filename; an artifact exists in a store exactly when this file does, so it is always written last.

ArtifactCache ¶

ArtifactCache(local: LocalArtifactStore, remote: ArtifactStore | None = None, *, poll_s: float = 2.0, heartbeat_s: float = HEARTBEAT_S, verify: bool = False, push_required: bool = False)

Local tier always; an optional remote tier is pulled through on a hit and pushed to after a publish.

Parameters:

local (LocalArtifactStore) –

The local tier, always consulted first and always published to.
remote (ArtifactStore | None, default: None ) –

The optional remote tier, pulled through on a hit and pushed to after a publish.
poll_s (float, default: 2.0 ) –

Seconds a producer waits between checks while another claim holds its key.
heartbeat_s (float, default: HEARTBEAT_S ) –

Seconds between refreshes of a held claim while producing.
verify (bool, default: False ) –

Verify file hashes when pulling from the remote tier.
push_required (bool, default: False ) –

Raise when a publish cannot reach the remote tier, instead of logging a warning.

Methods:

key –

The content key of spec under kind.
lookup –

The record of an existing artifact in any tier, without materializing it.
fetch –

A hit-only request: the artifact from whichever tier holds it, else None.
get_or_produce –

Serve spec from a tier, or produce it under an in-flight claim and publish it.
publish –

Commit files that already exist under source (an in-place kind after the fact); first-wins like any publish.
push –

Copy a locally committed artifact to the remote tier; a failure warns unless push_required.
rehydrate –

Materialize one referenced artifact into into again; raises ArtifactMissing when no tier holds it.

Source code in src/femforge/artifacts/cache.py

def __init__(
    self,
    local: LocalArtifactStore,
    remote: ArtifactStore | None = None,
    *,
    poll_s: float = 2.0,
    heartbeat_s: float = HEARTBEAT_S,
    verify: bool = False,
    push_required: bool = False,
) -> None:
    """Serve from ``local`` first, then from ``remote`` when one is given.

    Args:
        local: The local tier, always consulted first and always published to.
        remote: The optional remote tier, pulled through on a hit and pushed to after a publish.
        poll_s: Seconds a producer waits between checks while another claim holds its key.
        heartbeat_s: Seconds between refreshes of a held claim while producing.
        verify: Verify file hashes when pulling from the remote tier.
        push_required: Raise when a publish cannot reach the remote tier, instead of logging a warning.
    """
    self.local: LocalArtifactStore = local
    self.remote: ArtifactStore | None = remote
    self.poll_s: float = poll_s
    self.heartbeat_s: float = heartbeat_s
    self.verify: bool = verify
    self.push_required: bool = push_required

key ¶

key(kind: ArtifactKind[S], spec: S) -> ArtifactKey[S]

The content key of spec under kind.

Source code in src/femforge/artifacts/cache.py

def key[S: BaseModel](self, kind: ArtifactKind[S], spec: S) -> ArtifactKey[S]:
    """The content key of ``spec`` under ``kind``."""
    return artifact_key(kind, spec)

lookup ¶

lookup(kind: ArtifactKind[S], spec: S) -> ArtifactRecord | None

The record of an existing artifact in any tier, without materializing it.

Source code in src/femforge/artifacts/cache.py

def lookup[S: BaseModel](self, kind: ArtifactKind[S], spec: S) -> ArtifactRecord | None:
    """The record of an existing artifact in any tier, without materializing it."""
    key: ArtifactKey[S] = self.key(kind, spec)
    record: ArtifactRecord | None = self.local.record(kind, key)
    if record is None and self.remote is not None:
        record = self.remote.record(kind, key)
    return record

fetch ¶

fetch(kind: ArtifactKind[S], spec: S, into: Path | None) -> ArtifactHandle[S] | None

A hit-only request: the artifact from whichever tier holds it, else None.

Parameters:

kind (ArtifactKind[S]) –

The artifact kind.
spec (S) –

The spec that keys the artifact.
into (Path | None) –

The directory to materialize the files into; None leaves them in the local store.

Source code in src/femforge/artifacts/cache.py

def fetch[S: BaseModel](self, kind: ArtifactKind[S], spec: S, into: Path | None) -> ArtifactHandle[S] | None:
    """A hit-only request: the artifact from whichever tier holds it, else None.

    Args:
        kind: The artifact kind.
        spec: The spec that keys the artifact.
        into: The directory to materialize the files into; None leaves them in the local store.
    """
    return self._fetch_existing(kind, self.key(kind, spec), into)

get_or_produce ¶

get_or_produce(kind: ArtifactKind[S], spec: S, into: Path | None, produce: Producer[S] | None = None) -> ArtifactHandle[S]

Serve spec from a tier, or produce it under an in-flight claim and publish it.

Parameters:

kind (ArtifactKind[S]) –

The artifact kind.
spec (S) –

The spec that keys the artifact.
into (Path | None) –

The directory that receives the files, and where an in_place kind produces; None keeps them in the local store only.
produce (Producer[S] | None, default: None ) –

The producer to run on a miss; None uses the kind's own.

Source code in src/femforge/artifacts/cache.py

def get_or_produce[S: BaseModel](self, kind: ArtifactKind[S], spec: S, into: Path | None, produce: Producer[S] | None = None) -> ArtifactHandle[S]:
    """Serve ``spec`` from a tier, or produce it under an in-flight claim and publish it.

    Args:
        kind: The artifact kind.
        spec: The spec that keys the artifact.
        into: The directory that receives the files, and where an ``in_place`` kind produces; None keeps them in the local store only.
        produce: The producer to run on a miss; None uses the kind's own.
    """
    key: ArtifactKey[S] = self.key(kind, spec)
    producer: Producer[S] | None = produce if produce is not None else kind.produce

    handle: ArtifactHandle[S] | None = self._fetch_existing(kind, key, into)
    if handle is not None:
        return handle

    if producer is None:
        raise ArtifactMissing(f"{kind.name}/{key.name}: not in any tier and the kind has no producer")

    if kind.in_place and into is None:
        raise ArtifactError(f"{kind.name}: an in-place kind needs a directory to produce into")

    authority: ArtifactStore = self.remote if self.remote is not None else self.local
    claim: ArtifactClaim | None = authority.try_claim(kind, key)
    while claim is None:
        time.sleep(self.poll_s)
        handle = self._fetch_existing(kind, key, into)
        if handle is not None:
            return handle

        claim = authority.try_claim(kind, key)

    try:
        handle = self._fetch_existing(kind, key, into)
        if handle is not None:
            return handle

        with keep_alive(claim, self.heartbeat_s):
            return self._produce(
                kind,
                spec,
                key,
                into,
                producer,
            )
    finally:
        claim.release()

publish ¶

publish(kind: ArtifactKind[S], spec: S, source: Path, files: Sequence[str], tools: ToolVersions | None = None) -> ArtifactHandle[S]

Commit files that already exist under source (an in-place kind after the fact); first-wins like any publish.

Parameters:

kind (ArtifactKind[S]) –

The artifact kind.
spec (S) –

The spec that keys the artifact.
source (Path) –

The directory holding the files.
files (Sequence[str]) –

The files to commit, relative to source.
tools (ToolVersions | None, default: None ) –

The tool versions to record; None records only femforge's.

Source code in src/femforge/artifacts/cache.py

def publish[S: BaseModel](self, kind: ArtifactKind[S], spec: S, source: Path, files: Sequence[str], tools: ToolVersions | None = None) -> ArtifactHandle[S]:
    """Commit files that already exist under ``source`` (an in-place kind after the fact); first-wins like any publish.

    Args:
        kind: The artifact kind.
        spec: The spec that keys the artifact.
        source: The directory holding the files.
        files: The files to commit, relative to ``source``.
        tools: The tool versions to record; None records only femforge's.
    """
    key: ArtifactKey[S] = self.key(kind, spec)
    record: ArtifactRecord = ArtifactRecord.build(
        kind=kind.name,
        key=key,
        producer_version=kind.producer_version,
        spec_type=_spec_type_name(spec),
        spec=spec_payload(spec),
        workdir=source,
        files=files,
        tools=tools,
    )
    won: bool = self.local.put(
        kind,
        key,
        source,
        record,
    )
    if won and self.remote is not None:
        self.push(kind, key)
    return self._handle(
        kind,
        key,
        ArtifactOrigin.PRODUCED if won else ArtifactOrigin.LOCAL,
        source,
        record,
    )

push ¶

push(kind: ArtifactKind[S], key: ArtifactKey[S]) -> None

Copy a locally committed artifact to the remote tier; a failure warns unless push_required.

Source code in src/femforge/artifacts/cache.py

def push[S: BaseModel](self, kind: ArtifactKind[S], key: ArtifactKey[S]) -> None:
    """Copy a locally committed artifact to the remote tier; a failure warns unless ``push_required``."""
    if self.remote is None:
        return

    record: ArtifactRecord | None = self.local.record(kind, key)
    if record is None:
        raise ArtifactMissing(f"{kind.name}/{key.name} is not in the local tier")

    try:
        _ = self.remote.put(
            kind,
            key,
            self.local.path(kind, key),
            record,
        )
    except Exception as error:  # noqa: BLE001 - the remote tier is best-effort unless required
        if self.push_required:
            raise ArtifactError(f"{kind.name}/{key.name}: remote publish failed: {error}") from error

        log.bind(kind=kind.name, key=key.short, error=str(error)).warning("Remote publish failed.")

rehydrate ¶

rehydrate(ref: ArtifactRef, into: Path, kind: ArtifactKind[S]) -> ArtifactHandle[S]

Materialize one referenced artifact into into again; raises ArtifactMissing when no tier holds it.

Parameters:

ref (ArtifactRef) –

The reference a consumer recorded.
into (Path) –

The directory to materialize the files into.
kind (ArtifactKind[S]) –

The kind whose name the reference must carry.

Source code in src/femforge/artifacts/cache.py

def rehydrate[S: BaseModel](self, ref: ArtifactRef, into: Path, kind: ArtifactKind[S]) -> ArtifactHandle[S]:
    """Materialize one referenced artifact into ``into`` again; raises [`ArtifactMissing`][femforge.artifacts.store.ArtifactMissing] when no tier holds it.

    Args:
        ref: The reference a consumer recorded.
        into: The directory to materialize the files into.
        kind: The kind whose name the reference must carry.
    """
    if ref.kind != kind.name:
        raise ArtifactError(f"{ref.kind}/{ref.key[:12]}: the reference is not a {kind.name!r} artifact")

    handle: ArtifactHandle[S] | None = self._fetch_existing(kind, ArtifactKey[S](ref.key), into)
    if handle is None:
        raise ArtifactMissing(f"{ref.kind}/{ref.key[:12]} is in no tier")

    return handle

ArtifactClaim ¶

Bases: Protocol

A held claim: refreshed by heartbeat while the holder works, dropped by release.

Methods:

heartbeat –

Refresh the claim so it does not go stale.
release –

Drop the claim.

heartbeat ¶

heartbeat() -> None

Refresh the claim so it does not go stale.

Source code in src/femforge/artifacts/lock.py

def heartbeat(self) -> None:
    """Refresh the claim so it does not go stale."""
    ...

release ¶

release() -> None

Drop the claim.

Source code in src/femforge/artifacts/lock.py

def release(self) -> None:
    """Drop the claim."""
    ...

ArtifactCorrupt ¶

Bases: ArtifactError

A stored artifact disagrees with its sidecar.

ArtifactError ¶

Bases: RuntimeError

An artifact-store failure.

ArtifactHandle `dataclass` ¶

ArtifactHandle(kind: ArtifactKind[S], key: ArtifactKey[S], origin: ArtifactOrigin, path: Path, record: ArtifactRecord)

A served artifact: where its files are, where they came from, and the record behind them.

Attributes:

kind (ArtifactKind[S]) –

The kind of artifact served.
key (ArtifactKey[S]) –

The content key it is stored under.
origin (ArtifactOrigin) –

Whether the local store held it, a remote tier sent it, or a producer just made it.
path (Path) –

The local directory that holds its files.
record (ArtifactRecord) –

The sidecar record that describes it.
files (tuple[str, ...]) –

The artifact-relative names of its files.
ref (ArtifactRef) –

The pointer a consumer records.

kind `instance-attribute` ¶

kind: ArtifactKind[S]

The kind of artifact served.

key `instance-attribute` ¶

key: ArtifactKey[S]

The content key it is stored under.

origin `instance-attribute` ¶

origin: ArtifactOrigin

Whether the local store held it, a remote tier sent it, or a producer just made it.

path `instance-attribute` ¶

path: Path

The local directory that holds its files.

record `instance-attribute` ¶

record: ArtifactRecord

The sidecar record that describes it.

files `property` ¶

files: tuple[str, ...]

The artifact-relative names of its files.

ref `property` ¶

ref: ArtifactRef

The pointer a consumer records.

ArtifactKey ¶

Bases: str

A content key (hex SHA-256), phantom-typed by the spec model it was hashed from; name is its readable directory form.

Attributes:

name (str) –

The <adjective>-<noun>-<hash8> directory name derived from this key.
short (str) –

The first 12 hex chars, for logs.

name `property` ¶

name: str

The <adjective>-<noun>-<hash8> directory name derived from this key.

short `property` ¶

short: str

The first 12 hex chars, for logs.

ArtifactKind `dataclass` ¶

ArtifactKind(name: str, spec_type: type[S], producer_version: int = 1, required_files: tuple[str, ...] = (), produce: Producer[S] | None = None, key: Callable[[S], str] | None = None, in_place: bool = False)

One class of artifact; a module-level constant callers pass by reference.

Attributes:

name (str) –

The store path segment ([a-z0-9_]+; case-insensitive filesystems are in play).
spec_type (type[S]) –

The pydantic model whose instances key this kind.
producer_version (int) –

Bump to invalidate every existing key of this kind when the producer's output semantics change.
required_files (tuple[str, ...]) –

Files a complete artifact must contain; publish refuses a producer that left any out.
produce (Producer[S] | None) –

The default cache-miss producer; None means the caller supplies one per request.
key (Callable[[S], str] | None) –

An identity override (for a kind whose key must coincide with an existing hash); else the generic hash applies.
in_place (bool) –

The producer works in the consumer's own directory and publish copies out of it (a run bundle), rather than in a staging dir.

name `instance-attribute` ¶

name: str

The store path segment ([a-z0-9_]+; case-insensitive filesystems are in play).

spec_type `instance-attribute` ¶

spec_type: type[S]

The pydantic model whose instances key this kind.

producer_version `class-attribute` `instance-attribute` ¶

producer_version: int = 1

Bump to invalidate every existing key of this kind when the producer's output semantics change.

required_files `class-attribute` `instance-attribute` ¶

required_files: tuple[str, ...] = ()

Files a complete artifact must contain; publish refuses a producer that left any out.

produce `class-attribute` `instance-attribute` ¶

produce: Producer[S] | None = None

The default cache-miss producer; None means the caller supplies one per request.

key `class-attribute` `instance-attribute` ¶

key: Callable[[S], str] | None = None

An identity override (for a kind whose key must coincide with an existing hash); else the generic hash applies.

in_place `class-attribute` `instance-attribute` ¶

in_place: bool = False

The producer works in the consumer's own directory and publish copies out of it (a run bundle), rather than in a staging dir.

ArtifactMissing ¶

Bases: ArtifactError

The requested artifact is in no tier and no producer was given.

ArtifactOrigin ¶

Bases: StrEnum

Where a served artifact came from.

Attributes:

LOCAL –

The local store already held it.
REMOTE –

A remote tier sent it, and the local store now holds a copy.
PRODUCED –

A producer made it on this cache miss.

LOCAL `class-attribute` `instance-attribute` ¶

LOCAL = 'local'

The local store already held it.

REMOTE `class-attribute` `instance-attribute` ¶

REMOTE = 'remote'

A remote tier sent it, and the local store now holds a copy.

PRODUCED `class-attribute` `instance-attribute` ¶

PRODUCED = 'produced'

A producer made it on this cache miss.

ArtifactRecord `pydantic-model` ¶

Bases: BaseModel

The commit marker plus provenance of one stored artifact.

Config:

use_attribute_docstrings: True

Fields:

kind (str)
key (str)
name (str)
producer_version (int)
spec_type (str)
spec (dict[str, object])
files (list[FileEntry])
created_at (datetime)
producer_host (str)
producer_pid (int)
tools (ToolVersions)

Validators:

_derive_name

kind `pydantic-field` ¶

kind: str

The artifact kind name.

key `pydantic-field` ¶

key: str

The content key the artifact is stored under.

name `pydantic-field` ¶

name: str = ''

The <adjective>-<noun>-<hash8> store directory name; always derived from key, never read from input.

producer_version `pydantic-field` ¶

producer_version: int

The kind's producer version at publish time.

spec_type `pydantic-field` ¶

spec_type: str

module:QualName of the spec model, for humans reading the store.

spec `pydantic-field` ¶

spec: dict[str, object]

The exact payload that was hashed into key.

files `pydantic-field` ¶

files: list[FileEntry]

Every file the artifact consists of.

created_at `pydantic-field` ¶

created_at: datetime

When it was published (UTC).

producer_host `pydantic-field` ¶

producer_host: str

Hostname of the machine that produced it.

producer_pid `pydantic-field` ¶

producer_pid: int

Process id of the producer, for correlating logs.

tools `pydantic-field` ¶

tools: ToolVersions

Tool versions involved.

file_names `property` ¶

file_names: tuple[str, ...]

The artifact-relative names, in record order.

total_bytes `property` ¶

total_bytes: int

The sum of every file size.

build `classmethod` ¶

build(*, kind: str, key: str, producer_version: int, spec_type: str, spec: dict[str, object], workdir: Path, files: Sequence[str], tools: ToolVersions | None = None) -> ArtifactRecord

Assemble the record for the files under workdir, sizing and hashing each.

Parameters:

kind (str) –

The artifact kind name.
key (str) –

The content key the artifact is stored under.
producer_version (int) –

The kind's producer version.
spec_type (str) –

module:QualName of the spec model.
spec (dict[str, object]) –

The exact payload hashed into key.
workdir (Path) –

The directory holding the artifact's files.
files (Sequence[str]) –

The artifact-relative names of the files.
tools (ToolVersions | None, default: None ) –

The tool versions involved; None records the running femforge release only.

Source code in src/femforge/artifacts/sidecar.py

@classmethod
def build(
    cls,
    *,
    kind: str,
    key: str,
    producer_version: int,
    spec_type: str,
    spec: dict[str, object],
    workdir: Path,
    files: Sequence[str],
    tools: ToolVersions | None = None,
) -> ArtifactRecord:
    """Assemble the record for the files under ``workdir``, sizing and hashing each.

    Args:
        kind: The artifact kind name.
        key: The content key the artifact is stored under.
        producer_version: The kind's producer version.
        spec_type: ``module:QualName`` of the spec model.
        spec: The exact payload hashed into ``key``.
        workdir: The directory holding the artifact's files.
        files: The artifact-relative names of the files.
        tools: The tool versions involved; None records the running femforge release only.
    """
    entries: list[FileEntry] = []
    for name in files:
        path: Path = workdir / name
        entries.append(FileEntry(name=Path(name).as_posix(), size=path.stat().st_size, sha256=sha256_file(path)))
    return cls(
        kind=kind,
        key=key,
        producer_version=producer_version,
        spec_type=spec_type,
        spec=spec,
        files=entries,
        created_at=datetime.now(timezone.utc),
        producer_host=socket.gethostname(),
        producer_pid=os.getpid(),
        tools=tools if tools is not None else ToolVersions(),
    )

verify ¶

verify(directory: Path, *, hashes: bool = False) -> list[str]

The names missing under directory or differing in size (and in content, when hashes).

Source code in src/femforge/artifacts/sidecar.py

def verify(self, directory: Path, *, hashes: bool = False) -> list[str]:
    """The names missing under ``directory`` or differing in size (and in content, when ``hashes``)."""
    problems: list[str] = []
    for entry in self.files:
        path: Path = directory / entry.name
        if not path.is_file() or path.stat().st_size != entry.size:
            problems.append(entry.name)
        elif hashes and sha256_file(path) != entry.sha256:
            problems.append(entry.name)
    return problems

ArtifactRef `pydantic-model` ¶

Bases: BaseModel

A consumer's pointer to a store artifact: enough to re-fetch it and to know which of its files are re-fetchable.

Config:

use_attribute_docstrings: True

Fields:

kind (str)
key (str)
name (str)
origin (ArtifactOrigin)
files (list[str])

Validators:

_derive_name

kind `pydantic-field` ¶

kind: str

The artifact kind name.

key `pydantic-field` ¶

key: str

The content key.

name `pydantic-field` ¶

name: str = ''

The <adjective>-<noun>-<hash8> store directory name; always derived from key, never read from input.

origin `pydantic-field` ¶

origin: ArtifactOrigin

Whether this consumer got a hit, pulled it through from the remote tier, or produced it.

files `pydantic-field` ¶

files: list[str]

The consumer-relative posix paths the artifact materialized.

ArtifactStore ¶

Bases: Protocol

One storage tier: an artifact exists exactly when its sidecar can be read, and put is first-wins and atomic.

Methods:

record –

The committed sidecar of key, or None when this tier lacks it.
exists –

Whether this tier holds key.
fetch –

Materialize every file key lists into into and return its record.
put –

Commit the files record lists under source as key; False when the key already exists.
delete –

Remove key from this tier.
keys –

Every committed key of kind in this tier.
try_claim –

Claim key for production so other producers wait; None when another live claim holds it.

record ¶

record(kind: ArtifactKind[S], key: ArtifactKey[S]) -> ArtifactRecord | None

The committed sidecar of key, or None when this tier lacks it.

Source code in src/femforge/artifacts/store.py

def record[S: BaseModel](self, kind: ArtifactKind[S], key: ArtifactKey[S]) -> ArtifactRecord | None:
    """The committed sidecar of ``key``, or None when this tier lacks it."""
    ...

exists ¶

exists(kind: ArtifactKind[S], key: ArtifactKey[S]) -> bool

Whether this tier holds key.

Source code in src/femforge/artifacts/store.py

def exists[S: BaseModel](self, kind: ArtifactKind[S], key: ArtifactKey[S]) -> bool:
    """Whether this tier holds ``key``."""
    ...

fetch ¶

fetch(kind: ArtifactKind[S], key: ArtifactKey[S], into: Path, *, verify: bool = False) -> ArtifactRecord

Materialize every file key lists into into and return its record.

Parameters:

kind (ArtifactKind[S]) –

The artifact kind.
key (ArtifactKey[S]) –

The artifact's content key.
into (Path) –

The directory to write the files into.
verify (bool, default: False ) –

Also check each file's sha256.

Source code in src/femforge/artifacts/store.py

def fetch[S: BaseModel](self, kind: ArtifactKind[S], key: ArtifactKey[S], into: Path, *, verify: bool = False) -> ArtifactRecord:
    """Materialize every file ``key`` lists into ``into`` and return its record.

    Args:
        kind: The artifact kind.
        key: The artifact's content key.
        into: The directory to write the files into.
        verify: Also check each file's sha256.
    """
    ...

put ¶

put(kind: ArtifactKind[S], key: ArtifactKey[S], source: Path, record: ArtifactRecord) -> bool

Commit the files record lists under source as key; False when the key already exists.

Parameters:

kind (ArtifactKind[S]) –

The artifact kind.
key (ArtifactKey[S]) –

The artifact's content key.
source (Path) –

The directory holding the files.
record (ArtifactRecord) –

The sidecar to commit, listing the files.

Source code in src/femforge/artifacts/store.py

def put[S: BaseModel](self, kind: ArtifactKind[S], key: ArtifactKey[S], source: Path, record: ArtifactRecord) -> bool:
    """Commit the files ``record`` lists under ``source`` as ``key``; False when the key already exists.

    Args:
        kind: The artifact kind.
        key: The artifact's content key.
        source: The directory holding the files.
        record: The sidecar to commit, listing the files.
    """
    ...

delete ¶

delete(kind: ArtifactKind[S], key: ArtifactKey[S]) -> None

Remove key from this tier.

Source code in src/femforge/artifacts/store.py

def delete[S: BaseModel](self, kind: ArtifactKind[S], key: ArtifactKey[S]) -> None:
    """Remove ``key`` from this tier."""
    ...

keys ¶

keys(kind: ArtifactKind[S]) -> Iterator[ArtifactKey[S]]

Every committed key of kind in this tier.

Source code in src/femforge/artifacts/store.py

def keys[S: BaseModel](self, kind: ArtifactKind[S]) -> Iterator[ArtifactKey[S]]:
    """Every committed key of ``kind`` in this tier."""
    ...

try_claim ¶

try_claim(kind: ArtifactKind[S], key: ArtifactKey[S]) -> ArtifactClaim | None

Claim key for production so other producers wait; None when another live claim holds it.

Source code in src/femforge/artifacts/store.py

def try_claim[S: BaseModel](self, kind: ArtifactKind[S], key: ArtifactKey[S]) -> ArtifactClaim | None:
    """Claim ``key`` for production so other producers wait; None when another live claim holds it."""
    ...

BuiltinKind ¶

Bases: StrEnum

Store path segments of the built-in kinds; the only place these strings are spelled.

Attributes:

STEP –

The solve-frame STEP a geometry builds.
ELMER_MESH –

The Elmer mesh files a config meshes into.
RUN_BUNDLE –

A finished run bundle.

STEP `class-attribute` `instance-attribute` ¶

STEP = 'step'

The solve-frame STEP a geometry builds.

ELMER_MESH `class-attribute` `instance-attribute` ¶

ELMER_MESH = 'elmer_mesh'

The Elmer mesh files a config meshes into.

RUN_BUNDLE `class-attribute` `instance-attribute` ¶

RUN_BUNDLE = 'run_bundle'

A finished run bundle.

DirLock ¶

DirLock(path: Path, *, stale_after_s: float = STALE_AFTER_S)

Bases: ArtifactClaim

A claim held as a directory with an owner file; stale (no heartbeat for stale_after_s) claims are broken by the next claimant.

Methods:

owner –

The recorded owner, or None when the claim is not held or its owner file is unreadable.
is_stale –

Whether a held claim has gone stale_after_s without a heartbeat (the directory mtime stands in for a missing owner file).
try_acquire –

Claim the key; False when another live claim holds it. A stale claim is broken and retried once.
heartbeat –

Refresh the owner file's heartbeat; a no-op unless this object holds the claim.
release –

Drop the claim directory: the holder letting go, or a reaper breaking a claim it has proven silent.

Attributes:

held (bool) –

Whether this object currently holds the claim.

Source code in src/femforge/artifacts/lock.py

def __init__(self, path: Path, *, stale_after_s: float = STALE_AFTER_S) -> None:
    """Address the claim directory ``path``, which goes stale after ``stale_after_s`` seconds without a heartbeat."""
    self.path: Path = path
    self.stale_after_s: float = stale_after_s
    self._held: bool = False

held `property` ¶

held: bool

Whether this object currently holds the claim.

owner ¶

owner() -> LockOwner | None

The recorded owner, or None when the claim is not held or its owner file is unreadable.

Source code in src/femforge/artifacts/lock.py

def owner(self) -> LockOwner | None:
    """The recorded owner, or None when the claim is not held or its owner file is unreadable."""
    try:
        return LockOwner.model_validate_json((self.path / OWNER_FILE).read_text(encoding="utf-8"))
    except (OSError, ValidationError, ValueError):
        return None

is_stale ¶

is_stale() -> bool

Whether a held claim has gone stale_after_s without a heartbeat (the directory mtime stands in for a missing owner file).

Source code in src/femforge/artifacts/lock.py

def is_stale(self) -> bool:
    """Whether a held claim has gone ``stale_after_s`` without a heartbeat (the directory mtime stands in for a missing owner file)."""
    if not self.path.is_dir():
        return False

    owner: LockOwner | None = self.owner()
    last: datetime = owner.heartbeat_at if owner is not None else datetime.fromtimestamp(self.path.stat().st_mtime, tz=timezone.utc)
    return (datetime.now(timezone.utc) - last).total_seconds() > self.stale_after_s

try_acquire ¶

try_acquire() -> bool

Claim the key; False when another live claim holds it. A stale claim is broken and retried once.

Source code in src/femforge/artifacts/lock.py

def try_acquire(self) -> bool:
    """Claim the key; False when another live claim holds it. A stale claim is broken and retried once."""
    for _ in range(2):
        try:
            self.path.mkdir(parents=True)
        except FileExistsError:
            if not self.is_stale():
                return False

            shutil.rmtree(self.path, ignore_errors=True)
            continue

        self._write_owner(datetime.now(timezone.utc))
        self._held = True
        return True
    return False

heartbeat ¶

heartbeat() -> None

Refresh the owner file's heartbeat; a no-op unless this object holds the claim.

Source code in src/femforge/artifacts/lock.py

def heartbeat(self) -> None:
    """Refresh the owner file's heartbeat; a no-op unless this object holds the claim."""
    if not self._held:
        return

    owner: LockOwner | None = self.owner()
    self._write_owner(owner.started_at if owner is not None else datetime.now(timezone.utc))

release ¶

release() -> None

Drop the claim directory: the holder letting go, or a reaper breaking a claim it has proven silent.

Source code in src/femforge/artifacts/lock.py

def release(self) -> None:
    """Drop the claim directory: the holder letting go, or a reaper breaking a claim it has proven silent."""
    self._held = False
    shutil.rmtree(self.path, ignore_errors=True)

FileEntry `pydantic-model` ¶

Bases: BaseModel

One file of an artifact: its store-relative name, size, and content hash.

Config:

use_attribute_docstrings: True

Fields:

name (str)
size (int)
sha256 (str)

name `pydantic-field` ¶

name: str

Artifact-relative posix path.

size `pydantic-field` ¶

size: int

Bytes.

sha256 `pydantic-field` ¶

sha256: str

Hex SHA-256 of the content.

LocalArtifactStore ¶

LocalArtifactStore(root: str | Path, *, link: bool = False, claim_stale_s: float = STALE_AFTER_S)

Bases: ArtifactStore

A directory store: <root>/<kind>/<adjective-noun-hash8>/ holds each committed artifact beside its artifact.json sidecar.

Parameters:

root (str | Path) –

The store root, holding one directory per kind.
link (bool, default: False ) –

Hardlink files instead of copying them, where the volume allows.
claim_stale_s (float, default: STALE_AFTER_S ) –

Seconds a claim may go without a heartbeat before another producer breaks it.

Methods:

path –

Where a committed artifact lives.
staging_dir –

A fresh, unique workdir for producing key; put renames it into place.
record –

The committed sidecar, or None when absent, unreadable (then quarantined), or belonging to a different key that shares the name.
exists –

Whether key is committed under kind with a readable sidecar.
fetch –

Materialize every file key lists into into and return its record.
materialize –

Copy (or link) every listed file of a scanned artifact into into; returns the paths written.
put –

Commit source as key by renaming a staging dir into place; returns False, discarding the copy, when the key already exists.
delete –

Remove the artifact directory of key; a missing one is not an error.
committed –

The artifact directories under one kind directory: no staging, claims, or quarantines.
keys –

Every committed key of kind, read from the sidecars.
kinds –

The kind directories present under the root.
read –

The artifact committed in directory, or None when it holds no readable sidecar.
scan –

Every committed artifact under the root (or under one kind), by kind then name.
locate –

The committed artifacts a human reference names: an artifact directory, a full key, a key prefix of 8+ hex chars, or a <adjective>-<noun>-<hash8> name.
try_claim –

Claim key for production; None when another live claim holds it.

Source code in src/femforge/artifacts/local.py

def __init__(self, root: str | Path, *, link: bool = False, claim_stale_s: float = STALE_AFTER_S) -> None:
    """Address the store rooted at ``root``.

    Args:
        root: The store root, holding one directory per kind.
        link: Hardlink files instead of copying them, where the volume allows.
        claim_stale_s: Seconds a claim may go without a heartbeat before another producer breaks it.
    """
    self.root: Path = Path(root)
    self.link: bool = link
    self.claim_stale_s: float = claim_stale_s

path ¶

path(kind: ArtifactKind[S], key: ArtifactKey[S]) -> Path

Where a committed artifact lives.

Source code in src/femforge/artifacts/local.py

def path[S: BaseModel](self, kind: ArtifactKind[S], key: ArtifactKey[S]) -> Path:
    """Where a committed artifact lives."""
    return self.root / kind.name / key.name

staging_dir ¶

staging_dir(kind: ArtifactKind[S], key: ArtifactKey[S]) -> Path

A fresh, unique workdir for producing key; put renames it into place.

Source code in src/femforge/artifacts/local.py

def staging_dir[S: BaseModel](self, kind: ArtifactKind[S], key: ArtifactKey[S]) -> Path:
    """A fresh, unique workdir for producing ``key``; ``put`` renames it into place."""
    directory: Path = self.root / kind.name / STAGING_DIR / f"{key.name}.{socket.gethostname()}.{os.getpid()}.{uuid.uuid4().hex[:8]}"
    directory.mkdir(parents=True)
    return directory

record ¶

record(kind: ArtifactKind[S], key: ArtifactKey[S]) -> ArtifactRecord | None

The committed sidecar, or None when absent, unreadable (then quarantined), or belonging to a different key that shares the name.

Source code in src/femforge/artifacts/local.py

def record[S: BaseModel](self, kind: ArtifactKind[S], key: ArtifactKey[S]) -> ArtifactRecord | None:
    """The committed sidecar, or None when absent, unreadable (then quarantined), or belonging to a different key that shares the name."""
    home: Path = self.path(kind, key)
    if not (home / ARTIFACT_FILE).is_file():
        return None

    record: ArtifactRecord | None = self._sidecar(home)
    if record is None:
        self._quarantine(kind, key)
        return None

    if record.key != key:
        log.bind(
            kind=kind.name,
            key=key.short,
            existing=record.key[:12],
            name=key.name,
        ).error("Artifact name collision.")
        return None

    return record

exists ¶

exists(kind: ArtifactKind[S], key: ArtifactKey[S]) -> bool

Whether key is committed under kind with a readable sidecar.

Source code in src/femforge/artifacts/local.py

def exists[S: BaseModel](self, kind: ArtifactKind[S], key: ArtifactKey[S]) -> bool:
    """Whether ``key`` is committed under ``kind`` with a readable sidecar."""
    return self.record(kind, key) is not None

fetch ¶

fetch(kind: ArtifactKind[S], key: ArtifactKey[S], into: Path, *, verify: bool = False) -> ArtifactRecord

Materialize every file key lists into into and return its record.

Parameters:

kind (ArtifactKind[S]) –

The artifact kind.
key (ArtifactKey[S]) –

The artifact's content key.
into (Path) –

The directory to write the files into.
verify (bool, default: False ) –

Also check each file's sha256; sizes are always checked.

Source code in src/femforge/artifacts/local.py

def fetch[S: BaseModel](self, kind: ArtifactKind[S], key: ArtifactKey[S], into: Path, *, verify: bool = False) -> ArtifactRecord:
    """Materialize every file ``key`` lists into ``into`` and return its record.

    Args:
        kind: The artifact kind.
        key: The artifact's content key.
        into: The directory to write the files into.
        verify: Also check each file's sha256; sizes are always checked.
    """
    record: ArtifactRecord | None = self.record(kind, key)
    if record is None:
        raise ArtifactMissing(f"{kind.name}/{key.name} is not in {self.root}")

    _ = self._copy_out(self.path(kind, key), record, into, verify=verify)
    return record

materialize ¶

materialize(artifact: StoredArtifact, into: Path, *, verify: bool = False) -> list[Path]

Copy (or link) every listed file of a scanned artifact into into; returns the paths written.

Parameters:

artifact (StoredArtifact) –

The scanned artifact.
into (Path) –

The directory to write the files into.
verify (bool, default: False ) –

Also check each file's sha256; sizes are always checked.

Source code in src/femforge/artifacts/local.py

def materialize(self, artifact: StoredArtifact, into: Path, *, verify: bool = False) -> list[Path]:
    """Copy (or link) every listed file of a scanned artifact into ``into``; returns the paths written.

    Args:
        artifact: The scanned artifact.
        into: The directory to write the files into.
        verify: Also check each file's sha256; sizes are always checked.
    """
    return self._copy_out(artifact.path, artifact.record, into, verify=verify)

put ¶

put(kind: ArtifactKind[S], key: ArtifactKey[S], source: Path, record: ArtifactRecord) -> bool

Commit source as key by renaming a staging dir into place; returns False, discarding the copy, when the key already exists.

Parameters:

kind (ArtifactKind[S]) –

The artifact kind.
key (ArtifactKey[S]) –

The artifact's content key.
source (Path) –

A staging dir, pruned to the listed files, or any other directory whose listed files are first linked or copied into one.
record (ArtifactRecord) –

The sidecar to commit, listing the files.

Source code in src/femforge/artifacts/local.py

def put[S: BaseModel](self, kind: ArtifactKind[S], key: ArtifactKey[S], source: Path, record: ArtifactRecord) -> bool:
    """Commit ``source`` as ``key`` by renaming a staging dir into place; returns False, discarding the copy, when the key already exists.

    Args:
        kind: The artifact kind.
        key: The artifact's content key.
        source: A staging dir, pruned to the listed files, or any other directory whose listed files are first linked or copied into one.
        record: The sidecar to commit, listing the files.
    """
    final: Path = self.path(kind, key)
    existing: ArtifactRecord | None = self._sidecar(final)
    if existing is not None and existing.key != key:
        raise ArtifactError(f"{kind.name}/{key.name}: name collision with key {existing.key[:12]}; move that artifact aside")

    if existing is not None:
        if self._is_staging(kind, source):
            shutil.rmtree(source, ignore_errors=True)
        return False

    if self._is_staging(kind, source):
        staging: Path = source
        keep: set[Path] = {(staging / entry.name).resolve() for entry in record.files}
        for path in sorted(staging.rglob("*"), reverse=True):
            if path.is_file() and path.resolve() not in keep and path.name != ARTIFACT_FILE:
                path.unlink()
            elif path.is_dir() and not any(path.iterdir()):
                path.rmdir()
    else:
        staging = self.staging_dir(kind, key)
        for entry in record.files:
            self._materialize(source / entry.name, staging / entry.name)

    write_atomic(staging / ARTIFACT_FILE, record.model_dump_json(indent=2))

    try:
        _rename_dir(staging, final)
    except OSError:
        # Another publisher won the rename; content-addressed, so its copy is as good as ours.
        shutil.rmtree(staging, ignore_errors=True)
        return False

    log.bind(
        kind=kind.name,
        key=key.short,
        name=key.name,
        files=len(record.files),
        bytes=record.total_bytes,
    ).info("Artifact published.")
    return True

delete ¶

delete(kind: ArtifactKind[S], key: ArtifactKey[S]) -> None

Remove the artifact directory of key; a missing one is not an error.

Source code in src/femforge/artifacts/local.py

def delete[S: BaseModel](self, kind: ArtifactKind[S], key: ArtifactKey[S]) -> None:
    """Remove the artifact directory of ``key``; a missing one is not an error."""
    shutil.rmtree(self.path(kind, key), ignore_errors=True)

committed ¶

committed(kind: str) -> Iterator[Path]

The artifact directories under one kind directory: no staging, claims, or quarantines.

Source code in src/femforge/artifacts/local.py

def committed(self, kind: str) -> Iterator[Path]:
    """The artifact directories under one kind directory: no staging, claims, or quarantines."""
    home: Path = self.root / kind
    if not home.is_dir():
        return

    for entry in sorted(home.iterdir()):
        if entry.is_dir() and not entry.name.startswith(".") and not entry.name.endswith(LOCK_SUFFIX) and CORRUPT_MARK not in entry.name:
            yield entry

keys ¶

keys(kind: ArtifactKind[S]) -> Iterator[ArtifactKey[S]]

Every committed key of kind, read from the sidecars.

Source code in src/femforge/artifacts/local.py

def keys[S: BaseModel](self, kind: ArtifactKind[S]) -> Iterator[ArtifactKey[S]]:
    """Every committed key of ``kind``, read from the sidecars."""
    for entry in self.committed(kind.name):
        record: ArtifactRecord | None = self._sidecar(entry)
        if record is not None and record.kind == kind.name:
            yield ArtifactKey[S](record.key)

kinds ¶

kinds() -> list[str]

The kind directories present under the root.

Source code in src/femforge/artifacts/local.py

def kinds(self) -> list[str]:
    """The kind directories present under the root."""
    if not self.root.is_dir():
        return []

    return sorted(entry.name for entry in self.root.iterdir() if entry.is_dir() and not entry.name.startswith("."))

read ¶

read(directory: Path) -> StoredArtifact | None

The artifact committed in directory, or None when it holds no readable sidecar.

Source code in src/femforge/artifacts/local.py

def read(self, directory: Path) -> StoredArtifact | None:
    """The artifact committed in ``directory``, or None when it holds no readable sidecar."""
    record: ArtifactRecord | None = self._sidecar(directory)
    return StoredArtifact(path=directory, record=record) if record is not None else None

scan ¶

scan(kind: str | None = None) -> Iterator[StoredArtifact]

Every committed artifact under the root (or under one kind), by kind then name.

Source code in src/femforge/artifacts/local.py

def scan(self, kind: str | None = None) -> Iterator[StoredArtifact]:
    """Every committed artifact under the root (or under one kind), by kind then name."""
    for kind_name in self.kinds() if kind is None else [kind]:
        for directory in self.committed(kind_name):
            artifact: StoredArtifact | None = self.read(directory)
            if artifact is not None and artifact.kind == kind_name:
                yield artifact

locate ¶

locate(ref: str) -> list[StoredArtifact]

The committed artifacts a human reference names: an artifact directory, a full key, a key prefix of 8+ hex chars, or a <adjective>-<noun>-<hash8> name.

Source code in src/femforge/artifacts/local.py

def locate(self, ref: str) -> list[StoredArtifact]:
    """The committed artifacts a human reference names: an artifact directory, a full key, a key prefix of 8+ hex chars, or a ``<adjective>-<noun>-<hash8>`` name."""
    directory = Path(ref)
    if (directory / ARTIFACT_FILE).is_file():
        artifact: StoredArtifact | None = self.read(directory)
        return [artifact] if artifact is not None else []

    text: str = ref.strip().lower()
    is_hex: bool = _HEX_REF.fullmatch(text) is not None

    matches: list[StoredArtifact] = []
    for kind_name in self.kinds():
        for candidate in self._candidates(kind_name, text, is_hex):
            found: StoredArtifact | None = self.read(candidate)
            if found is None or found.kind != kind_name:
                continue

            if not is_hex or found.key.startswith(text):
                matches.append(found)

    return matches

try_claim ¶

try_claim(kind: ArtifactKind[S], key: ArtifactKey[S]) -> ArtifactClaim | None

Claim key for production; None when another live claim holds it.

Source code in src/femforge/artifacts/local.py

def try_claim[S: BaseModel](self, kind: ArtifactKind[S], key: ArtifactKey[S]) -> ArtifactClaim | None:
    """Claim ``key`` for production; None when another live claim holds it."""
    lock = DirLock(self.root / kind.name / f"{key.name}{LOCK_SUFFIX}", stale_after_s=self.claim_stale_s)
    return lock if lock.try_acquire() else None

LockOwner `pydantic-model` ¶

Bases: BaseModel

Who holds a claim, since when, and when it last beat.

Config:

use_attribute_docstrings: True

Fields:

host (str)
pid (int)
started_at (datetime)
heartbeat_at (datetime)

host `pydantic-field` ¶

host: str

The machine the claiming process runs on.

pid `pydantic-field` ¶

pid: int

The claiming process's id on that machine.

started_at `pydantic-field` ¶

started_at: datetime

When the claim was taken (UTC).

heartbeat_at `pydantic-field` ¶

heartbeat_at: datetime

When the claim last beat (UTC); a claim silent for longer than its stale limit is broken.

Produced `dataclass` ¶

Produced(files: tuple[str, ...], tools: ToolVersions | None = None)

What a producer reports: the files it wrote under its workdir (relative, /-separated) and the tool versions it knows.

Attributes:

files (tuple[str, ...]) –

The files written, relative to the workdir and /-separated.
tools (ToolVersions | None) –

The tool versions that made them; None when the producer knows none.

files `instance-attribute` ¶

files: tuple[str, ...]

The files written, relative to the workdir and /-separated.

tools `class-attribute` `instance-attribute` ¶

tools: ToolVersions | None = None

The tool versions that made them; None when the producer knows none.

Producer ¶

Bases: Protocol

The cache-miss callback: populate workdir for spec and report the files written.

RecordKind `dataclass` ¶

RecordKind(kind: ArtifactKind[S], result_type: type[R])

Pairs an artifact kind with the result model its single JSON file holds.

Methods:

key –

The content key of spec under this kind.
get_or_produce –

The stored value for spec, else produce(spec) written and published under a claim.
load –

The stored value for spec, or None when no tier holds it.
load_all –

Every value of this kind in the local tier, with the record that identifies it.

Attributes:

kind (ArtifactKind[S]) –

The artifact kind the records are stored as.
result_type (type[R]) –

The model the record file is read back into.

kind `instance-attribute` ¶

kind: ArtifactKind[S]

The artifact kind the records are stored as.

result_type `instance-attribute` ¶

result_type: type[R]

The model the record file is read back into.

key ¶

key(spec: S) -> ArtifactKey[S]

The content key of spec under this kind.

Source code in src/femforge/artifacts/records.py

def key(self, spec: S) -> ArtifactKey[S]:
    """The content key of ``spec`` under this kind."""
    return artifact_key(self.kind, spec)

get_or_produce ¶

get_or_produce(cache: ArtifactCache, spec: S, produce: Callable[[S], R]) -> RecordOutcome[S, R]

The stored value for spec, else produce(spec) written and published under a claim.

Parameters:

cache (ArtifactCache) –

The cache to serve from and publish to.
spec (S) –

The spec that keys the record.
produce (Callable[[S], R]) –

Computes the value on a miss.

Source code in src/femforge/artifacts/records.py

def get_or_produce(self, cache: ArtifactCache, spec: S, produce: Callable[[S], R]) -> RecordOutcome[S, R]:
    """The stored value for ``spec``, else ``produce(spec)`` written and published under a claim.

    Args:
        cache: The cache to serve from and publish to.
        spec: The spec that keys the record.
        produce: Computes the value on a miss.
    """

    def _write(spec: S, workdir: Path) -> Produced:
        value: R = produce(spec)
        write_atomic(workdir / RECORD_FILE, value.model_dump_json(indent=2))
        return Produced(files=(RECORD_FILE,))

    handle: ArtifactHandle[S] = cache.get_or_produce(
        self.kind,
        spec,
        into=None,
        produce=_write,
    )
    return RecordOutcome(value=self._read(handle), origin=handle.origin, key=handle.key)

load ¶

load(cache: ArtifactCache, spec: S) -> R | None

The stored value for spec, or None when no tier holds it.

Source code in src/femforge/artifacts/records.py

def load(self, cache: ArtifactCache, spec: S) -> R | None:
    """The stored value for ``spec``, or None when no tier holds it."""
    handle: ArtifactHandle[S] | None = cache.fetch(self.kind, spec, into=None)
    return self._read(handle) if handle is not None else None

load_all ¶

load_all(cache: ArtifactCache) -> Iterator[tuple[ArtifactRecord, R]]

Every value of this kind in the local tier, with the record that identifies it.

Source code in src/femforge/artifacts/records.py

def load_all(self, cache: ArtifactCache) -> Iterator[tuple[ArtifactRecord, R]]:
    """Every value of this kind in the local tier, with the record that identifies it."""
    for key in cache.local.keys(self.kind):
        record: ArtifactRecord | None = cache.local.record(self.kind, key)
        if record is None:  # pragma: no cover - keys() only yields keys whose sidecar just read cleanly
            continue

        yield record, self.result_type.model_validate_json((cache.local.path(self.kind, key) / RECORD_FILE).read_text(encoding="utf-8"))

RecordOutcome `dataclass` ¶

RecordOutcome(value: R, origin: ArtifactOrigin, key: ArtifactKey[S])

A served record: the typed value, where it came from, and its key.

Attributes:

value (R) –

The typed record.
origin (ArtifactOrigin) –

Whether the local store held it, a remote tier sent it, or produce just made it.
key (ArtifactKey[S]) –

The content key it is stored under.

value `instance-attribute` ¶

value: R

The typed record.

origin `instance-attribute` ¶

origin: ArtifactOrigin

Whether the local store held it, a remote tier sent it, or produce just made it.

key `instance-attribute` ¶

key: ArtifactKey[S]

The content key it is stored under.

StoredArtifact `dataclass` ¶

StoredArtifact(path: Path, record: ArtifactRecord)

One committed artifact found on disk: its directory and the sidecar inside it.

Attributes:

path (Path) –

The artifact's directory.
record (ArtifactRecord) –

The sidecar inside it.
kind (str) –

The artifact kind name.
key (str) –

The content key the artifact is stored under.
name (str) –

The artifact's store directory name.

path `instance-attribute` ¶

path: Path

The artifact's directory.

record `instance-attribute` ¶

record: ArtifactRecord

The sidecar inside it.

kind `property` ¶

kind: str

The artifact kind name.

key `property` ¶

key: str

The content key the artifact is stored under.

name `property` ¶

name: str

The artifact's store directory name.

ToolVersions `pydantic-model` ¶

Bases: BaseModel

Provenance of the tools that produced an artifact; never part of its key.

Config:

use_attribute_docstrings: True

Fields:

femforge (str)
code_sha (str | None)
gmsh (str | None)
elmer (str | None)

femforge `pydantic-field` ¶

femforge: str = femforge.__version__

The femforge release that produced it.

code_sha `pydantic-field` ¶

code_sha: str | None = None

The producing checkout's commit (-dirty when uncommitted changes were present).

gmsh `pydantic-field` ¶

gmsh: str | None = None

gmsh version, when meshing was involved.

elmer `pydantic-field` ¶

elmer: str | None = None

ElmerSolver version, when a solve was involved.

artifact_dir_of ¶

artifact_dir_of(path: Path) -> Path | None

The artifact directory path is or lies inside (the nearest ancestor holding a sidecar), or None.

Source code in src/femforge/artifacts/local.py

def artifact_dir_of(path: Path) -> Path | None:
    """The artifact directory ``path`` is or lies inside (the nearest ancestor holding a sidecar), or None."""
    for candidate in (path, *path.parents):
        if (candidate / ARTIFACT_FILE).is_file():
            return candidate

    return None

artifact_key ¶

artifact_key(kind: ArtifactKind[S], spec: S) -> ArtifactKey[S]

The content key of one artifact: the kind's own identity when it overrides one, else the hash of the kind name, its producer version and the spec payload.

Source code in src/femforge/artifacts/keys.py

def artifact_key[S: BaseModel](kind: ArtifactKind[S], spec: S) -> ArtifactKey[S]:
    """The content key of one artifact: the kind's own identity when it overrides one, else the hash of the kind name, its producer version and the spec payload."""
    if not isinstance(spec, kind.spec_type):
        raise TypeError(f"{kind.name}: expected a {kind.spec_type.__name__} spec, got {type(spec).__name__}")

    if kind.key is not None:
        return ArtifactKey[S](kind.key(spec))

    return ArtifactKey[S](sha256_json({"kind": kind.name, "producer_version": kind.producer_version, "spec": spec_payload(spec)}))

json_record_kind ¶

json_record_kind(name: str, *, spec_type: type[S], result_type: type[R], producer_version: int = 1) -> RecordKind[S, R]

Declare a record kind: name keys spec_type instances to one JSON result_type.

Parameters:

name (str) –

The kind name, which is also its store directory.
spec_type (type[S]) –

The spec model that keys a record.
result_type (type[R]) –

The model the record's JSON file holds.
producer_version (int, default: 1 ) –

Bump to invalidate every stored record of this kind.

Source code in src/femforge/artifacts/records.py

def json_record_kind[S: BaseModel, R: BaseModel](name: str, *, spec_type: type[S], result_type: type[R], producer_version: int = 1) -> RecordKind[S, R]:
    """Declare a record kind: ``name`` keys ``spec_type`` instances to one JSON ``result_type``.

    Args:
        name: The kind name, which is also its store directory.
        spec_type: The spec model that keys a record.
        result_type: The model the record's JSON file holds.
        producer_version: Bump to invalidate every stored record of this kind.
    """
    return RecordKind(
        kind=ArtifactKind(
            name=name,
            spec_type=spec_type,
            producer_version=producer_version,
            required_files=(RECORD_FILE,),
        ),
        result_type=result_type,
    )

keep_alive ¶

keep_alive(claim: ArtifactClaim, interval_s: float = HEARTBEAT_S) -> Iterator[None]

Beat claim every interval_s from a daemon thread while the block runs, so a live holder never looks stale.

Source code in src/femforge/artifacts/lock.py

@contextmanager
def keep_alive(claim: ArtifactClaim, interval_s: float = HEARTBEAT_S) -> Iterator[None]:
    """Beat ``claim`` every ``interval_s`` from a daemon thread while the block runs, so a live holder never looks stale."""
    stop = threading.Event()

    def _beat() -> None:
        while not stop.wait(interval_s):
            try:
                claim.heartbeat()
            except OSError as error:
                log.bind(error=str(error)).warning("Claim heartbeat failed.")

    # Copied context: a failed beat's warning keeps the caller's run_id and telemetry route.
    thread = threading.Thread(
        target=contextvars.copy_context().run,
        args=(_beat,),
        name="artifact-claim-heartbeat",
        daemon=True,
    )
    thread.start()
    try:
        yield
    finally:
        stop.set()
        thread.join(timeout=interval_s + 1.0)

sha256_json ¶

sha256_json(payload: object) -> str

Hex SHA-256 of json.dumps(payload, sort_keys=True) — the one serialization every artifact key shares.

Source code in src/femforge/artifacts/keys.py

def sha256_json(payload: object) -> str:
    """Hex SHA-256 of ``json.dumps(payload, sort_keys=True)`` — the one serialization every artifact key shares."""
    return hashlib.sha256(json.dumps(payload, sort_keys=True).encode("utf-8")).hexdigest()

spec_payload ¶

spec_payload(spec: BaseModel, *, exclude: Mapping[str | int, object] | None = None) -> dict[str, object]

spec.model_dump(mode="json") with every HashExclude() field (any depth) plus exclude removed — the dict that is hashed and recorded.

Source code in src/femforge/artifacts/keys.py

def spec_payload(spec: BaseModel, *, exclude: Mapping[str | int, object] | None = None) -> dict[str, object]:
    """``spec.model_dump(mode="json")`` with every ``HashExclude()`` field (any depth) plus ``exclude`` removed — the dict that is hashed and recorded."""
    excluded: ExcludeSpec = merge_exclude(hash_exclude_spec(spec), exclude)
    return spec.model_dump(mode="json", exclude=excluded or None)  # pyright: ignore[reportArgumentType]  # pydantic accepts nested int/str keyed specs

store_root_of ¶

store_root_of(path: Path) -> Path | None

The store root two levels above the artifact directory path is or lies inside, or None.

Source code in src/femforge/artifacts/local.py

def store_root_of(path: Path) -> Path | None:
    """The store root two levels above the artifact directory ``path`` is or lies inside, or None."""
    directory: Path | None = artifact_dir_of(path)
    return directory.parent.parent if directory is not None else None

Artifacts¶

femforge.artifacts ¶

ARTIFACT_FILE module-attribute ¶

ArtifactCache ¶

key ¶

lookup ¶

fetch ¶

get_or_produce ¶

publish ¶

push ¶

rehydrate ¶

ArtifactClaim ¶

heartbeat ¶

release ¶

ArtifactCorrupt ¶

ArtifactError ¶

ArtifactHandle dataclass ¶

kind instance-attribute ¶

key instance-attribute ¶

origin instance-attribute ¶

path instance-attribute ¶

record instance-attribute ¶

files property ¶

ref property ¶

ArtifactKey ¶

name property ¶

short property ¶

ArtifactKind dataclass ¶

name instance-attribute ¶

spec_type instance-attribute ¶

producer_version class-attribute instance-attribute ¶

required_files class-attribute instance-attribute ¶

produce class-attribute instance-attribute ¶

key class-attribute instance-attribute ¶

in_place class-attribute instance-attribute ¶

ArtifactMissing ¶

ArtifactOrigin ¶

LOCAL class-attribute instance-attribute ¶

REMOTE class-attribute instance-attribute ¶

PRODUCED class-attribute instance-attribute ¶

ArtifactRecord pydantic-model ¶

kind pydantic-field ¶

key pydantic-field ¶

name pydantic-field ¶

producer_version pydantic-field ¶

spec_type pydantic-field ¶

spec pydantic-field ¶

files pydantic-field ¶

created_at pydantic-field ¶

producer_host pydantic-field ¶

producer_pid pydantic-field ¶

tools pydantic-field ¶

file_names property ¶

total_bytes property ¶

build classmethod ¶

verify ¶

ArtifactRef pydantic-model ¶

kind pydantic-field ¶

key pydantic-field ¶

name pydantic-field ¶

origin pydantic-field ¶

files pydantic-field ¶

ArtifactStore ¶

record ¶

exists ¶

fetch ¶

put ¶

delete ¶

keys ¶

try_claim ¶

BuiltinKind ¶

STEP class-attribute instance-attribute ¶

ELMER_MESH class-attribute instance-attribute ¶

RUN_BUNDLE class-attribute instance-attribute ¶

DirLock ¶

held property ¶

owner ¶

is_stale ¶

try_acquire ¶

heartbeat ¶

ARTIFACT_FILE `module-attribute` ¶

ArtifactHandle `dataclass` ¶

kind `instance-attribute` ¶

key `instance-attribute` ¶

origin `instance-attribute` ¶

path `instance-attribute` ¶

record `instance-attribute` ¶

files `property` ¶

ref `property` ¶

name `property` ¶

short `property` ¶

ArtifactKind `dataclass` ¶

name `instance-attribute` ¶

spec_type `instance-attribute` ¶

producer_version `class-attribute` `instance-attribute` ¶

required_files `class-attribute` `instance-attribute` ¶

produce `class-attribute` `instance-attribute` ¶

key `class-attribute` `instance-attribute` ¶

in_place `class-attribute` `instance-attribute` ¶

LOCAL `class-attribute` `instance-attribute` ¶

REMOTE `class-attribute` `instance-attribute` ¶

PRODUCED `class-attribute` `instance-attribute` ¶

ArtifactRecord `pydantic-model` ¶

kind `pydantic-field` ¶

key `pydantic-field` ¶

name `pydantic-field` ¶

producer_version `pydantic-field` ¶

spec_type `pydantic-field` ¶

spec `pydantic-field` ¶

files `pydantic-field` ¶

created_at `pydantic-field` ¶

producer_host `pydantic-field` ¶

producer_pid `pydantic-field` ¶

tools `pydantic-field` ¶

file_names `property` ¶

total_bytes `property` ¶

build `classmethod` ¶

ArtifactRef `pydantic-model` ¶

kind `pydantic-field` ¶

key `pydantic-field` ¶

name `pydantic-field` ¶

origin `pydantic-field` ¶

files `pydantic-field` ¶

STEP `class-attribute` `instance-attribute` ¶

ELMER_MESH `class-attribute` `instance-attribute` ¶

RUN_BUNDLE `class-attribute` `instance-attribute` ¶

held `property` ¶

FileEntry `pydantic-model` ¶

name `pydantic-field` ¶

size `pydantic-field` ¶

sha256 `pydantic-field` ¶

LockOwner `pydantic-model` ¶

host `pydantic-field` ¶

pid `pydantic-field` ¶

started_at `pydantic-field` ¶

heartbeat_at `pydantic-field` ¶

Produced `dataclass` ¶

files `instance-attribute` ¶

tools `class-attribute` `instance-attribute` ¶

RecordKind `dataclass` ¶

kind `instance-attribute` ¶

result_type `instance-attribute` ¶

RecordOutcome `dataclass` ¶

value `instance-attribute` ¶

origin `instance-attribute` ¶

key `instance-attribute` ¶

StoredArtifact `dataclass` ¶

path `instance-attribute` ¶

record `instance-attribute` ¶

kind `property` ¶

key `property` ¶

name `property` ¶

ToolVersions `pydantic-model` ¶

femforge `pydantic-field` ¶

code_sha `pydantic-field` ¶

gmsh `pydantic-field` ¶

elmer `pydantic-field` ¶