Great use case — you want an alert if the final ETL flow hasn’t finished by 5:00 AM on weekdays. In Prefect 3.x, you’ve got two good patterns. Pick the one that best matches how your ETL runs. Option A: Use a Time-to-Completion SLA (Cloud, simplest if start time is predictable) - Best when the final flow starts at a consistent time (e.g., midnight). You attach a 5-hour SLA to the deployment; if the run exceeds that duration, Prefect emits an SLA violation event you can notify on. - Add an SLA when deploying your flow:

from prefect import flow
from prefect._experimental.sla.objects import TimeToCompletionSla

@flow
def final_etl_step():
    ...

# If this flow starts ~00:00 and must be done by 05:00
final_etl_step.deploy(
    name="final-etl",
    work_pool_name="your-pool",
    cron="0 0 * * 1-5",  # if you schedule this flow directly
    _sla=[TimeToCompletionSla(name="finish-by-5am", duration=5*60*60)]
)

- Then create an automation in Prefect Cloud that reacts to SLA violations for this deployment and sends a notification (Slack, email, PagerDuty, etc.). - Docs: - SLAs (Prefect Cloud) - Creating automations - Event triggers Notes: - This is Cloud-only and SLA is duration-based (relative to start). If the final step can start late, this may not align with a fixed 05:00 deadline. Option B: A 5:00 AM “deadline checker” deployment (most flexible, aligns to wall-clock time) - Create a tiny flow scheduled for 05:00 on weekdays that checks whether the specific deployment has a completed flow run since the start of the business day (e.g., since midnight in your TZ). If not, send a notification. - This works regardless of when your ETL starts and exactly matches “by 5:00 AM M–F”. Skeleton you can adapt: ``` from datetime import datetime, time, timezone from zoneinfo import ZoneInfo from prefect import flow, get_run_logger from prefect.blocks.notifications import SlackWebhook from prefect.client.orchestration import get_client # Filters are available if you want to do server-side filtering: # from prefect.client.schemas.filters import ( # DeploymentFilterName, FlowRunFilter, FlowRunFilterStartTime # ) TZ = ZoneInfo("America/New_York") # set your timezone DEPLOYMENT_QUALNAME = "your-flow-name/your-deployment-name" # or use tags @flow async def etl_deadline_check(): logger = get_run_logger() now_local = datetime.now(TZ) day_start_local = datetime.combine(now_local.date(), time.min, tzinfo=TZ) day_start_utc = day_start_local.astimezone(timezone.utc) async with get_client() as client: # Fetch recent runs for the target deployment (you can also fetch by tag) # Keep this simple; we’ll filter in Python. runs = await client.read_flow_runs( limit=200 # adjust as needed # You can pass server-side filters if you prefer: # deployments=DeploymentFilterName(any_=[DEPLOYMENT_QUALNAME]), # flow_runs=FlowRunFilter(start_time=FlowRunFilterStartTime(gte=day_start_utc)), ) # Filter to your deployment and today’s runs todays_runs = [ r for r in runs if r.deployment and f"{r.flow_name}/{r.deployment.name}" == DEPLOYMENT_QUALNAME and r.start_time and r.start_time >= day_start_utc ] completed = any(r.state_type.value == "COMPLETED" for r in todays_runs) if not completed: # Send Slack (or any other notification block) slack = await SlackWebhook.load("my-slack-webhook") # create in UI once msg = ( f"ETL deadline missed: no completed run for {DEPLOYMENT_QUALNAME} " f"by {now_local.strftime('%Y-%m-%d %H:%M %Z')}.\n" f"Found {len(todays_runs)} run(s) since {day_start_local.isoformat()}" ) await slack.notify(msg) logger.warning(msg)

else: logger.info(f"OK: {DEPLOYMENT_QUALNAME} has completed since {day_start_local.isoformat()}")

Deploy it to run at 5:00 on weekdays:

prefect deploy flows/etl_deadline_check.py:etl_deadline_check \ --name etl-deadline-check \ --cron "0 5 * * 1-5" \ --work-pool your-pool ``` - You can swap SlackWebhook for Teams/Email/PagerDuty, etc. See Notification blocks. Which should you choose? - If the final step reliably starts at midnight: Option A (SLA) is minimal config and fires a precise SLA violation event you can automate. - If start times vary or you want a strict 05:00 wall-clock cutoff: Option B (deadline checker) is the most accurate and explicit. A couple of clarifications to tailor this for you: - What timezone should “5:00 AM” use? - What’s the exact flow/deployment name of your final ETL step (or do you use a distinct tag we can match on)? - Do you want the alert only when there’s no Completed run, or also when it’s Failed before 5:00 AM? - Any holiday/weekend exceptions beyond Mon–Fri? Happy to wire this up end-to-end (including a ready-to-run checker flow) once you confirm those details.