get_client()

sync_client

False

True

SyncPrefectClient

PrefectClient

await

python
from prefect import flow
from prefect import get_client

@flow
def orchestrate_deployments():
    client = get_client(sync_client=True)
    
    # Deploy and run first flow
    first_deployment = first_flow.from_source(
        source="<https://github.com/your/repo>",
        entrypoint="path/to/flow.py:flow_name"
    ).deploy(...)
    
    # Create and wait for first flow run
    first_run = client.create_flow_run(deployment=first_deployment.id)
    first_run_result = client.wait_for_flow_run(first_run.id, timeout=0)  # timeout=0 means wait indefinitely
    
    # Only proceed if the first flow succeeded
    if first_run_result.state.is_completed():
        # Deploy and run second flow
        second_deployment = second_flow.from_source(...)
        ...

deployment.py

run.py

run.py

python
deployment_manager.from_source(...).deploy()

python
# deployment.py
from prefect import flow, get_client
from prefect.states import State

@flow(name="deployment-manager", retries=0)
def deployment_manager():
    # Use sync client
    client = get_client(sync_client=True)
    
    # Deploy flow1
    print("Deploying flow1...")
    flow1_deployment = flow1.from_source(
        source="<https://github.com/your/repo>",
        entrypoint="code.py:flow1"
    ).deploy(
        name="flow1-deployment",
        work_pool_name="your-work-pool",
        skip_upload=True  # Add this to prevent multiple uploads
    )
    
    print(f"Created deployment: {flow1_deployment.id}")
    
    # Run flow1 and wait
    print("Starting flow1...")
    flow1_run = client.create_flow_run(
        deployment=flow1_deployment.id,
        state=State(name="SCHEDULED")  # Explicitly set initial state
    )
    
    print(f"Waiting for flow1 run {flow1_run.id}...")
    flow1_result = client.wait_for_flow_run(
        flow1_run.id,
        timeout=0,  # wait indefinitely
        return_on_states=[State.COMPLETED, State.FAILED, State.CRASHED]
    )
    
    print(f"Flow1 finished with state: {flow1_result.state.name}")
    
    if not flow1_result.state.is_completed():
        raise Exception(f"Flow1 failed with state: {flow1_result.state.name}")
    
    # Similar pattern for flow2
    print("Deploying flow2...")
    # ... rest of the code

run.py

python
# run.py
from deployment import deployment_manager

if __name__ == "__main__":
    deployment_manager()

run.py

python
existing_deployments = client.read_deployments(
    name="flow1-deployment"
)
if existing_deployments:
    # Either use existing deployment or delete it first
    pass

python
# You can check work pool status
work_pools = client.read_work_pools()
workers = client.read_workers()

run.py

bash
prefect work-pool create "process-pool" --type process

bash
pip install prefect-github

python
from prefect_github import GitHubRepository
from prefect.client.schemas.schedules import CronSchedule

# For private repositories, create a GitHub block
github_credentials = GitHubRepository.create(
    name="my-github",
    repository="<https://github.com/your-org/your-repo>",
    reference="main",  # or your branch name
    access_token="your-token"  # if it's a private repository
)
github_credentials.save("my-github")

def create_deployment():
    from deployment import main
    
    # For public repository:
    deployment_id = main.from_source(
        source="<https://github.com/your-org/your-repo>",
        entrypoint="deployment.py:main"
    ).deploy(
        name="deployment-manager",
        work_pool_name="default-agent-pool",
        schedule=CronSchedule(cron="18 11 * * *"),
        tags=["deployment", "manager", "scheduled"],
        ignore_warnings=True
    )

    # OR for private repository using the block:
    deployment_id = main.from_source(
        source="github/my-github",  # Reference the saved block
        entrypoint="deployment.py:main"
    ).deploy(
        name="deployment-manager",
        work_pool_name="default-agent-pool",
        schedule=CronSchedule(cron="18 11 * * *"),
        tags=["deployment", "manager", "scheduled"],
        ignore_warnings=True
    )

