# Artifacts
> This bundle contains all pages in the Artifacts section.
> Source: https://www.union.ai/docs/v1/union/user-guide/core-concepts/artifacts/

=== PAGE: https://www.union.ai/docs/v1/union/user-guide/core-concepts/artifacts ===

# Artifacts

> **📝 Note**
>
> An LLM-optimized bundle of this entire section is available at [`section.md`](https://www.union.ai/docs/v1/union/user-guide/core-concepts/section.md).
> This single file contains all pages in this section, optimized for AI coding agent context.

Union.ai produces many intermediate outputs when running tasks and workflows. These outputs are stored internally in Union.ai and are accessible through the relevant executions, but are not usually directly accessible to users.

The Artifact service indexes and adds semantic meaning to outputs of all Union.ai task and workflow executions, such as models, files, or any other kinds of data, enabling you to directly access, track, and orchestrate pipelines through the outputs themselves. Artifacts allow you to store additional metadata for these outputs in the form of **Core concepts > Artifacts > Partitions**, which are key-value pairs that describe the artifact and which can be used to query the Artifact Service to locate artifacts. Artifacts allow for loose coupling of workflows—for example, a downstream workflow can be configured to consume the latest result of an upstream workflow. With this higher-order abstraction, Union.ai aims to ease collaboration across teams, provide for reactivity and automation, and give you a broader view of how artifacts move across executions.

## Versioning

Artifacts are uniquely identified and versioned by the following information:

* Project
* Domain
* Artifact name
* Artifact version

You can set an artifact's name in code when you **Core concepts > Artifacts > Declaring artifacts** and the artifact version is automatically generated when the artifact is materialized as part of any task or workflow execution that emits an artifact with this name. Any execution of a task or workflow that emits an artifact creates a new version of that artifact.

## Partitions

When you declare an artifact, you can define partitions for it that enable semantic grouping of artifacts. Partitions are metadata that take the form of key-value pairs, with the keys defined at registration time and the values supplied at runtime. You can specify up to 10 partition keys for an artifact. You can set an optional partition called `time_partition` to capture information about the execution timestamp to your desired level of granularity. For more information, see **Core concepts > Artifacts > Declaring artifacts**.

> [!NOTE]
> The `time_partition` partition is not enabled by default. To enable it, set `time_partitioned=True` in the artifact declaration.
> For more information, see the **Core concepts > Artifacts > Declaring artifacts**.

## Queries

To consume an artifact in a workflow, you can define a query containing the artifact’s name as well as any required partition values. You then supply the query as an input value to the workflow definition. At execution time, the query will return the most recent version of the artifact that meets the criteria by default. You can also query for a specific artifact version.

For more information on querying for and consuming artifacts in workflows, see **Core concepts > Artifacts > Consuming artifacts in workflows**.

To query for artifacts programmatically in a Python script using `UnionRemote`, see [UnionRemote](https://www.union.ai/docs/v1/union/user-guide/development-cycle/union-remote).

> [!NOTE] `UnionRemote` vs `FlyteRemote`
> `UnionRemote` is identical to `FlyteRemote`, with additional functionality to handle artifacts.
> You cannot interact with artifacts using `FlyteRemote`.

## Lineage

Once an artifact is materialized, its lineage is visible in the UI. For more information, see **Core concepts > Artifacts > Viewing artifacts**.

=== PAGE: https://www.union.ai/docs/v1/union/user-guide/core-concepts/artifacts/declaring-artifacts ===

# Declaring artifacts

In order to define a task or workflow that emits an artifact, you must first declare the artifact and the keys for any **Core concepts > Artifacts > Declaring artifacts** you wish for it to have. For the `Artifact` class parameters and methods, see the [Artifact API documentation]().
<!-- TODO: Add link to API -->

## Basic artifact

In the following example, an artifact called `BasicTaskData` is declared, along with a task that emits that artifact. Since it is a basic artifact, it doesn't have any partitions.

> [!NOTE]
> To use the example code on this page, you will need to add your `registry`
> to the `pandas_image` ImageSpec block.

```python
# basic.py

import pandas as pd
import union
from typing_extensions import Annotated

pandas_image = union.ImageSpec(
    packages=["pandas==2.2.2"]
)

BasicTaskData = union.Artifact(
    name="my_basic_artifact"
)

@union.task(container_image=pandas_image)
def t1() -> Annotated[pd.DataFrame, BasicTaskData]:
    my_df = pd.DataFrame({"col1": [1, 2, 3], "col2": ["a", "b", "c"]})
    return BasicTaskData.create_from(my_df)

@union.workflow
def wf() -> pd.DataFrame:
    return t1()
```

## Time-partitioned artifact

By default, time partitioning is not enabled for artifacts. To enable it, declare the artifact with `time_partitioned` set to `True`. You can optionally set the granularity for the time partition to `MINUTE`, `HOUR`, `DAY`, or `MONTH`; the default is `DAY`.

You must also pass a value to `time_partition`, which you can do at runtime or by binding `time_partition` to an input.

### Passing a value to `time_partition` at runtime

```python
# time_partition_runtime.py

from datetime import datetime

import pandas as pd
import union
from flytekit.core.artifact import Granularity
from typing_extensions import Annotated

pandas_image = union.ImageSpec(
    packages=["pandas==2.2.2"]
)

BasicArtifact = union.Artifact(
    name="my_basic_artifact",
    time_partitioned=True,
    time_partition_granularity=Granularity.HOUR
)

@union.task(container_image=pandas_image)
def t1() -> Annotated[pd.DataFrame, BasicArtifact]:
    df = pd.DataFrame({"col1": [1, 2, 3], "col2": ["a", "b", "c"]})
    dt = datetime.now()
    return BasicArtifact.create_from(df, time_partition=dt)

@union.workflow
def wf() -> pd.DataFrame:
    return t1()
```
<!-- TODO :emphasize-lines: 1,5,14-15,21-23 -->

### Passing a value to `time_partition` by input

```python
# time_partition_input.py

from datetime import datetime

import pandas as pd
import union
from flytekit.core.artifact import Granularity
from typing_extensions import Annotated

pandas_image = union.ImageSpec(
    packages=["pandas==2.2.2"]
)

BasicArtifact = union.Artifact(
    name="my_basic_artifact",
    time_partitioned=True,
    time_partition_granularity=Granularity.HOUR
)

@union.task(container_image=pandas_image)
def t1(date: datetime) -> Annotated[pd.DataFrame, BasicArtifact]:
    df = pd.DataFrame({"col1": [1, 2, 3], "col2": ["a", "b", "c"]})
    return BasicArtifact.create_from(df, time_partition=date)

@union.workflow
def wf(run_date: datetime):
    return t1(date=run_date)
```
<!-- TODO: emphasize-lines: 20-21,28 -->

## Artifact with custom partition keys

You can specify up to 10 custom partition keys when declaring an artifact. Custom partition keys can be set at runtime or be passed as inputs.

### Passing a value to a custom partition key at runtime

```python
# partition_keys_runtime.py

from datetime import datetime

import pandas as pd
import union
from flytekit.core.artifact import Inputs, Granularity
from typing_extensions import Annotated

pandas_image = union.ImageSpec(
    packages=["pandas==2.2.2"]
)

BasicArtifact = union.Artifact(
    name="my_basic_artifact",
    time_partitioned=True,
    time_partition_granularity=Granularity.HOUR,
    partition_keys=["key1"]
)

@union.task(container_image=pandas_image)
def t1(
    key1: str, date: datetime
) -> Annotated[pd.DataFrame, BasicArtifact(key1=Inputs.key1)]:
    df = pd.DataFrame({"col1": [1, 2, 3], "col2": ["a", "b", "c"]})
    return BasicArtifact.create_from(
        df,
        time_partition=date
    )

@union.workflow
def wf():
    run_date = datetime.now()
    values = ["value1", "value2", "value3"]
    for value in values:
        t1(key1=value, date=run_date)
```
<!-- TODO: emphasize-lines: 16,35-36 -->

### Passing a value to a custom partition key by input

```python
# partition_keys_input.py

from datetime import datetime

import pandas as pd
import union
from flytekit.core.artifact import Inputs, Granularity
from typing_extensions import Annotated

pandas_image = union.ImageSpec(
    packages=["pandas==2.2.2"]
)

BasicArtifact = union.Artifact(
    name="my_basic_artifact",
    time_partitioned=True,
    time_partition_granularity=Granularity.HOUR,
    partition_keys=["key1"]
)

@union.task(container_image=pandas_image)
def t1(
    key1: str, dt: datetime
) -> Annotated[pd.DataFrame, BasicArtifact(key1=Inputs.key1)]:
    df = pd.DataFrame({"col1": [1, 2, 3], "col2": ["a", "b", "c"]})
    return BasicArtifact.create_from(
        df,
        time_partition=dt,
        key1=key1
    )

@union.workflow
def wf(dt: datetime, val: str):
    t1(key1=val, dt=dt)
```
<!-- TODO: emphasize-lines: 16,34 -->

## Artifact with model card example

You can attach a model card with additional metadata to your artifact, formatted in Markdown:

```python
# model_card.py

import pandas as pd
import union
from union.artifacts import ModelCard
from typing_extensions import Annotated

pandas_image = union.ImageSpec(
    packages=["pandas==2.2.2"]
)

BasicArtifact = union.Artifact(name="my_basic_artifact")

def generate_md_contents(df: pd.DataFrame) -> str:
    contents = "# Dataset Card\n" "\n" "## Tabular Data\n"
    contents = contents + df.to_markdown()
    return contents

@union.task(container_image=pandas_image)
def t1() -> Annotated[pd.DataFrame, BasicArtifact]:
    df = pd.DataFrame({"col1": [1, 2, 3], "col2": ["a", "b", "c"]})

    return BasicArtifact.create_from(
        df,
        ModelCard(generate_md_contents(df))
    )

@union.workflow
def wf():
    t1()
```
<!-- TODO: emphasize-lines: 4,14-17,26 -->

=== PAGE: https://www.union.ai/docs/v1/union/user-guide/core-concepts/artifacts/materializing-artifacts ===

# Materializing artifacts

You can materialize an artifact by executing the task or workflow that emits the artifact.

In the example below, to materialize the `BasicArtifact` artifact, the `t1` task must be executed. The `wf` workflow runs the `t1` task three times with different values for the `key1` partition each time.
Note that each time `t1` is executed, it emits a new version of the `BasicArtifact` artifact.

> [!NOTE]
> To use the example code on this page, you will need to add your `registry` to the `pandas_image` ImageSpec block.

```python
# partition_keys_runtime.py

from datetime import datetime

import pandas as pd
import union
from flytekit.core.artifact import Inputs, Granularity
from typing_extensions import Annotated

pandas_image = union.ImageSpec(
    packages=["pandas==2.2.2"]
)

BasicArtifact = union.Artifact(
    name="my_basic_artifact",
    time_partitioned=True,
    time_partition_granularity=Granularity.HOUR,
    partition_keys=["key1"]
)

@union.task(container_image=pandas_image)
def t1(
    key1: str, date: datetime
) -> Annotated[pd.DataFrame, BasicArtifact(key1=Inputs.key1)]:
    df = pd.DataFrame({"col1": [1, 2, 3], "col2": ["a", "b", "c"]})
    return BasicArtifact.create_from(
        df,
        time_partition=date
    )

@union.workflow
def wf():
    run_date = datetime.now()
    values = ["value1", "value2", "value3"]
    for value in values:
        t1(key1=value, date=run_date)
```

> [!NOTE]
> You can also materialize an artifact by executing the `create_artifact` method of `UnionRemote`.
> For more information, see the [UnionRemote documentation](https://www.union.ai/docs/v1/union/user-guide/core-concepts/development-cycle/union-remote).

=== PAGE: https://www.union.ai/docs/v1/union/user-guide/core-concepts/artifacts/consuming-artifacts-in-workflows ===

# Consuming artifacts in workflows

## Defining a workflow that consumes an artifact

You can define a workflow that consumes an artifact by defining a query and passing it as an input to the consuming workflow.

The following code defines a query, `data_query`, that searches across all versions of `BasicArtifact` that match the partition values. This query binds parameters to the workflow's `key1` and `time_partition` inputs and returns the most recent version of the artifact.

> [!NOTE]
> To use the example code on this page, you will need to add your `registry` to the `pandas_image` ImageSpec block.

```python
# query.py

from datetime import datetime

import pandas as pd
import union
from flytekit.core.artifact import Inputs

pandas_image = union.ImageSpec(
    packages=["pandas==2.2.2"]
)

BasicArtifact = union.Artifact(
    name="my_basic_artifact"
)

@union.task(container_image=pandas_image)
def t1(key1: str, dt: datetime, data: pd.DataFrame):
    print(f"key1: {key1}")
    print(f"Date: {dt}")
    print(f"Data retrieved from query: {data}")

data_query = BasicArtifact.query(
    time_partition=Inputs.dt,
    key1=Inputs.key1,
)

@union.workflow
def query_wf(
    key1: str,
    dt: datetime,
    data: pd.DataFrame = data_query
):
    t1(key1=key1, dt=dt, data=data)
```
<!-- TODO :emphasize-lines: 23-26,35 -->

You can also directly reference a particular artifact version in a query using the `get()` method:

```python
data = BasicArtifact.get(<organization>/<domain>/BasicArtifact@<artifact-version>)
```

> [!NOTE]
> For a full list of Artifact class methods, see the [Artifact API documentation]().
<!-- TODO: Add link to API -->

## Launching a workflow that consumes an artifact

To launch a workflow that consumes an artifact as one of its inputs, navigate to the workflow in the UI and click **Launch Workflow**:

![Launch workflow UI with artifact query](https://www.union.ai/docs/v1/union/_static/images/user-guide/core-concepts/artifacts/consuming-artifacts-in-workflows/launch-workflow-artifact-query.png)

In the `query_wf` example, the workflow takes three inputs: `key1`, `dt`, and a `BasicArtifact` artifact query. In order to create the workflow execution, you would enter values for `key1` and `dt` and click **Launch**. The artifacts service will supply the latest version of the `BasicData` artifact that meets the partition query criteria.

You can also override the artifact query from the launch form by clicking **Override**, directly supplying the input that the artifact references (in this case, a blob store URI), and clicking **Launch**:

![Launch workflow UI with artifact query override](https://www.union.ai/docs/v1/union/_static/images/user-guide/core-concepts/artifacts/consuming-artifacts-in-workflows/launch-workflow-artifact-query-override.png)

=== PAGE: https://www.union.ai/docs/v1/union/user-guide/core-concepts/artifacts/connecting-workflows-with-artifact-event-triggers ===

# Connecting workflows with artifact event triggers

In the following example, we define an upstream workflow and a downstream workflow, and define a [trigger](https://www.union.ai/docs/v1/union/user-guide/core-concepts/launch-plans/reactive-workflows) in a launch plan to connect the two workflows via an [artifact event](https://www.union.ai/docs/v1/union/user-guide/core-concepts/launch-plans/reactive-workflows).

## Imports

> [!NOTE]
> To use the example code on this page, you will need to add your `registry` to the `pandas_image` ImageSpec block.

First we import the required packages:

```python
from datetime import datetime

import pandas as pd
import union
from union.artifacts import OnArtifact
from flytekit.core.artifact import Inputs
from typing_extensions import Annotated

```

## Upstream artifact and workflow definition

Then we define an upstream artifact and a workflow that emits a new version of `UpstreamArtifact` when executed:

```python
UpstreamArtifact = union.Artifact(
    name="my_upstream_artifact",
    time_partitioned=True,
    partition_keys=["key1"],
)

@union.task(container_image=pandas_image)
def upstream_t1(key1: str) -> Annotated[pd.DataFrame, UpstreamArtifact(key1=Inputs.key1)]:
    dt = datetime.now()
    my_df = pd.DataFrame({"col1": [1, 2, 3], "col2": ["a", "b", "c"]})
    return UpstreamArtifact.create_from(my_df, key1=key1, time_partition=dt)

@union.workflow
def upstream_wf() -> pd.DataFrame:
    return upstream_t1(key1="value1")
```

## Artifact event definition

Next we define the artifact event that will link the upstream and downstream workflows together:

```python
on_upstream_artifact = OnArtifact(
    trigger_on=UpstreamArtifact,
)
```

## Downstream workflow definition

Then we define the downstream task and workflow that will be triggered when the upstream artifact is created:

```python
@union.task
def downstream_t1():
    print("Downstream task triggered")

@union.workflow
def downstream_wf():
    downstream_t1()
```

## Launch plan with trigger definition

Finally, we create a launch plan with a trigger set to an `OnArtifact` object to link the two workflows via the `Upstream` artifact. The trigger will initiate an execution of the downstream `downstream_wf` workflow upon the creation of a new version of the `Upstream` artifact.

```python
downstream_triggered = union.LaunchPlan.create(
    "downstream_with_trigger_lp",
    downstream_wf,
    trigger=on_upstream_artifact
)
```

> [!NOTE]
> The `OnArtifact` object must be attached to a launch plan in order for the launch plan to be triggered by the creation of a new version of the artifact.

## Full example code

Here is the full example code file:

```python
# trigger_on_artifact.py
from datetime import datetime

import pandas as pd
import union
from union.artifacts import OnArtifact
from flytekit.core.artifact import Inputs
from typing_extensions import Annotated

pandas_image = union.ImageSpec(
    packages=["pandas==2.2.2"]
)

UpstreamArtifact = union.Artifact(
    name="my_upstream_artifact",
    time_partitioned=True,
    partition_keys=["key1"],
)

@union.task(container_image=pandas_image)
def upstream_t1(key1: str) -> Annotated[pd.DataFrame, UpstreamArtifact(key1=Inputs.key1)]:
    dt = datetime.now()
    my_df = pd.DataFrame({"col1": [1, 2, 3], "col2": ["a", "b", "c"]})
    return UpstreamArtifact.create_from(my_df, key1=key1, time_partition=dt)

@union.workflow
def upstream_wf() -> pd.DataFrame:
    return upstream_t1(key1="value1")

on_upstream_artifact = OnArtifact(
    trigger_on=UpstreamArtifact,
)

@union.task
def downstream_t1():
    print("Downstream task triggered")

@union.workflow
def downstream_wf():
    downstream_t1()

downstream_triggered = union.LaunchPlan.create(
    "downstream_with_trigger_lp",
    downstream_wf,
    trigger=on_upstream_artifact
)
```

=== PAGE: https://www.union.ai/docs/v1/union/user-guide/core-concepts/artifacts/viewing-artifacts ===

# Viewing artifacts

## Artifacts list

Artifacts can be viewed in the UI by navigating to the artifacts app in the left sidebar:

![Artifacts overview](https://www.union.ai/docs/v1/union/_static/images/user-guide/core-concepts/artifacts/viewing-artifacts/artifacts-list.png)

## Artifact view

Selecting a specific artifact from the artifact list will take you to that artifact's **Overview** page:

![Single artifact overview](https://www.union.ai/docs/v1/union/_static/images/user-guide/core-concepts/artifacts/viewing-artifacts/artifact-view.png)

Here you can see relevant metadata about the artifact, including:
* Its version
* Its partitions
* The task or workflow that produced it
* Its creation time
* Its object store URI
* Code for accessing the artifact via [UnionRemote](https://www.union.ai/docs/v1/union/user-guide/core-concepts/development-cycle/union-remote)

You can also view the artifact's object structure, model card, and lineage graph.

### Artifact lineage graph

Once an artifact is materialized, you can view its lineage in the UI, including the specific upstream task or workflow execution that created it, and any downstream workflows that consumed it. You can traverse the lineage graph by clicking between artifacts and inspecting any relevant workflow executions in order to understand and reproduce any step in the AI development process.

![Artifact lineage overview](https://www.union.ai/docs/v1/union/_static/images/user-guide/core-concepts/artifacts/viewing-artifacts/artifact-lineage.png)

You can navigate through the lineage graph by clicking from artifact to artifact.