<@ULVA73B9P> Do functions with a "@task" decorator...
# ask-marvin
l
@Marvin Do functions with a "@task" decorator re-run the script when they're called? We have a variable at the start of the script outside of any function definitions that stores the current datetime with datetime.today() up to the hour; when the task is called later, and the hour has changed, referencing that variable, the timestamp is different
m
thought for 103.0 seconds
Short answer: No—the
@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:
Copy code
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):
Copy code
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)