from my_package.workflows import example
prefect_flow = example.to_prefect_flow()

Deployment.create()

client.create_deployment

"""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())

async with get_client() as client:

deployment.py