Skip to main content

Catalog

The catalog holds versioned values under paths. A task publishes a value at a path, other tasks read the current version at that path, and an execution can be re-run when a newer version lands.

A value is anything that can be passed to a task: an asset, a data structure holding assets, a reference to data in an external system, a plain number.

import coflux as cf
from pathlib import Path

datasets = cf.Catalog("datasets/{name}")
models = cf.Catalog("models/{name}")


@cf.task()
def build_dataset() -> int:
    Path("customers.parquet").write_bytes(fetch_customers())
    return datasets.at(name="customers").publish(cf.asset("customers.parquet"))


@cf.task(memo=True)
def train(dataset: cf.Asset) -> int:
    paths = dataset.restore()
    model = fit(paths["customers.parquet"])
    return models.at(name="churn").publish({"weights": cf.asset(model), "auc": model.auc})

A path is a slash-separated name like datasets/customers. Each publish appends a version: which value, who published it, and when. Versions are never modified or deleted.

Handles​

cf.Catalog(path) is a handle to a path: declared once, usually at module level, and used from any target. A {placeholder} in the path stands for a part to be filled in later. at() returns another handle with the placeholder bound, and the same type, so one declaration covers a family of paths:

models = cf.Catalog("models/{name}")
churn = models.at(name="churn")     # Catalog('models/churn')

Nothing round-trips until a handle is used, and a handle with a placeholder still unbound refuses to be: publish(), current() and next() all raise. An invalid path, or a value that wouldn't make one, is a ValueError where the handle is declared or bound rather than a request the server refuses.

A handle isn't a value: pass the path, or the value it holds, to a task, not the handle.

Publishing​

handle.publish(value) publishes a value at the handle's path and returns the new version's number.

configs = cf.Catalog("configs/{name}")
number = configs.at(name="training").publish({"threshold": 0.7, "epochs": 20})  # 6

Facts about a publish — a metric, what it was built from — go in the value, alongside the thing itself. There is no separate metadata: a dict holding an asset and its scores is one value, rendered as such in Studio, and a reader gets both from current().

Publishing what is already the latest version — the same value — writes nothing and returns the existing version. Values are content-addressed, so this is exact, and it's what makes a publish safe to run again after a suspension: a body that publishes and then suspends republishes on resumption without creating a duplicate. A change to any part of the value is a new version.

Versions are numbered per path, once, across every workspace: models/churn@6 names exactly one version wherever you are.

What a version pins​

The catalog pins the value, not what the value points at. A data structure or an asset is immutable by construction. A handle — an execution, an input — resolves to whatever it resolves to when it's read, which can change as the execution is retried or the input answered. A locator for external data (a table name, a URI) is only as stable as the data behind it, so publish a snapshot identifier or a content hash alongside it when lineage needs to be exact:

tables = cf.Catalog("tables/{name}")
tables.at(name="customers").publish({"table": "analytics.customers", "snapshot": 4182})

A value can also be published from outside a run, with the CLI — a JSON document, or an existing asset:

coflux catalog publish configs/training '{"threshold": 0.7}'
coflux catalog publish models/churn --asset <asset-id>

Reading​

handle.current() reads the value at the handle's path.

churn = models.at(name="churn")
churn.current()    # the current value

An execution sees the catalog as it was when it started. Every execution is pinned to the catalog at the moment it's assigned to a worker, and current() answers from that snapshot, so repeated reads agree with each other and with reads of other paths. There are two exceptions. Anything the execution's own run has published since is visible: a parent that waits on a child which publishes sees the child's version. And a path that had nothing when the execution started shows its first version once one lands, since the alternative is never answering.

That's why it's current rather than latest: a version published after the execution started isn't current for it. next(), below, is how to wait for one.

current() on a path with nothing published yet waits for the first publish, following the suspense rule every read follows: it blocks outside a cf.suspense scope, holding the worker slot, and suspends inside one.

Reads are recorded, so Studio shows which versions each execution used alongside its other dependencies, and which it published.

Snapshots across attempts​

A step's later attempts keep the snapshot of the attempt before them. A retry, a re-run from Studio or the CLI, and the execution that resumes a suspended step all see what the previous attempt saw, so re-running reproduces a run rather than quietly picking up whatever has been published since. Two kinds of attempt start afresh instead: the successor of a next(), which exists to see the newer version, and each iteration of a recurring target.

The snapshot can also be chosen. A run can be submitted as of a version — the catalog as it was when that version was published, every path included — and a step can be re-run against a version, or against the latest:

coflux submit --catalog models/churn@4 myapp/evaluate
coflux runs rerun --catalog models/churn@4 R1a2b3:1
coflux runs rerun --catalog latest R1a2b3:1

A run submitted as of a version gives every execution in it that snapshot, including steps it schedules. A version has to exist and be visible from the workspace the run is in.

Typed values​

A handle can declare the type of the values at its path. With Pydantic installed, publish() validates the value against it and stores plain data, and current() validates what it reads and returns it as that type. Any type Pydantic can validate works: a model, a dataclass, a TypedDict, or a plain annotation.

from pydantic import BaseModel


class TrainingConfig(BaseModel):
    threshold: float
    epochs: int


