Tom Matthews
05/23/2024, 10:21 AMMarvin
05/23/2024, 10:22 AMcreate_link_artifactpython
    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()get_dbt_cloud_run_artifactpython
    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()manifest.json