from datetime import timedelta

import pendulum
from prefect import flow, task
from prefect.context import FlowRunContext, get_run_context
from prefect.states import Failed


MAX_SECONDS_LATE = 10


@task
def my_task():
    pass


@flow
def my_flow():
    ctx: FlowRunContext = get_run_context()
    seconds_late = (
        pendulum.now().diff(ctx.flow_run.expected_start_time, abs=False).in_seconds()
        * -1
    )

    if seconds_late > 10:
        return Failed(message=f"Flow started {seconds_late} seconds late.")

    my_task()



from prefect.deployments import DeploymentSpec
from prefect.flow_runners import SubprocessFlowRunner

DeploymentSpec(
    flow=my_flow,
    flow_runner=SubprocessFlowRunner(),
    schedule={"interval": timedelta(seconds=20)},
)

DaskTaskRunner

UniversalFlowRunner, SubprocessFlowRunner, DockerFlowRunner, KubernetesFlowRunner