Tom Matthews
05/23/2024, 10:21 AMMarvin
05/23/2024, 10:22 AMcreate_link_artifact
function:
python
from prefect import flow, task
from prefect.artifacts import create_link_artifact
@task
def my_first_task():
create_link_artifact(
key="irregular-data",
link="<https://nyc3.digitaloceanspaces.com/my-bucket-name/highly_variable_data.csv>",
description="## Highly variable data",
)
@task
def my_second_task():
create_link_artifact(
key="irregular-data",
link="<https://nyc3.digitaloceanspaces.com/my-bucket-name/low_pred_data.csv>",
description="# Low prediction accuracy",
)
@flow
def my_flow():
my_first_task()
my_second_task()
if __name__ == "__main__":
my_flow()
After running the above flow, you can find your artifacts in the Artifacts page of the Prefect UI.
More Information on Artifacts
2. Retrieving Artifacts via DBT Cloud:
You can also retrieve artifacts using the get_dbt_cloud_run_artifact
function:
python
from prefect import flow
from prefect_dbt.cloud import DbtCloudCredentials
from <http://prefect_dbt.cloud.jobs|prefect_dbt.cloud.jobs> import get_dbt_cloud_run_artifact
@flow
def get_artifact_flow():
credentials = DbtCloudCredentials(api_key="my_api_key", account_id=123456789)
get_run_artifact_result = get_dbt_cloud_run_artifact(
dbt_cloud_credentials=credentials,
run_id=42,
path="manifest.json"
)
with open("manifest.json", "w") as file:
json.dump(get_run_artifact_result, file)
if __name__ == "__main__":
get_artifact_flow()
This retrieves the specified artifact (in this example, manifest.json
) from DBT Cloud.
More Information on DBT Cloud Artifacts
For further details on querying artifacts directly, check out these resources:
- Artifacts Overview
- Using the Client
Remember, if you face issues specific to your setup, searching through relevant GitHub issues can also help, such as this one.
Let me know if you need further assistance.