Define task name dynamically

Archive
1

Hi there,

I’m trying to define a dynamic name for my tasks, but i can’t figure out how. I want to have the exact table name in the task name.

I tried this, but with no luck:

@prefect.task(name="load_table_{table}")
def trigger_data_load(table: str):
    url = f"https://importerurl?table={table}"

    headers = {"Authorization": f"bearer {get_creds_token(url)}"}
    logger.info(f"Starting to extract data from {table}")
    ret = requests.get(
        url,
        headers=headers,
    )

    if ret.status_code != 200 and ret.status_code != 201:
        logger.warn(f"API response {ret.text}")
        raise Exception(ret.text)

    return ret


with Flow(
    name="import_data",
    schedule=CronSchedule("0 1 * * *", start_date=pendulum.today(tz="Europe/Berlin")),
) as f:

    for table in tables:
        trigger_data_load(table)

I saw a documentation entry for defining run_names though…

1 Like
2

we had an open GitHub issue for this for flow run names Allow setting a custom flow run name · Issue #5968 · PrefectHQ/prefect · GitHub

do you want to open a feature request on GitHub and explain there why adding such functionality is important for your use case?