Jack Sundberg
07/31/2022, 11:23 PMdeployment = Deployment(
name="example-deploy",
flow=my_flow_object, # defined elsewhere
packager=OrionPackager(serializer=PickleSerializer()),
tags=[
"some-tag1",
"some-tag2",
],
)
deployment_id = await deployment.create()
But this code is now broken and the docs recommend sticking to yaml files. How would I go about updating this? Can we still specify a packager
and is it possible to create a deployment in python without creating a yaml file?Khuyen Tran
08/01/2022, 12:59 AMDeployment
is no longer supported in 2.0. I recommend that you run the following:
prefect deployment build name-to-your-flow:my-flow-object -t "some-tag1" -t "some-tag2"
at the directory that contains your flow and its importsJack Sundberg
08/01/2022, 1:02 AMfrom my_package.workflows import example
prefect_flow = example.to_prefect_flow()
Anna Geller
08/01/2022, 9:27 AMOwen Cook
08/01/2022, 10:09 AMOrionClient.create_deployment
or is there a more efficient way?Anna Geller
08/01/2022, 11:13 AMfrom prefect.infrastructure import DockerContainer
infra_block = DockerContainer()
infra_block.save("docker")
same for file system
from prefect.filesystems import S3
block = S3()
block.save("dev")
one you have your blocks and flows, you can even run a subprocess to build and apply
we anticipate most users will be creating deployments from CI/CD after a PR review and merge, that's why it's optimized for CLI/API-driven UXOwen Cook
08/01/2022, 1:14 PMJack Sundberg
08/01/2022, 3:09 PMDeployment
and PickleSerializer
are now staged for removal.
I may be confused on the new yaml approach, but it still seems like a single file is required for deployments. Seems like a major restriction.
PS -- yes, I often run in a local process, but I build in the option for my users to switch where runs complete if they'd like to use Docker etc..Deployment.create()
method contains a higher-level call to client.create_deployment
though the method currently contains a bug and it's depreciated 😅Owen Cook
08/01/2022, 7:41 PM"""Create deployment using prefect api."""
import asyncio
import json
import time
from pathlib import Path
from prefect import Manifest, flow, get_client
from prefect.client import OrionClient
from prefect.deployments import FlowScript
from prefect.filesystems import LocalFileSystem
from prefect.flows import Flow, load_flow_from_script
from prefect.infrastructure import Process
from prefect.packaging.orion import OrionPackager
from prefect.utilities.callables import parameter_schema
@flow
def example_flow(num: int) -> None:
"""Example flow."""
for i in range(10):
print(i + 1)
time.sleep(0.1)
print(f"Input number: {num}")
def flow_from_flow_script(flow_script: FlowScript) -> Flow:
"""Load flow from a flow script.
Args:
flow_script (FlowScript): Flow script to be turned into flow.
Returns:
Flow: flow to be produced from flow script.
"""
return load_flow_from_script(
flow_script.path.expanduser().resolve(), flow_name=flow_script.name
)
async def create_deployment() -> None:
"""Create a deployment."""
client: OrionClient = get_client()
# set the flow
my_flow = flow_from_flow_script(
FlowScript(
path=Path(__file__).parent / "deployment.py",
name="example-flow",
)
)
# create manifest from flow (OrionPackager)
manifest = await OrionPackager().package(my_flow)
# create flow and get flow id
flow_id = await client.create_flow(my_flow)
# create infrastructure
updates = {}
if "image" in manifest.__fields__:
updates["image"] = manifest.image
infrastructure = Process()
infrastructure = infrastructure.copy(update=updates)
infrastructure_id = await infrastructure._save(is_anonymous=True)
# set up storage
storage = LocalFileSystem(basepath=Path(__file__).parent.absolute())
storage_id = await storage._save(is_anonymous=True)
# set up parameters
params = {"num": 5}
openapi_schema = parameter_schema(my_flow)
# get the manifest
manifest_path = f"{my_flow.fn.__name__}-manifest.json"
manifest = Manifest(
flow_name=my_flow.name,
import_path=f"{Path(__file__)}:{my_flow.fn.__name__}",
parameter_openapi_schema=openapi_schema,
)
with open(manifest_path, "w", encoding="UTF-8") as manifest_file:
json.dump(manifest.dict(), manifest_file, indent=4)
# create deployment
return await client.create_deployment(
flow_id=flow_id,
name=my_flow.name,
parameters=params,
description="Some description.",
tags=["test"],
manifest_path=manifest_path,
storage_document_id=storage_id,
infrastructure_document_id=infrastructure_id,
parameter_openapi_schema=openapi_schema,
)
asyncio.run(create_deployment())
Jack Sundberg
08/02/2022, 2:17 PMasync with get_client() as client:
. Also I need to make sure the solution also doesn't depend on a single file script (e.g. deployment.py
) with the flow in it.Owen Cook
08/02/2022, 2:54 PMAnna Geller
08/02/2022, 5:00 PMJack Sundberg
08/02/2022, 5:07 PM