How to get task name from task_run_id?

[tonal]

@task
async def tsk_copy_all(conf):
  defers = [
    tsk_copy_data.submit(conf, cls=FirmParam),
    tsk_add_props_help.submit(conf),
    tsk_copy_goods.submit(conf),
    tsk_copy_cats.submit(conf)]
    
  res = wait(defers)
  if errs := tuple(r for r in res.done if r.state.is_failed or r.state.is_crashed):
    err_tsks_names = ???
    print('tasks error:', err_tsks_names)
    return Failed(message=f'Not all tasks Ok: {err_tsks_names}'

[zzstoatzz]

hi @tonal - to read the run names I think you'd have to go back to the API since the run name doesn't live on state_details (which is available here is fut.state.state_details, which may not be ideal because it might take a second for the run info to propagate per #14661

for example

import asyncio
import time
from uuid import UUID

from prefect import flow, get_client, task
from prefect.client.schemas.filters import TaskRunFilter, TaskRunFilterId
from prefect.futures import wait


@task
def errors_on_odd(number: int) -> int:
    if number % 2 == 0:
        return number
    else:
        raise ValueError("This is a test error")


async def _task_run_names_from_ids(ids: tuple[UUID, ...]) -> list[str]:
    results = []

    while len(results) != len(list(ids)):
        async with get_client() as client:
            results = [
                run.name
                for run in await client.read_task_runs(
                    task_run_filter=TaskRunFilter(id=TaskRunFilterId(any_=list(ids)))
                )
            ]
        await asyncio.sleep(0.1)

    return results


@flow(log_prints=True)
async def some_flow():
    futures = errors_on_odd.map(range(10))
    if errored_task_run_ids := tuple(
        f.task_run_id
        for f in wait(futures).done
        if f.state.is_failed() or f.state.is_crashed()
    ):
        start = time.monotonic()
        err_names = await _task_run_names_from_ids(errored_task_run_ids)
        print(f"finished with errors: {err_names}")
        print(f"took {time.monotonic() - start} seconds to get task run names")
    else:
        print("finished without errors")


if __name__ == "__main__":
    asyncio.run(some_flow())

...
14:58:06.928 | INFO    | Task run 'errors_on_odd-d9e' - Finished in state Completed()
...
14:58:07.195 | ERROR   | Task run 'errors_on_odd-ed9' - Finished in state Failed('Task run encountered an exception ValueError: This is a test error')
14:58:09.185 | INFO    | Flow run 'peridot-mantis' - finished with errors: ['errors_on_odd-f97', 'errors_on_odd-37b', 'errors_on_odd-ed9', 'errors_on_odd-8e5', 'errors_on_odd-7ce']
14:58:09.190 | INFO    | Flow run 'peridot-mantis' - took 1.993007583077997 seconds to get task run names
14:58:09.324 | INFO    | Flow run 'peridot-mantis' - Finished in state Completed()

maybe we ought to find a way to make this easier (perhaps by putting the run name on state_details), since "the names of the runs that failed" seems like information that should be more easily accessible (though you could track this information yourself)

also just to call out that here in your example you have

  if errs := tuple(r for r in res.done if r.state.is_failed or r.state.is_crashed):

this would always be true because you're checking the truthy-ness of the methods themselves instead of calling them

before

if r.state.is_failed or r.state.is_crashed

after

if r.state.is_failed() or r.state.is_crashed()