lamindb.flow ¶

lamindb.flow(uid=None, global_run='clear', track_arg_aliases=True)¶

Use @flow() to track a function as a workflow.

You will be able to see inputs, outputs, and parameters of the function in the data lineage graph.

The decorator creates a Transform with kind "script" that maps onto the file in which the function is defined. The function maps onto an entrypoint of the transform. A function execution creates a Run object that stores the function name in run.entrypoint. If the function is defined in a notebook cell or another ephemeral context, the transform is created with kind "function".

By default @ln.flow(), like ln.track(), creates a global run context that can be accessed with ln.context.run.

Parameters:

• uid (str | None, default: None) – Persist the uid to identify a transform across renames.

• global_run (Literal['memorize', 'clear', 'none'], default: 'clear') – If "clear", set the global run context ln.context.run and clear after the function completes. If "memorize", set the global run context and do not clear after the function completes. Set this to "none" if you want to track concurrent executions of a flow() in the same Python process.

• track_arg_aliases (bool, default: True) – If True (default), maps function arguments with names project, space, branch, plan, and initiated_by_run to matching ln.track() arguments while also keeping them in run.params for reproducibility. Pass False to disable this mapping.

Return type:

Callable[[Callable[[ParamSpec(P, bound= None)], TypeVar(R)]], Callable[[ParamSpec(P, bound= None)], TypeVar(R)]]

Examples

To sync a workflow with a file in a git repo, see: Organize local development.

For an extensive guide, see: Manage workflows. Here follow some examples.

my_workflow.py¶

import lamindb as ln


@ln.flow()
def ingest_dataset(key: str) -> ln.Artifact:
    df = ln.examples.datasets.mini_immuno.get_dataset1()
    artifact = ln.Artifact.from_dataframe(df, key=key).save()
    return artifact


if __name__ == "__main__":
    ingest_dataset(key="my_analysis/dataset.parquet")

my_workflow_with_step.py¶

import lamindb as ln


@ln.step()
def subset_dataframe(
    artifact: ln.Artifact,
    subset_rows: int = 2,
    subset_cols: int = 2,
) -> ln.Artifact:
    df = artifact.load()
    new_data = df.iloc[:subset_rows, :subset_cols]
    new_key = artifact.key.replace(".parquet", "_subsetted.parquet")
    return ln.Artifact.from_dataframe(new_data, key=new_key).save()


@ln.flow()
def ingest_dataset(key: str, subset: bool = False) -> ln.Artifact:
    df = ln.examples.datasets.mini_immuno.get_dataset1()
    artifact = ln.Artifact.from_dataframe(df, key=key).save()
    if subset:
        artifact = subset_dataframe(artifact)
    return artifact


if __name__ == "__main__":
    ingest_dataset(key="my_analysis/dataset.parquet", subset=True)

my_workflow_with_click.py¶

import click
import lamindb as ln


@click.command()
@click.option("--key", required=True)
@ln.flow()
def main(key: str):
    df = ln.examples.datasets.mini_immuno.get_dataset2()
    ln.Artifact.from_dataframe(df, key=key).save()


if __name__ == "__main__":
    main()