yaml
# deployment.yaml
deployments:
  - name: "deployment-manager"
    entrypoint: "deployment.py:main"  # your flow function
    schedule:
      cron: "18 11 * * *"
    tags:
      - "deployment"
      - "manager"
      - "scheduled"
    work_pool_name: "default-agent-pool"

    # For public repository:
    source:
      type: "github"
      repository: "<https://github.com/your-org/your-repo>"
      reference: "main"  # or your branch name

    # OR for private repository:
    # source:
    #   type: "github"
    #   repository: "<https://github.com/your-org/your-repo>"
    #   reference: "main"  # or your branch name
    #   access_token: "{{ GITHUB_ACCESS_TOKEN }}"  # Use environment variable

bash
prefect deploy --file deployment.yaml

bash
# Set your GitHub token as an environment variable
export GITHUB_ACCESS_TOKEN="your-token"

# Then deploy
prefect deploy --file deployment.yaml

python
# In a Python script or notebook
from prefect_github import GitHubRepository

github_block = GitHubRepository.create(
    name="my-github",
    repository="<https://github.com/your-org/your-repo>",
    reference="main",
    access_token="your-token"
)
github_block.save("my-github")

yaml
# deployment.yaml
deployments:
  - name: "deployment-manager"
    entrypoint: "deployment.py:main"
    schedule:
      cron: "18 11 * * *"
    tags:
      - "deployment"
      - "manager"
      - "scheduled"
    work_pool_name: "default-agent-pool"
    source:
      type: "github"
      block: "github/my-github"  # Reference the saved block

entrypoint

work_pool_name

yaml
deployments:
  - name: "deployment-manager"
    entrypoint: "deployment.py:main"
    schedule:
      cron: "18 11 * * *"
    tags:
      - "deployment"
      - "manager"
      - "scheduled"
    work_pool_name: "default-agent-pool"
    source:
      type: "github"
      repository: "<https://github.com/your-org/your-repo>"
      reference: "main"
    is_schedule_active: true  # Enable/disable schedule
    parameters: {}  # Add any flow parameters here
    enforce_parameter_schema: true
    version: null  # Optional version tag
    description: "Deployment manager flow"  # Optional description
    path: null  # Optional path within the repository
    pull_steps:  # Optional pull steps configuration
      - prefect.deployments.steps.git_clone

bash
prefect deploy --file deployment.yaml

bash
prefect deployment ls

python
from prefect import flow
from prefect.client import get_client
from prefect.deployments import run_deployment
from prefect.flows import load_flow_from_source

@flow(name="deployment_creator")
def create_and_run_deployments():
    client = get_client(sync_client=True)
    
    # 1. Create first flow deployment
    flow1_deployment = flow1.from_source(
        source="<https://github.com/your-org/your-repo>",
        entrypoint="flows.py:flow1"
    ).deploy(
        name="flow1-deployment",
        work_pool_name="your-work-pool",
        tags=["flow1"]
    )
    
    # 2. Run the deployment and wait for completion
    flow1_run = run_deployment(
        name="flow1-deployment",
        timeout=0,  # wait indefinitely
        as_subflow=False  # Run as independent flow, not as subflow
    )
    
    # 3. Only proceed if flow1 succeeded
    if not flow1_run.state.is_completed():
        raise Exception(f"Flow1 failed with state: {flow1_run.state.name}")
    
    # 4. Create and run second flow deployment
    flow2_deployment = flow2.from_source(
        source="<https://github.com/your-org/your-repo>",
        entrypoint="flows.py:flow2"
    ).deploy(
        name="flow2-deployment",
        work_pool_name="your-work-pool",
        tags=["flow2"]
    )
    
    flow2_run = run_deployment(
        name="flow2-deployment",
        timeout=0,
        as_subflow=False
    )
    
    return "All deployments completed"

# For scheduled execution of the deployment creator
@flow(name="scheduled_deployment_manager")
def scheduled_deployment_manager():
    return create_and_run_deployments()

as_subflow=False

run_deployment

python
# deployment.yaml
deployments:
  - name: "deployment-manager"
    entrypoint: "your_file.py:scheduled_deployment_manager"
    schedule:
      cron: "0 0 * * *"  # Example schedule
    work_pool_name: "your-work-pool"
    source:
      type: "github"
      repository: "<https://github.com/your-org/your-repo>"
      reference: "main"

