orion_on_schedule.py

"""Prefect scheduling example in Orion"""
import datetime
import subprocess

from prefect import task, flow
from prefect.client import OrionClient
from prefect.deployments import DeploymentSpec, create_deployment_from_spec
from prefect.orion.services.scheduler import Scheduler
from prefect.orion.schemas.schedules import IntervalSchedule


@task
def say_hello():
    print("Hello, world!")


@flow(name="greetings")
def greetings_flow():
    print("greetings flow")
    say_hello()


def main():
    schedule = IntervalSchedule(interval=datetime.timedelta(seconds=10))
    deployment_spec = DeploymentSpec(name="hola", flow=greetings_flow, schedule=schedule)
    client = OrionClient()
    deployment_id = create_deployment_from_spec(spec=deployment_spec, client=client)
    subprocess.run("prefect orion start", shell=True)
    # Not working yet:
    # deployment = client.create_deployment(deployment_spec)
    # client.create_flow_run_from_deployment(deployment_spec)
    # greetings_flow()


if __name__ == '__main__':
    main()