## 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.

 Parameters will be validated against their type annotations, if
 they are type annotated.

 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. A "space" argument is published on the
 global run context for the call, same as the run, so the run
 and everything created during it, including its report, use
 that space. With "global_run="memorize"" that space stays;
 otherwise the previous space is restored afterward. 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: Sync code with
 git.

 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()