python
scheduled_deployment_manager.from_source(
    source="<https://github.com/your-org/your-repo>",
    entrypoint="your_file.py:scheduled_deployment_manager"
).deploy(
    name="deployment-manager",
    work_pool_name="your-work-pool",
    schedule=(CronSchedule(cron="0 0 * * *")),
)

python
client = get_client(sync_client=True)
deployments = client.read_deployments()
for deployment in deployments:
    print(f"Deployment {deployment.name}: {deployment.status}")

as_subflow

True

as_subflow=True

as_subflow=False

python
from prefect import flow
from prefect.deployments import run_deployment

@flow
def orchestrator_flow():
    # This will run as a subflow (default behavior)
    run1 = run_deployment(
        name="flow1-deployment"  # as_subflow defaults to True
    )
    
    # This will run as an independent flow
    run2 = run_deployment(
        name="flow2-deployment",
        as_subflow=False
    )

as_subflow=False

from_source

to_deployment

as_subflow

as_subflow

python
# First create the deployment as you're doing
flow_from_source = self.flows[index].from_source(
    source=source_dir,
    entrypoint=self.entrypoints[index]
)

# Create deployment
deployment = flow_from_source.to_deployment(
    name=deployment_name,
    work_pool_name=work_pool_name,
    job_variables=self.job_variables[index]
)

# Deploy it
deployment_id = deployment.apply()

# Then when you want to run it, you can control as_subflow
from prefect.deployments import run_deployment

# Run the deployment with as_subflow=False
flow_run = run_deployment(
    name=deployment_name,  # or f"{deployment.flow_name}/{deployment_name}"
    as_subflow=False,     # This will run it independently
    timeout=0             # Optional: wait indefinitely for completion
)

python
from prefect import flow
from prefect.deployments import run_deployment

@flow
def create_and_run_deployments():
    deployments = []
    
    # Create all deployments first
    for index in range(len(self.flows)):
        flow_from_source = self.flows[index].from_source(
            source=source_dir,
            entrypoint=self.entrypoints[index]
        )
        
        deployment = flow_from_source.to_deployment(
            name=f"deployment-{index}",
            work_pool_name=work_pool_name,
            job_variables=self.job_variables[index]
        )
        
        deployment_id = deployment.apply()
        deployments.append(deployment)
    
    # Then run them sequentially
    for deployment in deployments:
        flow_run = run_deployment(
            name=deployment.name,
            as_subflow=False,  # Run independently
            timeout=0          # Wait for completion
        )
        
        # Optional: Check if the run completed successfully
        if not flow_run.state.is_completed():
            raise Exception(f"Deployment {deployment.name} failed with state: {flow_run.state.name}")

to_deployment()

apply()

run_deployment()

as_subflow

as_subflow=False

parent_task_run_id=None

wait_for_flow_run

python
from prefect import flow

@flow
def subflow(x):
    return x * 2

@flow
def parent_flow():
    # This creates a subflow
    result = subflow(5)
    return result

python
@flow
def resource_intensive_subflow():
    # Uses same resources as parent
    large_computation()

@flow
def parent_flow():
    # These run in the same process
    for _ in range(10):
        resource_intensive_subflow()  # Might cause memory/CPU issues

python
@flow
def subflow():
    raise Exception("Subflow error")

@flow
def parent_flow():
    try:
        subflow()  # Parent flow will fail if subflow fails
    except Exception as e:
        # Need explicit handling
        pass

python
@flow
def deep_subflow():
    return "deep"

@flow
def middle_subflow():
    return deep_subflow()

@flow
def parent_flow():
    return middle_subflow()  # Can become hard to debug

python
from prefect.deployments import run_deployment

@flow
def orchestrator_flow():
    # Run as independent flow, not subflow
    run_deployment(
        name="other-flow/deployment",
        as_subflow=False
    )

python
@flow
def data_processing_subflow(data):
    return process(data)

@flow
def main_flow():
    data = load_data()
    # Use subflow for logical grouping
    processed = data_processing_subflow(data)

python
@flow
def parent_flow():
    # Run subflows in parallel with care
    from prefect.tasks import task_input_hash
    
    results = []
    for item in items:
        result = subflow.submit(item)  # Careful with memory
        results.append(result)

python
from prefect import flow, task
from typing import List

@task
def check_resources():
    # Verify available memory/CPU
    pass

@flow
def safe_subflow(data: List):
    check_resources()
    return process_chunk(data)

