hi <@ULVA73B9P> from prefect import flow, task f...
# ask-community
k
hi @Marvin from prefect import flow, task from prefect.task_runners import ConcurrentTaskRunner import time @task(log_prints=True) def task1(a): print(f"Running task{a}") time.sleep(300) return f"task{a} done" @flow(task_runner=ConcurrentTaskRunner()) def my_flow(): # Create renamed variants of the same task task1_a = task1.with_options(name="SAPS_MoveTrades_SERVER") task1_b = task1.with_options(name="SAGEN_PriceCopy_ZAR_CFD_SABOR_SERVER") f1 = task1_a.submit(1) f2 = task1_b.submit(2) r1 = f1.result() r2 = f2.result() print("Both finished:", r1, r2) if name == "__main__": my_flow() thsu si s ione script, from prefect import flow, task #from prefect.task_runners import ConcurrentTaskRunner import time @task(log_prints=True) def task1(a): print(f"Running task{a}") time.sleep(300) return f"task{a} done" # @task(log_prints=True) # def task2(): # print("Running task2") # time.sleep(300) # return "task2 done" @flow(log_prints=True) def my_flow(): f1 = task1.with_options(task_run_name="SAPS_MoveTrades_SERVER").submit(1) f2 = task1.with_options(task_run_name="SAGEN_PriceCopy_ZAR_CFD_SABOR_SERVER").submit(2) r1 = f1.result() r2 = f2.result() print("Both finished:", r1, r2) if name == "__main__": my_flow() thsi si another script,i can see the task logs in teh UI, but why can't i see teh task seusn themselves in teh UI., eventhough iam using submit(which willl run asynchrnous tasks)
m
thought for 177.0 seconds
Thanks for sharing the snippets. A couple of key points that should help explain what you’re seeing. - In Prefect 3.x, task runs show up inside the flow run’s detail page (Timeline and the Task Runs table), not as a separate global “Task Runs” page. Open the specific flow run in the UI to see individual task runs. - Using
.submit()
does create distinct task runs (and you’ll see their logs tagged with the task run name), but you’ll only see them as entries on the flow run’s page. A few things to check if you still don’t see the task runs themselves: - Make sure you’re looking at the flow run’s detail page and that no filters are active (status filters can hide them). - Confirm you’re connected to a running API (Cloud/Server). If you’re running against the ephemeral API (no
PREFECT_API_URL
), runs are short-lived and sometimes you’ll only catch logs while the process is alive. For a persistent view, set
PREFECT_API_URL
and run via a worker or start a local server. - Double-check there are no places where the task is called directly like
task1(1)
(only
.submit()
or
await task1(1)
will create a task run). - Upgrade to the latest Prefect 3.x if you’re on an older version. On naming: -
with_options(name=...)
sets the task’s definition name (how it’s grouped in the UI). -
with_options(task_run_name=...)
sets the per-run display name (what you see in logs and the run row). Both are fine; if you want each run to have a specific label, use
task_run_name
. Example that will create two visible task runs under a single flow run:
Copy code
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner
import time

@task(log_prints=True)
def task1(a):
    print(f"Running task{a}")
    time.sleep(300)
    return f"task{a} done"

@flow(task_runner=ConcurrentTaskRunner(), log_prints=True)
def my_flow():
    f1 = task1.with_options(task_run_name="SAPS_MoveTrades_SERVER").submit(1)
    f2 = task1.with_options(task_run_name="SAGEN_PriceCopy_ZAR_CFD_SABOR_SERVER").submit(2)
    r1 = f1.result()
    r2 = f2.result()
    print("Both finished:", r1, r2)

if __name__ == "__main__":
    my_flow()
Helpful refs: - Task runners in Prefect 3.x: Task runners concept If this doesn’t line up with what you’re seeing, can you share: - Your Prefect version (
import prefect; print(prefect.__version__)
) - Whether you’re running locally via
python script.py
or via a deployment/worker - Whether
PREFECT_API_URL
is set (Cloud/Server) and a screenshot of the flow run page you’re checking I’ll help you zero in from there.