How to generate a custom SDK for your deployments

prefect sdk generate --output ./my_sdk.py

prefect sdk generate --output ./my_sdk.py --flow my-etl-flow

prefect sdk generate --output ./my_sdk.py --deployment my-flow/production

prefect sdk generate --output ./my_sdk.py \
    --flow etl-flow \
    --flow data-sync \
    --deployment analytics/daily

from my_sdk import deployments

# Get a deployment by name
deployment = deployments.from_name("my-etl-flow/production")

# Run with parameters
future = deployment.run(
    source="s3://my-bucket/data",
    batch_size=100,
)

# Get the flow run ID immediately
print(f"Started flow run: {future.flow_run_id}")

# Wait for completion and get result
result = future.result()

from my_sdk import deployments
from datetime import datetime, timedelta

future = deployments.from_name("my-etl-flow/production").with_options(
    tags=["manual", "production"],
    idempotency_key="daily-run-2024-01-15",
    scheduled_time=datetime.now() + timedelta(hours=1),
    flow_run_name="custom-run-name",
).run(
    source="s3://bucket",
)

from my_sdk import deployments

future = deployments.from_name("my-etl-flow/production").with_infra(
    image="my-registry/my-image:latest",
    cpu_request="2",
    memory="8Gi",
).run(
    source="s3://bucket",
)

import asyncio
from my_sdk import deployments

async def trigger_deployment():
    future = await deployments.from_name("my-etl-flow/production").run_async(
        source="s3://bucket",
    )
    result = await future.result()
    return result

# Run it
result = asyncio.run(trigger_deployment())

from my_sdk import deployments

future = (
    deployments.from_name("my-etl-flow/production")
    .with_options(tags=["production"])
    .with_infra(memory="8Gi")
    .run(source="s3://bucket", batch_size=100)
)

prefect sdk generate --output ./my_sdk.py