@flow
def parent_flow(all_data: List):
    # Break into manageable chunks
    chunks = split_data(all_data)
    
    results = []
    for chunk in chunks:
        # Process in controlled manner
        result = safe_subflow(chunk)
        results.append(result)
    
    return combine_results(results)

python
from prefect import flow
from prefect.deployments import run_deployment

# Instead of this (subflow that can't use different work pool):
@flow
def subflow():
    return "result"

@flow
def parent_flow():
    result = subflow()  # Will run in same work pool as parent

# Do this instead (independent flows that can use different work pools):
@flow
def flow_to_run_separately():
    return "result"

@flow
def orchestrator_flow():
    # This will run in a different work pool
    result = run_deployment(
        name="flow-to-run-separately/deployment-name",
        as_subflow=False,  # Important: run as independent flow
        work_queue_name="different-work-queue"  # Optional: specify work queue
    )

# Deploy the flows to different work pools
flow_to_run_separately.from_source(
    source="your_source",
    entrypoint="flows.py:flow_to_run_separately"
).deploy(
    name="deployment-name",
    work_pool_name="different-work-pool"  # This flow uses a different work pool
)

orchestrator_flow.from_source(
    source="your_source",
    entrypoint="flows.py:orchestrator_flow"
).deploy(
    name="orchestrator-deployment",
    work_pool_name="main-work-pool"  # This flow uses the main work pool
)

python
from prefect import flow
from prefect import get_client

@flow
def orchestrator_flow():
    client = get_client(sync_client=True)
    
    # Create flow run in different work pool
    flow_run = client.create_flow_run_from_deployment(
        deployment_id="deployment-id",
        parent_task_run_id=None,  # Run independently
    )
    
    # Wait for completion if needed
    final_state = client.wait_for_flow_run(
        flow_run_id=str(flow_run.id),
        timeout=0
    )

run_deployment

as_subflow=False

python
# deployment_setup.py
def setup_deployments():
    # Create all deployments here, outside of any flow
    flow1.from_source(
        source="your_repo",
        entrypoint="flows.py:flow1"
    ).deploy(
        name="flow1-deployment",
        work_pool_name="your-pool"
    )
    
    flow2.from_source(
        source="your_repo",
        entrypoint="flows.py:flow2"
    ).deploy(
        name="flow2-deployment",
        work_pool_name="your-pool"
    )

if __name__ == "__main__":
    setup_deployments()

python
from prefect import flow
from prefect import get_client

@flow
def deployment_manager():
    client = get_client(sync_client=True)
    
    # Check if deployment already exists
    existing_deployments = client.read_deployments(
        name="flow1-deployment"
    )
    
    if not existing_deployments:
        # Only create if it doesn't exist
        new_deployment = flow1.from_source(
            source="your_repo",
            entrypoint="flows.py:flow1"
        ).deploy(
            name="flow1-deployment",
            work_pool_name="your-pool",
            schedule=None  # Important: avoid automatic scheduling
        )
    
    # Run the deployment
    flow_run = client.create_flow_run_from_deployment(
        deployment=existing_deployments[0].id if existing_deployments else new_deployment.id
    )

python
from prefect.deployments import DeploymentEventTrigger

# Create deployments with event triggers
flow2.from_source(
    source="your_repo",
    entrypoint="flows.py:flow2"
).deploy(
    name="flow2-deployment",
    work_pool_name="your-pool",
    triggers=[
        DeploymentEventTrigger(
            deployment_name="flow1-deployment",
            event_type="completed"
        )
    ]
)

python
# Create deployments in a separate script
# run.py
def setup():
    create_all_deployments()

def run_orchestration():
    run_existing_deployments()

if __name__ == "__main__":
    setup()  # Run once
    run_orchestration()  # Run as needed

python
@flow
def safe_deployment_manager():
    client = get_client(sync_client=True)
    
    def get_or_create_deployment(name, flow_obj, **kwargs):
        existing = client.read_deployments(name=name)
        if existing:
            return existing[0]
        
        return flow_obj.from_source(...).deploy(
            name=name,
            **kwargs
        )

