https://prefect.io logo
Title
a

Axel Sundberg

11/08/2022, 3:55 PM
Hi! I'm on Prefect 1.2 and use function tasks with
@task
annotation. I have defined two tasks in a flow like this:
@task(name="Backup product catalogue", log_stdout=True)
def backup_task():
    backup_product_catalogue_to_gcs()


@task(name="Update Product Catalogue", log_stdout=True, trigger=triggers.all_finished)
def upsert_pack_products_task():
    upsert_pack_products()


with Flow(
    name="Product catalogue",
    run_config=get_run_config(),
    storage=get_storage("product_catalogue/flow.py"),
) as product_catalogue_flow:
    backup_task()
    upsert_pack_products_task(upstream_tasks=[backup_task])
I renamed the function name
upsert_pack_products_task
and re-deployed. Now I can see two
Backup product catalogue
in my flow on Prefect and they both run (see image) when the flow is triggered. How can I remove that extra task? And how can I debug what is going on?
2
It's because I reference
backup_task
twice in the flow definition 🤦
🙌 2