Hey Prefect community I m working with Orion and trying to c Prefect Community #ask-community

Hey Prefect community, I’m working with Orion and...

Cole Murray

07/08/2022, 3:27 AM

Hey Prefect community, I’m working with Orion and trying to create a deployment from the API client. I’m able to create it from the CLI, but for my use-case would prefer calling with the api client to ease of use / response parsing. It appears we haven’t updated this yet on the docs page: https://orion-docs.prefect.io/concepts/deployments/#with-the-api Playing around, I have something like this:

Copy code

from typing import List
from uuid import uuid4

from prefect.client import get_client
from prefect.flow_runners import SubprocessFlowRunner
from prefect.flows import Flow
from prefect.orion.schemas.data import DataDocument
from prefect.orion.schemas.schedules import CronSchedule
from workflow_etl.flows.flow import hello_world

def main():
    prefect_client = get_client()

    schedules = [] # TODO

    for schedule in schedules:
        flow_id = await prefect_client.create_flow(hello_world)
        prefect_client.create_deployment(flow_id=flow_id,
                                         name=schedule.id,
                                         schedule=CronSchedule(cron="0 * * * * *"),
                                         parameters={
                                             **schedule.workflow_params
                                         },
                                         tags={
                                             'owner_id': schedule.owner_id,
                                         },
                                         flow_runner=SubprocessFlowRunner(),
                                         flow_data=# UNKNOWN WHAT TO PUT HERE
                                         )

I’m a bit stuck on what to put in the flow_data argument. Anyone tried this / have a link to sample?

Cole Murray

07/08/2022, 3:39 AM

Reverse engineering the CLI, it looks like I’d need to create a new Block document and write it to storage: https://github.com/PrefectHQ/prefect/blob/def2581947449ac271ba8b82d8b6f210d9028ade/src/prefect/packaging/script.py#L107 Might be easier to create the DeploymentSpec with the parse_obj and call create on it

Cole Murray

07/08/2022, 4:55 AM

Ended up with this as a proxy way of using the api client and avoiding CLI:

Copy code

from typing import List

from prefect.deployments import DeploymentSpec
from prefect.flow_runners import SubprocessFlowRunner
from prefect.orion.schemas.schedules import CronSchedule
from workflow_etl.flows.flow import hello_world

def main():
    schedules = [] # Redacted

    for schedule in schedules:
        spec = DeploymentSpec(
            name=schedule.id,
            flow=hello_world,
            parameters={
                **schedule.workflow_params
            },
            schedule=CronSchedule(cron=schedule.cron),
            tags=[
                f"owner_id={schedule.owner_id}",
            ],
            flow_runner=SubprocessFlowRunner()
        )

        spec.validate()
        spec.create()


if __name__ == '__main__':
    main()

Anna Geller

07/08/2022, 7:41 AM

I wanted to ask if you'd be willing to wait a bit since this is something we are working on/on a roadmap but glad that you figured it out

5 Views

Open in Slack

Previous Next