python
@flow
def orchestrate_flows():
    client = get_client(sync_client=True)
    
    # Create flow runs with explicit control
    flow_run = client.create_flow_run_from_deployment(
        deployment_id=deployment_id,
        state=None,  # Don't auto-schedule
        parent_task_run_id=None  # Run independently
    )
    
    # Wait for completion before proceeding
    final_state = client.wait_for_flow_run(
        flow_run_id=str(flow_run.id),
        timeout=0
    )

python
@flow
def versioned_deployment_manager():
    deployment = flow1.from_source(
        source="your_repo",
        entrypoint="flows.py:flow1"
    ).deploy(
        name=f"flow1-deployment-{version}",  # Version your deployments
        work_pool_name="your-pool",
        version=version  # Track versions
    )

python
from prefect import flow
from prefect import get_client
from prefect.artifacts import create_markdown_artifact

@flow
def limited_flow():
    client = get_client(sync_client=True)
    
    # Get count of previous runs
    flow_runs = client.read_flow_runs(
        flow_name="limited_flow",
        sort="-start_time"
    )
    
    max_runs = 5  # Set your desired limit
    current_runs = len(flow_runs)
    
    if current_runs >= max_runs:
        message = f"Flow has reached maximum runs ({max_runs})"
        create_markdown_artifact(
            key="max_runs_reached",
            markdown=message,
            description="Maximum run limit reached"
        )
        raise Exception(message)
    
    # Your flow logic here
    return "Flow executed successfully"

python
from prefect import flow
from prefect.client.schemas.objects import ConcurrencyLimit

@flow
def flow_with_concurrency():
    client = get_client(sync_client=True)
    
    # Create or update concurrency limit
    client.create_concurrency_limit(
        tag="my-flow-limit",
        concurrency_limit=5
    )
    
    # Your flow logic here

python
from prefect import flow

@flow
def flow_with_queue_limit():
    client = get_client(sync_client=True)
    
    # Set work queue concurrency limit
    client.update_work_queue(
        name="your-work-queue",
        concurrency_limit=5
    )

python
from prefect import flow
from prefect.blocks.system import JSON

@flow
def flow_with_persistent_counter():
    try:
        counter_block = JSON.load("run-counter")
        count = counter_block.value.get("count", 0)
    except:
        counter_block = JSON(value={"count": 0})
        count = 0
    
    max_runs = 5
    if count >= max_runs:
        raise Exception(f"Maximum runs ({max_runs}) reached")
    
    # Increment counter
    counter_block.value["count"] = count + 1
    counter_block.save("run-counter")
    
    # Your flow logic here

python
from datetime import timedelta
from prefect.deployments import Deployment
from prefect.server.schemas.schedules import CronSchedule

# Limit runs by scheduling window
deployment = Deployment.build_from_flow(
    flow=your_flow,
    name="limited-deployment",
    schedule=CronSchedule(
        cron="0 0 * * *",  # Daily at midnight
        timezone="UTC",
        end_date=datetime.utcnow() + timedelta(days=5)  # Run for 5 days only
    )
)

python
from prefect import flow, tags

@flow
def flow_with_tag_limit():
    with tags("limited-flow"):
        client = get_client(sync_client=True)
        
        # Set concurrency limit for tag
        client.create_concurrency_limit(
            tag="limited-flow",
            concurrency_limit=5
        )
        
        # Your flow logic here

python
from prefect import flow
from prefect.states import Cancelled

@flow
def monitored_flow():
    client = get_client(sync_client=True)
    
    flow_runs = client.read_flow_runs(
        flow_name="monitored_flow",
        state={"type": "COMPLETED"}
    )
    
    if len(flow_runs) >= 5:
        # Cancel any future runs
        client.set_flow_run_state(
            flow_run_id=flow_run_id,
            state=Cancelled(message="Maximum run limit reached")
        )
        return
    
    # Your flow logic here

python
@flow(description="This flow is limited to 5 runs")
def limited_flow():
    # Add documentation about limits
    pass

3. *Monitor Limits*:
```python
@flow
def monitored_limited_flow():
    client = get_client(sync_client=True)
    
    # Track run counts
    runs = client.read_flow_runs(
        flow_name="monitored_limited_flow"
    )
    
    # Create monitoring artifact
    create_markdown_artifact(
        key="run-count",
        markdown=f"Current run count: {len(runs)}"
    )