Marvin mentioned my_task.get_state() and my_task.get_task_run_id() but I don’t see it in the code in Prefect 3.2.14

actually I don’t see even in 3.5.0 here

@Marvin in Prefect I can do task1 = my_task1.submit(wait_for=[another_task]). Afterwards, I have task2 = my_task2.submit(wait_for=[allow_failure(task1)]). If another_task fails, my_task1 is not executed? What about my_task2?

Short answer (assuming Prefect 3.x): - If

another_task

fails,

my_task1

will not run. It won’t be “skipped”; it will never start and is effectively marked as not runnable due to its failed upstream (it typically ends the run as Cancelled/NotReady). -

my_task2

will still run. The

allow_failure(task1)

in

wait_for

tells Prefect that a non-success outcome for

task1

(Failed/Cancelled/Crashed) is acceptable.

my_task2

will wait until

task1

reaches a terminal state (in this case Cancelled because it never ran), then it proceeds. Key points: - Without

allow_failure

, downstream tasks block on non-completed upstreams and won’t run if an upstream fails. - With

allow_failure

, the dependency can fail/cancel and the downstream still proceeds; it only waits for the upstream to reach a terminal state. Example pattern:

t1 = my_task1.submit(wait_for=[another_task])             # won't run if another_task fails
t2 = my_task2.submit(wait_for=[allow_failure(t1)])        # will still run once t1 is terminal (e.g., Cancelled)

Docs for more detail: - States in Prefect 3.x: Concepts: States - Concurrency and `wait_for`: How-to: Run work concurrently

You’re running into Prefect’s log batching. In Prefect 3.x, task/flow logs are queued in-process and sent to the API in batches (by default every ~2 seconds or when the batch gets large).

future.wait()

only waits for the task state to finish, not for the logging queue to flush to the API — so a read right after can miss the latest lines. Ways to make sure you get all the logs: - If you’re calling from inside the same running process (e.g., within the flow run), explicitly flush the log queue before you read:

from prefect.logging.handlers import APILogHandler
  await future.wait()
  await APILogHandler.aflush()  # ensure all queued logs have been sent
  logs = await c.read_logs(
      log_filter=LogFilter(task_run_id={"any_": [future.task_run_id]}),
  )

- If you’re calling from a different process than the one emitting logs (e.g., a separate client script querying a remote worker), you can’t flush the remote queue — add a short poll/sleep after

future.wait()

to allow the worker to push the final batch:

import asyncio
  from prefect.client.schemas.filters import LogFilter
  from prefect.client.schemas.sorting import LogSort

  await future.wait()
  for _ in range(6):  # ~3s total
      logs = await c.read_logs(
          log_filter=LogFilter(task_run_id={"any_": [future.task_run_id]}),
          sort=LogSort.TIMESTAMP_ASC,
      )
      # break when counts stabilize
      await asyncio.sleep(0.5)

- Watch out for pagination. The API returns up to 200 logs per call. If your task emits more, you must page:

from prefect.client.schemas.filters import LogFilter
  from prefect.client.schemas.sorting import LogSort

  lf = LogFilter(task_run_id={"any_": [future.task_run_id]}, level={"ge_": 20})
  all_logs, offset = [], 0
  while True:
      batch = await c.read_logs(log_filter=lf, limit=200, offset=offset, sort=LogSort.TIMESTAMP_ASC)
      if not batch:
          break
      all_logs.extend(batch)
      if len(batch) < 200:
          break
      offset += len(batch)

- If you want logs to appear faster overall, reduce the batch interval (trade-off: more API calls):

prefect config set PREFECT_LOGGING_TO_API_BATCH_INTERVAL=1.0

Then restart your workers/flow run environment. Notes: - Your

LogFilter

usage (

task_run_id={"any_": [...]}, level={"ge_": 20}

) is correct. - If you want to fetch “latest first,” set

sort=LogSort.TIMESTAMP_DESC

. With the 200-log limit, ascending sort can make it look like “latest logs are missing” if there are more than 200 lines. Helpful docs: - Logs overview: Prefect logs concepts - Filters API reference: LogFilter reference - Client read_logs (API reference): Client orchestration API (see LogAsyncClient.read_logs) If you can share whether you’re querying from inside the same flow run (vs. an external client) and roughly how many logs you expect, I can tailor the code to your setup.

Short answer:

APILogHandler.aflush()

flushes the in-memory log queue for the current process. That means: - It will send all pending Prefect log records that were emitted by this process, regardless of which flow/task run they belong to. - It does not flush logs from other processes or machines. So if you call it from inside a running flow’s process, it will typically flush that flow run’s logs (and any subflows/tasks in the same process). If you have multiple runs emitting logs in the same process, it will flush all of their pending logs together. Your subsequent

read_logs(LogFilter(...))

will still only return the logs for the run you filter on. Example:

from prefect.logging.handlers import APILogHandler
from prefect.client.schemas.filters import LogFilter

await future.wait()
await APILogHandler.aflush()  # flushes all pending logs from this process
logs = await c.read_logs(
    log_filter=LogFilter(task_run_id={"any_": [future.task_run_id]}),
)

If your logs are produced by a remote worker/another process, calling

aflush()

locally won’t help; add a brief delay/poll as the worker flushes its own batch.