flyteplugins.union.factory

Factories: a declared graph of partitioned artifacts the platform can materialize on demand.

A factory is written in its own module, optionally in its own project, and wires together tasks owned by other teams without those tasks changing. Deploying it produces an ordinary task of type factory; running that task for a target and a partition selector is a materialization. Backfill is a materialization over a range.

Typical use::

from flyte.remote import Task
from flyteplugins.union import factory

clean = Task.get("ingest.clean", auto_version="latest")
report = Task.get("analytics.report", auto_version="latest")

raw = factory.source("raw_events")  # type and partitions come from the registry
events = factory.build("events").using(clean, raw=raw, min_quality=30)
daily_report = factory.build("daily_report").using(report, events=events.all("region"))

analytics = factory.Factory("analytics", daily_report)
analytics.deploy()
run = analytics.materialize(daily_report, date="2026-08-01..2026-08-31")

source and build declare; nothing runs until materialize. A build’s names line up with its task’s return tuple, and "_" skips an output that is not an artifact::

model, metrics = factory.build("_", "model", "metrics").using(train, data=features)

Partitions are identity in the artifact registry: a version is published with its typed partition values (one time partition with a granularity, any number of string partitions) and looked up by them. A source needs no type= or partitions= when the registry already knows the artifact; deploy reads them from it::

raw = factory.source("raw_events")

Directory

Classes

Methods

Variables

Methods

build()

def build(
    *outputs: str,
    partitions: Optional[Mapping[str, DimensionType]] = None,
    runcontext: Optional[Mapping[str, Any]] = None,
    backfill: str = 'partition',
    project: Optional[str] = None,
    domain: Optional[str] = None,
    kind: Union[None, str, Mapping[str, str]] = None,
    description: str = '',
) -> BuildSpec

Name the artifact (or artifacts) one task call makes. Follow with .using(task, **args).

outputs lines up with the task’s return tuple, left to right. Use "_" for an output that is not an artifact::

events = factory.build("events").using(clean, raw=raw_events)
model, metrics = factory.build("_", "model", "metrics", "_").using(train, data=features)

materialize()

def materialize(
    factory_name: str,
    target: Union[ArtifactHandle, str],
    partitions: Mapping[str, Any],
    params: Optional[Mapping[str, Mapping[str, Any]]] = None,
    rebuild: Optional[Sequence[Any]] = None,
    rebuild_all: bool = False,
    downstream: bool = False,
    plan_only: bool = False,
    concurrency: int = 0,
    queue: Optional[str] = None,
    versions: Optional[Mapping[str, str]] = None,
    nodes: Optional[Mapping[str, ArtifactHandle]] = None,
    project: Optional[str] = None,
    domain: Optional[str] = None,
    run_name: Optional[str] = None,
)

Start a run of <factory>.materialize and return the flyte.remote.Run.

queue puts the materialization run and every build it makes on that queue. versions pins sources to artifact versions ({"tracks": "v2"}) instead of the latest version.

partition()

def partition(
    dim: str,
) -> PartitionValue

Pass the instance’s value of dim to a task parameter.

factory.build("events").using(clean, raw=raw_events, day=factory.partition("date")) calls clean with day set to the date being built. A parameter that is left unset and has the same name as a dimension receives that dimension’s value without this.

source()

def source(
    name: str,
    project: Optional[str] = None,
    domain: Optional[str] = None,
    type: Any = None,
    partitions: Optional[Mapping[str, DimensionType]] = None,
    kind: Optional[str] = None,
    description: str = '',
) -> ArtifactHandle

An artifact made outside this factory. The factory resolves it from the registry.

type and partitions may be left out: at deploy they are read from the registry, which fixed the partition keys with the artifact’s first version (or a declaration). When partitions is given and the registry already has a schema for the name, the two must agree; when it has none, deploy declares it.