Luke He
10/24/2025, 3:17 PMMarvin
10/24/2025, 3:17 PMMarvin
10/24/2025, 3:19 PM@task decorator itself does not re-run your script. But your module-level code can be executed multiple times depending on how you run your flow.
Why your top-level datetime may change:
- Each flow run (e.g., via a deployment/worker) starts a fresh Python process that imports your module. Importing your module executes top-level code again, so a global like RUN_HOUR = datetime.today() will be recomputed for that run.
- If you use a process-based task runner (e.g., ProcessPool, Dask, Ray) or run on platforms that use “spawn” for multiprocessing (Windows/macOS), each worker process imports your module separately. That can recompute your top-level variable at different times.
- With the default thread-based runner in Prefect 3, tasks run in the same process; in that case a top-level datetime set at import should not change within the same flow run.
Recommended pattern
- Compute the “run timestamp” once inside the flow body and pass it into tasks. This ensures consistency even if tasks run in separate processes.
Example:
from datetime import datetime, timezone
from prefect import flow, task
@task
def use_ts(ts):
print(f"ts: {ts}")
@flow
def my_flow():
run_ts = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)
use_ts.submit(run_ts)
if __name__ == "__main__":
my_flow()
Or base it on the scheduled start time from Prefect runtime (stable across retries and processes):
import pendulum
from prefect import flow, task
from prefect.runtime import flow_run
@task
def use_ts(ts):
print(ts)
@flow
def my_flow():
base = flow_run.scheduled_start_time or pendulum.now("UTC")
run_ts = base.replace(minute=0, second=0, microsecond=0)
use_ts.submit(run_ts)
If you share how you’re running (local script vs. deployment/worker, which task runner, and OS), I can pinpoint exactly why you’re seeing the change and suggest the best fix.
Helpful docs:
- Task runners and execution model
- Runtime context (e.g., scheduled_start_time)