Metadata-Version: 2.3
Name: openlineage-prefect
Version: 0.0.1b5
Summary: OpenLineage integration with Prefect
Keywords: openlineage,prefect
Author: OpenLineage
License: Apache-2.0
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Dist: prefect>=3.8.5
Requires-Dist: openlineage-python>=1.53.0
Requires-Dist: mypy ; extra == 'dev'
Requires-Dist: openlineage-python ; extra == 'dev'
Requires-Dist: pillow ; extra == 'dev'
Requires-Dist: pre-commit ; extra == 'dev'
Requires-Dist: pytest>=8.3 ; extra == 'dev'
Requires-Dist: pytest-asyncio ; extra == 'dev'
Requires-Dist: pytest-cov ; extra == 'dev'
Requires-Dist: pytest-env ; extra == 'dev'
Requires-Dist: pytest-timeout>=2.3.1 ; extra == 'dev'
Requires-Dist: pytest-xdist ; extra == 'dev'
Requires-Python: >=3.10
Project-URL: Homepage, https://openlineage.io/
Project-URL: Source, https://github.com/OpenLineage/OpenLineage/tree/main/integration/prefect
Provides-Extra: dev
Description-Content-Type: text/markdown

# Openlineage Prefect

## Basic Configuration

At a minimum, define a namespace, Prefect API URL and transport consisting of a type, URL and endpoint. All three can be supplied using environment variables.

For example:

```sh
export OPENLINEAGE_NAMESPACE='prefect_test' &&
export OPENLINEAGE__TRANSPORT__TYPE='http' &&
export OPENLINEAGE__TRANSPORT__URL='http://lineageconsumer.com:5000' &&
export OPENLINEAGE__TRANSPORT__ENDPOINT='/api/v1/lineage' &&
export PREFECT_API_URL='http://prefecthost.com:4200/api'
```

For more details about OpenLineage transport options and how to configure them, consult the [OpenLineage Python Client Documentation](https://openlineage.io/docs/client/python/).

## Execution

Import the `openlineage_prefect` package and execute `collect_and_process_runs()` asynchronously in its own process.

For example:

```py
import asyncio
import openlineage_prefect
from openlineage_prefect.prefect_adapter.listener import PrefectOpenLineageListener

async def main():
    await PrefectOpenLineageListener().collect_and_process_runs()

if __name__ == "__main__":
    asyncio.run(main())
```

## Datasets

The integration looks for datasets in Prefect Artifacts. To attach input and output datasets to job runs, use `create_table_artifact()` from the Artifact library. Provide a namespace, typically the dataset URI, and name to the adapter via an artifact's `table` and `description`, respectively. Distinguish the type of dataset by appending `_output` or `_input` to the description.

For example:

```py
ol_table = [{"database_uri":"duckdb:///customers_db", "table":"customers"}]

create_table_artifact(
    key="upstream-insert",
    table=ol_table,
    description="ol-dataset_output"
)
```
