anyio

OrionAgent

anyio.run()

# -*- coding: utf-8
from __future__ import unicode_literals

"""
Custom Django management command to work with Prefect.
Via various flags, you can run a specific flow on demand,
register flows with the server, or run a local agent.
"""
import anyio
import os
from typing import Any

from django.core.management.base import BaseCommand, CommandError, CommandParser
from prefect.deployments import LocalFileSystem, Deployment
from prefect.agent import OrionAgent

# from deployments.flows.reset_timeseries_end_times import (
#     flow as reset_timeseries_end_times,
# )
# from deployments.flows import default_dataset_refresh

flows = {
    # "reset_timeseries_end_times": reset_timeseries_end_times,
    # default_dataset_refresh.flow_name: default_dataset_refresh.flow
}

# daily = Schedule(clocks=[CronClock("5 6 * * *")])


class Command(BaseCommand):
    help = "Run Prefect flow registration, agent, or a single flow"

    def add_arguments(self, parser: CommandParser) -> None:
        parser.add_argument(
            "--agent", help="Run local Prefect Agent", action="store_true"
        )
        parser.add_argument("--run_flow", help="Run a specified flow by name")

    def handle(self, *args: Any, **options: Any):
        if options["run_flow"]:
            try:
                flow = flows[options["run_flow"]]
            except KeyError:
                raise CommandError(
                    f"Specified flow ({options['run_flow']}) does not exist"
                )

            flow.run()

        if options["agent"]:
            anyio.run(self.agent)
        else:
            self.stdout.write("Not running local Prefect Agent")

    async def agent(self):
        """ Run a local Prefect agent """
        self.stdout.write("Running local Prefect Agent")

        agent = OrionAgent(["tenzinga_django_dev"])
        await agent.start()

(ppower) ┌─[ballisticpain@BallisticDevelopment] - [~/Development/tenzinga/ppower] - [Thu Oct 13, 14:05]
└─[$] <git:(feature/prefect-tasks*)> ./manage.py prefect --agent
Running local Prefect Agent

django-prefect

run_deployment()