Short answer: there isn’t a built-in “I am the last run of the day” flag. In Prefect 3.x, the way to do this from inside a flow is to use the runtime context to get this run’s scheduled time and deployment id, then query the API for any additional scheduled runs for the same deployment later that same day. Here’s a minimal, synchronous example that avoids using “current time” and instead relies on the deployment’s schedule and this run’s scheduled_start_time:

from prefect import flow, runtime
from prefect.client.orchestration import SyncPrefectClient

@flow
def my_flow():
    # Must be running from a deployment for this to work
    dep_id = runtime.deployment.id
    sched = runtime.flow_run.scheduled_start_time  # timezone-aware scheduled time
    if not dep_id or not sched:
        print("Not running from a deployment (or no scheduled time) — cannot determine.")
        return

    # Define end-of-day relative to the scheduled time’s day/timezone
    end_of_day = sched.replace(hour=23, minute=59, second=59, microsecond=999_999)

    with SyncPrefectClient() as client:
        # Ask Prefect for scheduled runs for this deployment up to end-of-day
        # Note: this returns only SCHEDULED runs
        scheduled_runs = client.get_scheduled_flow_runs_for_deployments(
            deployment_ids=[dep_id],
            scheduled_before=end_of_day,
            limit=50,  # any small number is fine; we'll filter below
        )

    # Are there any runs scheduled after this one but before end-of-day?
    remaining_today = [
        r for r in scheduled_runs
        if r.expected_start_time and r.expected_start_time > sched
    ]

    is_last_of_day = len(remaining_today) == 0
    print(f"Is last run of the day? {is_last_of_day}")

If you prefer the more general query API (and/or want to be explicit about filters), you can use

read_flow_runs

with filters on expected start time and state:

from prefect import flow, runtime
from prefect.client.orchestration import SyncPrefectClient
from prefect.client.schemas.filters import (
    FlowRunFilter,
    FlowRunFilterExpectedStartTime,
    FlowRunFilterStateType,
    DeploymentFilter,
)
from prefect.client.schemas.objects import StateType

@flow
def my_flow():
    dep_id = runtime.deployment.id
    sched = runtime.flow_run.scheduled_start_time
    if not dep_id or not sched:
        return

    eod = sched.replace(hour=23, minute=59, second=59, microsecond=999_999)

    with SyncPrefectClient() as client:
        # Look for any scheduled runs later today
        future_runs_today = client.read_flow_runs(
            deployment_filter=DeploymentFilter(id={"any_": [dep_id]}),
            flow_run_filter=FlowRunFilter(
                state_type=FlowRunFilterStateType(any_=[StateType.SCHEDULED]),
                expected_start_time=FlowRunFilterExpectedStartTime(
                    after_=sched,
                    before_=eod,
                ),
            ),
            limit=1,  # we only need to know if at least one exists
        )

    is_last_of_day = len(future_runs_today) == 0
    print(f"Is last run of the day? {is_last_of_day}")

Notes and caveats: - This checks “last scheduled occurrence” based on the deployment’s schedule and this run’s scheduled_start_time, not wall-clock “now”. - Define “day” carefully: the examples use the scheduled time’s own timezone to define end-of-day. If you want a different business timezone, adjust the end-of-day accordingly. - If you have multiple deployments or schedules for the same flow, decide whether “last of the day” should be per-deployment or across all deployments and adjust the query. - Backfills or schedule edits made after the run starts could change what “last” means for that day. Does this match your use case? If you can share which timezone should define the “day” (deployment’s schedule tz, UTC, or a business tz) and whether you have multiple schedules/deployments, I can tailor the snippet.

Yes — in Prefect 3.x you can pass multiple schedules to

to_deployment

and give each schedule its own

parameters

dict. Those per-schedule parameters are applied to runs generated by that schedule and can override the flow’s defaults (and any deployment-level parameters you set). Example with multiple schedules overriding different params:

from prefect import flow
from prefect.schedules import Cron, Interval

@flow
def my_flow(env: str = "dev", batch_size: int = 100):
    ...

deployment = my_flow.to_deployment(
    name="my-multi-schedule-deployment",
    work_pool_name="my-pool",
    # Optional: base parameters for the deployment (used when a schedule does not override)
    parameters={"env": "dev", "batch_size": 100},
    schedules=[
        Cron(
            "0 9 * * MON-FRI",                # Weekdays 9 AM
            timezone="America/New_York",
            parameters={"env": "prod", "batch_size": 500},
            slug="weekday-prod",
        ),
        Cron(
            "0 2 * * SAT",                    # Saturdays 2 AM
            timezone="America/New_York",
            parameters={"env": "staging", "batch_size": 200},
            slug="sat-staging",
        ),
        Interval(
            86400,                            # Daily
            parameters={"env": "dev", "batch_size": 50},
            active=False,                     # Can keep disabled initially
            slug="daily-dev",
        ),
    ],
)

Notes: - You can mix

Cron

,

Interval

, and

RRule

schedules in the same list. - You can also pass raw dicts or

DeploymentScheduleCreate

objects if you prefer. - If a schedule omits a parameter, it falls back to deployment-level parameters (if provided) or the flow’s default.

That error means the Prefect API rejected the deployment payload during apply. HTTP 422 “Unprocessable Entity” is a schema/validation error — something about the deployment or one of its schedules/fields doesn’t conform to what the server expects. Most common causes I see: - Invalid schedule definition - Cron string invalid, or

day_or

handling not supported by your expression - RRule string invalid (e.g., missing “RRULE:” prefix or malformed) - Interval anchor date/timezone invalid - Timezone not recognized (must be an IANA tz like “America/New_York”) - Per‑schedule parameters don’t match the flow’s parameter schema - Unknown parameter names - Wrong types (e.g., string instead of int) - Missing required parameters (when you set

enforce_parameter_schema=True

, which is the default in

to_deployment

) - Job variables don’t match your work pool’s job template - Keys or types that aren’t allowed for that pool’s base job template will fail validation - Naming/format constraints - Invalid deployment/schedule slug or name formatting (must be lowercase letters/numbers/dashes in some cases) - Tags must be a list of strings How to reveal the exact error detail - Turn on debug logs when applying: - Set env var

PREFECT_LOGGING_LEVEL=DEBUG

before running your script, or - If you’re using the CLI (

prefect deploy

), add

--log-level DEBUG

- Catch and print the inner HTTP error body (it usually contains a JSON “detail” explaining the field that failed):

from prefect.exceptions import PrefectHTTPStatusError

try:
    # e.g. my_flow.to_deployment(...) or my_flow.deploy(...)
    ...
except Exception as e:
    inner = getattr(e, "__cause__", None) or getattr(e, "__context__", None)
    print(f"{type(e).__name__}: {e}")
    if inner and hasattr(inner, "response"):
        try:
            print("Server detail:", inner.response.json())
        except Exception:
            print("Server detail (raw):", inner.response.text)

Quick checks you can run - Validate your work pool and its job template: -

prefect work-pool inspect "my-pool"

to see the template your

job_variables

must match - If you suspect schedules, try creating one via CLI to get clearer validation errors: -

prefect deployment schedule create "flow-name/deployment-name" --cron "0 9 * * MON-FRI" --timezone "America/New_York" --parameters '{"env":"prod","batch_size":500}'

- Confirm parameter names/types align with your flow signature and defaults. If you paste: - The code you’re using to apply the deployment (especially

schedules=

,

parameters=

,

work_pool_name

, and

job_variables

) - The full error body (from the snippet above) …I can pinpoint exactly which field is triggering the 422 and suggest the precise fix.