configs = cf.Catalog[TrainingConfig]("configs/{name}")
counts = cf.Catalog[dict[str, int]]("counts/{name}")

training = configs.at(name="training")
training.publish(TrainingConfig(threshold=0.7, epochs=20))
training.current().epochs   # 20

counts.at(name="words").publish({"the": 412})

A model is stored as its fields, so what's in the catalog doesn't depend on the class: Studio and the CLI show the fields, an untyped handle to the same path reads them as a dict, and a dict handle reads a version a model published. The type that validates a read is the reader's, the one declared in the code doing the reading, whatever the publisher used and wherever its class lives. So a reader declares the shape it expects, and a version that doesn't fit is an error at the read rather than somewhere downstream. A reader that only needs some of the fields can declare only those.

An asset, or another of Coflux's own types, is checked by instance: cf.Catalog[dict[str, cf.Asset]] needs nothing more. A model that holds one needs Pydantic told that an arbitrary type is fine there:

from pydantic import BaseModel, ConfigDict


class Churn(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)

    weights: cf.Asset
    auc: float

Without Pydantic, the type only informs type checkers, as with task arguments and results: the value is stored and returned as it is.

Re-running on a new version​

current() is a read: what is the value, as of my snapshot? next() is a request to re-run: run me again when this path has something I haven't seen. It suspends the execution — always, whether or not it's inside a cf.suspense scope — until the path has a version newer than the snapshot, and it returns nothing. The execution that resumes the step runs from the top with a fresh snapshot (the one case a resumed step doesn't keep the snapshot it had), so its current() returns the new value:

@cf.workflow()
def retrain_on_publish():
    customers = datasets.at(name="customers")
    train(customers.current())   # memoised, so a re-run with the same data is a hit
    customers.next()             # suspends until a newer version lands

Nothing is carried between attempts. The only thing the server holds for the suspended execution is what it's waiting for.

As with any suspension, the code before the wait runs again on resumption, so what it calls should be memoised. Here that is what you want anyway: when a new dataset lands, train sees a new argument, misses the memo, and trains on the new data.

next() needs no cf.suspense scope, and the body above has none, so waiting for train holds the worker slot and the step runs once per version. Put the reads in a scope if the path might be empty and you'd rather suspend than hold the slot until the first publish. Put the whole body in one and a slow train suspends too, after the scope's timeout, and the resumed attempt runs the body again: harmless, since train is memoised, but a second attempt per version.

Because the wait is relative to what the execution can see, an execution's own publish never wakes it. If a newer version already exists when next() is called, the execution suspends and resumes straight away.

A catalog handle is also a handle for cf.select. It resolves when the path has a version this execution hasn't seen — on an empty path, the first — so it is next() in select form, and the thing to do when a handle wins is call next() on it. What makes the combination useful is the other handles:

with cf.suspense():
    customers = datasets.at(name="customers")
    training = configs.at(name="training")
    retrain_now = cf.Prompt("Retrain now?").submit()
    ...
    winner, _ = cf.select([customers, training, retrain_now])   # whichever comes first
    if isinstance(winner, cf.Catalog):
        winner.next()   # re-run on the new data or config

Workspaces​

A version is published into the publishing execution's workspace. A read from a workspace sees every version published in it or any of its bases, and the current version is the newest of those, whichever workspace it came from — the same way cached results are inherited.

So a dev workspace derived from prod reads prod's versions, and keeps tracking prod's publishes after publishing its own: a dev publish is current in dev until prod publishes something newer. prod never sees dev's versions. Numbers are allocated across both, so prod may see @5 followed by @7, with @6 belonging to dev.

To keep a derived workspace off its base's publishes, use a different path prefix or a workspace with no base.

Passing values between tasks​

To make two steps agree on the same data, read it once and pass the value. A value is immutable, so the receiving task sees exactly what the reader saw, and a memoised task hits on identical content whichever version it came from. The read is recorded against the execution that made it, and Studio shows which version that was.

Studio and the CLI​

Studio's Catalog page lists every path visible from the workspace with its latest version and its value, and the versions behind each path. A run shows what each execution published and read.

It can also publish: Publish takes a path and either a JSON value or files to upload. Uploaded files become an asset — one file, several, or a folder, which becomes one entry per file keyed by its path within the folder. Archives are not expanded, so a zip you upload is a zip in the asset. The files go to the configured blob store and the server is given their keys, so this works the same whichever store is configured — though a worker only sees the blobs if its own coflux.toml lists that store too.

coflux catalog list [prefix]                        # paths and their latest versions
coflux catalog inspect <path>                       # the versions at a path
coflux catalog get <path>[@<number>]                # print a version's value
coflux catalog publish <path> <json>                # publish a JSON value
coflux catalog publish <path> --asset <asset-id>    # publish an existing asset
coflux catalog download <path>[@<number>] --to ./dir
coflux submit --catalog <path>@<number> ...         # run as of a version
coflux runs rerun --catalog <path>@<number>|latest <step-id>

download restores the assets a value holds. A value holding a single asset — the asset itself, or a structure with one asset somewhere in it — restores flat into the directory. One holding several assets restores each into a subdirectory named by the keys (or indices) leading to it, so {"train": a, "test": b} restores into train/ and test/. One holding no assets is an error; get shows it instead.