https://prefect.io
Join Slack
<@ULVA73B9P> from prefect.server.schemas.core import FlowRun def pipeline_status( flow: Flow, flow_...
# ask-marvin
m
@Marvin from prefect.server.schemas.core import FlowRun def pipeline_status( flow: Flow, flow_run: FlowRun, state: State): # Get basic info from the passed objects flow_run_name = flow_run.name flow_name = flow.name flow_state = state.name run_url = f'{PREFECT_UI_URL.value()}/flow-runs/flow-run/{flow_run.id}' flow_runstamp = flow_run.start_time flow_parameters = flow_run.parameters pipeline_status is a on_completion hook, i want to get flow_runstamp = flow_run.start_time inside a task too, i need the exact time.
m
✅ thought for 100.2 seconds
You can get the exact (actual) start time inside a task from the run context. In Prefect 3.x, 
prefect.runtime.flow_run.scheduled_start_time
is just the scheduled time; the actual start is available on the FlowRun object via 
get_run_context()
. Example:
Copy code
from prefect import flow, task
from prefect.context import get_run_context

@task
def my_task():
    ctx = get_run_context()          # This is a TaskRunContext inside a task
    flow_start = ctx.flow_run.start_time      # actual flow run start time (datetime)
    task_start = ctx.task_run.start_time      # actual task run start time (datetime)
    # do something with flow_start / task_start
    return flow_start

@flow
def my_flow():
    flow_ctx = get_run_context()     # FlowRunContext here
    actual_start = flow_ctx.flow_run.start_time
    ts = my_task()
    return actual_start, ts
Notes: - 
ctx.flow_run.start_time
is the actual time the flow began running. Likewise, 
ctx.task_run.start_time
is the actual task start time. - If you need the scheduled time, use 
prefect.runtime.flow_run.scheduled_start_time
. - Prefer importing schemas from 
prefect.schemas
rather than 
prefect.server.schemas.core
in user code:
Copy code
from prefect.schemas import FlowRun, State
- Very early in execution, 
start_time
can be None momentarily; if you hit that edge case, add a small guard/retry.
m
@Marvin raise AttributeError(f'{type(self).name!r} object has no attribute {item!r}') AttributeError: 'TaskRunContext' object has no attribute 'flow_run'
m
✅ thought for 222.4 seconds
You’re seeing that error because inside a task, 
get_run_context()
returns a 
TaskRunContext
, and 
TaskRunContext
does not have a 
flow_run
attribute. Use one of these two patterns in Prefect 3.x: 1) Easiest: grab the FlowRunContext from within the task
Copy code
from prefect import flow, task
from prefect.context import FlowRunContext

@task
def my_task():
    flow_ctx = FlowRunContext.get()  # available even inside tasks
    if flow_ctx and flow_ctx.flow_run and flow_ctx.flow_run.start_time:
        return flow_ctx.flow_run.start_time
    # fallback if somehow not set yet
    from prefect.runtime import flow_run as fr
    return fr.scheduled_start_time  # not exact, but available
2) Robust fallback: fetch the FlowRun via the client using the task’s parent flow_run_id
Copy code
from prefect import task, get_client
from prefect.context import TaskRunContext
from prefect.utilities.asyncutils import run_coro_as_sync

@task
def my_task():
    task_ctx = TaskRunContext.get()
    flow_run_id = task_ctx.task_run.flow_run_id

    async def _load():
        async with get_client() as client:
            return await client.read_flow_run(flow_run_id)

    flow_run = run_coro_as_sync(_load())
    return flow_run.start_time
Notes: - 
prefect.runtime.flow_run.scheduled_start_time
is the scheduled time; the exact execution start time is 
flow_run.start_time
. - If you’re importing schemas for typing, prefer 
from prefect.schemas import FlowRun, State
over 
prefect.server.schemas.core
in user code.
2 Views
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by