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?