Let me show you how to run deployments programmatically in both versions: ### Prefect 2.x

python
from prefect import run_deployment

# Run a deployment by name
flow_run = await run_deployment(
    name="my-flow/my-deployment",  # format: "flow-name/deployment-name"
    parameters={"param1": "value1"}
)

# Or run by deployment ID
flow_run = await run_deployment(
    deployment_id="deployment-id-here",
    parameters={"param1": "value1"}
)

### Prefect 3.x There are two main approaches in Prefect 3.x: 1. Running an existing deployment:

python
from prefect import run_deployment

# Run by name
flow_run = await run_deployment(
    name="my-flow/my-deployment",
    parameters={"param1": "value1"}
)

# Or run by deployment ID
flow_run = await run_deployment(
    deployment_id="deployment-id-here",
    parameters={"param1": "value1"}
)

2. Serving deployments (new in 3.x):

python
from prefect import flow, serve

@flow
def my_flow(param1: str = "default"):
    print(f"Running with {param1}")

# Serve the deployment
if __name__ == "__main__":
    # Synchronous version
    serve(my_flow)
    
    # Or asynchronous version
    # from prefect import aserve
    # await aserve(my_flow)

Key Differences: 1. The basic

run_deployment

functionality remains similar between versions 2. Prefect 3.x introduces the new

serve

and

aserve

functions for running deployments directly 3. The deployment creation process is different: - 2.x used

Deployment.build_from_flow()

(now removed) - 3.x uses

flow.from_source().deploy()

or the

prefect deploy

CLI command For new projects, I recommend using Prefect 3.x as it provides a more streamlined deployment experience with better ergonomics. The

serve

functionality in particular makes it easier to run deployments directly from your code. Links to relevant documentation: - Prefect 3.x Deployment Guide - Prefect 2.x Deployment Concepts

Here's an improved version of your code with several important enhancements:

python
from prefect import flow
from prefect import run_deployment  # Updated import path
from typing import Optional

@flow
async def parent_flow(log_prints: bool = True):
    try:
        flow_run = await run_deployment(
            name="test-prefect-flow/ingest-prefect-test-deployment",
            parameters={
                "script_type": "INGEST",
                "script_name": "PREFECT_TEST_1",
                "minutes": 5
            },
            timeout=3600,  # Optional timeout in seconds
            return_type="flow_run"  # Get the flow run object for monitoring
        )
        return flow_run
    except Exception as e:
        print(f"Error running deployment: {e}")
        raise

Here's why each improvement matters: 1. Correct Import Path: - Changed from

prefect.deployments

to just

prefect

- In Prefect 3.x,

run_deployment

is available directly from the main

prefect

package 2. Async/Await Pattern: - Added

async

to the flow definition and

await

to

run_deployment

-

run_deployment

is an async function, so this prevents blocking behavior and allows proper concurrent execution - This is especially important when running multiple deployments or handling long-running flows 3. Error Handling: - Added try/except block to catch and handle potential failures - Provides better visibility into what went wrong if the deployment fails - Re-raises the exception to ensure the parent flow fails appropriately 4. Return Value Handling: - Captures the

flow_run

object returned by

run_deployment

- Allows monitoring or further interaction with the spawned flow run - Useful for checking status or handling downstream dependencies 5. Type Hints: - Added type hint for the

log_prints

parameter - Makes the code more maintainable and helps with IDE support 6. Optional Parameters: - Added

timeout

parameter to prevent indefinite waiting - Set

return_type="flow_run"

to explicitly get the flow run object You might also consider these additional improvements depending on your use case:

python
from prefect import flow
from prefect import run_deployment
from typing import Optional, Dict, Any

@flow
async def parent_flow(
    deployment_name: str,
    parameters: Dict[str, Any],
    log_prints: bool = True,
    timeout: Optional[int] = 3600
) -> Any:
    """
    Run a child deployment with parameters.
    
    Args:
        deployment_name: Name of the deployment to run
        parameters: Parameters to pass to the deployment
        log_prints: Whether to log prints
        timeout: Maximum time to wait for deployment in seconds
        
    Returns:
        Flow run object from the deployment
    """
    try:
        flow_run = await run_deployment(
            name=deployment_name,
            parameters=parameters,
            timeout=timeout,
            return_type="flow_run"
        )
        
        # Optional: Wait for the flow run to complete
        # await flow_run.wait_for_completion()
        
        return flow_run
    except Exception as e:
        print(f"Error running deployment {deployment_name}: {e}")
        raise

This version adds: - Parameter flexibility by accepting deployment name and parameters as arguments - Documentation with docstring - Option to wait for flow completion if needed - More comprehensive type hints - Better reusability across different deployments Remember that when using this improved version, you'll need to call it like this:

python
parameters = {
    "script_type": "INGEST",
    "script_name": "PREFECT_TEST_1",
    "minutes": 5
}

await parent_flow(
    deployment_name="test-prefect-flow/ingest-prefect-test-deployment",
    parameters=parameters
)

These improvements make the code more robust, maintainable, and reusable while following Prefect's best practices for handling asynchronous operations and error cases.

Here's the corrected code:

python
from prefect import flow
from prefect.deployments import run_deployment  # Correct import path
from typing import Optional, Dict, Any

@flow
async def parent_flow(
    deployment_name: str,
    parameters: Dict[str, Any],
    log_prints: bool = True,
    timeout: Optional[int] = 3600
) -> Any:
    """
    Run a child deployment with parameters.
    
    Args:
        deployment_name: Name of the deployment to run
        parameters: Parameters to pass to the deployment
        log_prints: Whether to log prints
        timeout: Maximum time to wait for deployment in seconds
        
    Returns:
        Flow run object from the deployment
    """
    try:
        flow_run = await run_deployment(
            name=deployment_name,
            parameters=parameters,
            timeout=timeout,
            return_type="flow_run"
        )
        
        # Optional: Wait for the flow run to complete
        # await flow_run.wait_for_completion()
        
        return flow_run
    except Exception as e:
        print(f"Error running deployment {deployment_name}: {e}")
        raise

To use this flow:

python
parameters = {
    "script_type": "INGEST",
    "script_name": "PREFECT_TEST_1",
    "minutes": 5
}

# If running in an async context
await parent_flow(
    deployment_name="test-prefect-flow/ingest-prefect-test-deployment",
    parameters=parameters
)

# If running in a synchronous context, you'll need to use asyncio:
import asyncio

asyncio.run(parent_flow(
    deployment_name="test-prefect-flow/ingest-prefect-test-deployment",
    parameters=parameters
))

The key change is the import statement: - ❌

from prefect import run_deployment

# Incorrect - ✅

from prefect.deployments import run_deployment

# Correct This is the proper import path in Prefect 3.x. All other improvements and functionality remain the same as discussed in the previous response.

Your code is almost there, but there's one critical issue that needs to be addressed. When running async flows in a synchronous context (like your

__main__

block), you need to properly handle the async execution. Here's the improved version:

python
from prefect import flow
from prefect.deployments import run_deployment
from typing import Optional
import asyncio  # Add this import

@flow
async def parent_flow(log_prints: bool = True):
    try:
        print("Prefect iteration 1")
        flow_run = await run_deployment(
            name="test-prefect-flow/ingest-prefect-test-deployment",
            parameters={
                "script_type": "INGEST",
                "script_name": "PREFECT_TEST_1",
                "minutes": 1
            },
            timeout=3600  # Optional timeout in seconds
        )
        return flow_run
    except Exception as e:
        print(f"Error running deployment: {e}")
        raise

if __name__ == "__main__":
    asyncio.run(parent_flow())  # Wrap the flow call in asyncio.run()

Key improvements: 1. Added

asyncio.run()

to properly execute the async flow in the

__main__

block 2. Without this, the flow would create a coroutine object but never actually execute Optional improvements you might consider:

python
from prefect import flow
from prefect.deployments import run_deployment
from typing import Optional
import asyncio

@flow
async def parent_flow(log_prints: bool = True):
    try:
        print("Prefect iteration 1")
        flow_run = await run_deployment(
            name="test-prefect-flow/ingest-prefect-test-deployment",
            parameters={
                "script_type": "INGEST",
                "script_name": "PREFECT_TEST_1",
                "minutes": 1
            },
            timeout=3600,
            return_type="flow_run"  # Explicitly request flow run object
        )
        
        # Optionally wait for the flow run to complete
        # await flow_run.wait_for_completion()
        
        # You could add status checking
        print(f"Flow run status: {flow_run.state.name}")
        
        return flow_run
    except Exception as e:
        print(f"Error running deployment: {e}")
        raise

if __name__ == "__main__":
    try:
        asyncio.run(parent_flow())
    except KeyboardInterrupt:
        print("\nFlow execution interrupted by user")
    except Exception as e:
        print(f"Flow execution failed: {e}")

These additional improvements add: 1. Better error handling in the

__main__

block 2. Explicit

return_type="flow_run"

for clearer intent 3. Option to wait for flow completion and check status 4. Graceful handling of keyboard interrupts The choice between the simpler version and the more detailed version depends on your needs: - Use the simpler version if you just need to trigger the deployment - Use the detailed version if you need to monitor the flow run's progress or handle specific error cases

Ah, you're right. Let me correct that. The

return_type

parameter is not available in the current version of

run_deployment

. Here's the corrected code:

python
from prefect import flow
from prefect.deployments import run_deployment
from typing import Optional
import asyncio

@flow
async def parent_flow(log_prints: bool = True):
    try:
        print("Prefect iteration 1")
        flow_run = await run_deployment(
            name="test-prefect-flow/ingest-prefect-test-deployment",
            parameters={
                "script_type": "INGEST",
                "script_name": "PREFECT_TEST_1",
                "minutes": 1
            },
            timeout=3600
        )
        
        # You can still check the flow run status
        print(f"Flow run created with ID: {flow_run.id}")
        
        return flow_run
    except Exception as e:
        print(f"Error running deployment: {e}")
        raise

if __name__ == "__main__":
    try:
        asyncio.run(parent_flow())
    except KeyboardInterrupt:
        print("\nFlow execution interrupted by user")
    except Exception as e:
        print(f"Flow execution failed: {e}")

The main changes: 1. Removed the

return_type

parameter which was causing the error 2. The

run_deployment

function already returns a flow run object by default This version should work correctly now. The flow run object returned will contain information about the deployment execution that you can use for monitoring or further processing if needed. If you want to wait for the flow run to complete, you can still do:

python
@flow
async def parent_flow(log_prints: bool = True):
    try:
        print("Prefect iteration 1")
        flow_run = await run_deployment(
            name="test-prefect-flow/ingest-prefect-test-deployment",
            parameters={
                "script_type": "INGEST",
                "script_name": "PREFECT_TEST_1",
                "minutes": 1
            },
            timeout=3600
        )
        
        # Wait for completion if needed
        await flow_run.wait_completed()
        print(f"Flow run {flow_run.id} completed with state: {flow_run.state_name}")
        
        return flow_run
    except Exception as e:
        print(f"Error running deployment: {e}")
        raise

This version should work correctly with the current Prefect 3.x API.

No, you don't have to use async/await! Here's the simpler synchronous version of your code:

python
from prefect import flow
from prefect.deployments import run_deployment

@flow
def parent_flow(log_prints: bool = True):
    try:
        print("Prefect iteration 1")
        flow_run = run_deployment(
            name="test-prefect-flow/ingest-prefect-test-deployment",
            parameters={
                "script_type": "INGEST",
                "script_name": "PREFECT_TEST_1",
                "minutes": 1
            },
            timeout=3600
        )
        return flow_run
    except Exception as e:
        print(f"Error running deployment: {e}")
        raise

if __name__ == "__main__":
    parent_flow()

Key changes made: 1. Removed

async

keyword from the flow definition 2. Removed

await

from

run_deployment

call 3. Removed

asyncio

import and

asyncio.run()

4. Everything else remains the same This synchronous version is simpler and will work just fine. Use this version if you don't need async functionality in your workflow. The async version is more appropriate when you need to: - Run multiple deployments concurrently - Integrate with other async code - Handle complex concurrent operations Both approaches are valid in Prefect 3.x - choose the one that best fits your use case!

Here are two approaches to run deployments sequentially: ### Synchronous Approach (Simpler):

python
from prefect import flow
from prefect.deployments import run_deployment

@flow
def parent_flow(log_prints: bool = True):
    try:
        print("Starting first deployment")
        flow_run_1 = run_deployment(
            name="test-prefect-flow/ingest-prefect-test-deployment",
            parameters={
                "script_type": "INGEST",
                "script_name": "PREFECT_TEST_1",
                "minutes": 1
            },
            timeout=3600
        )
        # Wait for the first deployment to complete
        flow_run_1.wait_completed()
        print(f"First deployment completed with state: {flow_run_1.state_name}")

        print("Starting second deployment")
        flow_run_2 = run_deployment(
            name="test-prefect-flow/another-deployment",
            parameters={
                "script_type": "PROCESS",
                "script_name": "PREFECT_TEST_2",
                "minutes": 2
            },
            timeout=3600
        )
        flow_run_2.wait_completed()
        print(f"Second deployment completed with state: {flow_run_2.state_name}")

        # Add more deployments as needed...

    except Exception as e:
        print(f"Error running deployments: {e}")
        raise

if __name__ == "__main__":
    parent_flow()

### Asynchronous Approach (More Flexible):

python
from prefect import flow
from prefect.deployments import run_deployment
import asyncio

@flow
async def parent_flow(log_prints: bool = True):
    try:
        print("Starting first deployment")
        flow_run_1 = await run_deployment(
            name="test-prefect-flow/ingest-prefect-test-deployment",
            parameters={
                "script_type": "INGEST",
                "script_name": "PREFECT_TEST_1",
                "minutes": 1
            },
            timeout=3600
        )
        # Wait for the first deployment to complete
        await flow_run_1.wait_completed()
        print(f"First deployment completed with state: {flow_run_1.state_name}")

        print("Starting second deployment")
        flow_run_2 = await run_deployment(
            name="test-prefect-flow/another-deployment",
            parameters={
                "script_type": "PROCESS",
                "script_name": "PREFECT_TEST_2",
                "minutes": 2
            },
            timeout=3600
        )
        await flow_run_2.wait_completed()
        print(f"Second deployment completed with state: {flow_run_2.state_name}")

        # Add more deployments as needed...

    except Exception as e:
        print(f"Error running deployments: {e}")
        raise

if __name__ == "__main__":
    asyncio.run(parent_flow())

### For More Complex Scenarios: If you have many deployments to run sequentially, you might want to structure it like this: ```python from prefect import flow from prefect.deployments import run_deployment from typing import List, Dict, Any @flow def parent_flow(log_prints: bool = True): # Define your deployment configurations deployments = [ { "name": "test-prefect-flow/ingest-prefect-test-deployment", "parameters": { "script_type": "INGEST", "script_name": "PREFECT_TEST_1", "minutes": 1 } }, { "name": "test-prefect-flow/another-deployment", "parameters": { "script_type": "PROCESS", "script_name": "PREFECT_TEST_2", "minutes": 2 } } # Add more deployments as needed ] try: for i, deployment in enumerate(deployments, 1): print(f"Starting deployment {i}:

Here's the complete code for handling multiple sequential deployments in a more structured way:

python
from prefect import flow
from prefect.deployments import run_deployment
from typing import List, Dict, Any

@flow
def parent_flow(log_prints: bool = True):
    # Define your deployment configurations
    deployments = [
        {
            "name": "test-prefect-flow/ingest-prefect-test-deployment",
            "parameters": {
                "script_type": "INGEST",
                "script_name": "PREFECT_TEST_1",
                "minutes": 1
            }
        },
        {
            "name": "test-prefect-flow/another-deployment",
            "parameters": {
                "script_type": "PROCESS",
                "script_name": "PREFECT_TEST_2",
                "minutes": 2
            }
        },
        # Add more deployments as needed
    ]

    try:
        for i, deployment in enumerate(deployments, 1):
            print(f"Starting deployment {i}: {deployment['name']}")
            
            flow_run = run_deployment(
                name=deployment["name"],
                parameters=deployment["parameters"],
                timeout=3600
            )
            
            # Wait for deployment to complete
            flow_run.wait_completed()
            print(f"Deployment {i} completed with state: {flow_run.state_name}")

    except Exception as e:
        print(f"Error running deployments: {e}")
        raise

if __name__ == "__main__":
    parent_flow()

And here's a more advanced version with additional features: ```python from prefect import flow from prefect.deployments import run_deployment from typing import List, Dict, Any from datetime import timedelta class DeploymentConfig: def __init__(self, name: str, parameters: Dict[str, Any], timeout: int = 3600): self.name = name self.parameters = parameters self.timeout = timeout @flow def parent_flow(log_prints: bool = True, fail_fast: bool = True): # Define your deployment configurations deployments = [ DeploymentConfig( name="test-prefect-flow/ingest-prefect-test-deployment", parameters={ "script_type": "INGEST", "script_name": "PREFECT_TEST_1", "minutes": 1 } ), DeploymentConfig( name="test-prefect-flow/another-deployment", parameters={ "script_type": "PROCESS", "script_name": "PREFECT_TEST_2", "minutes": 2 }, timeout=7200 # Custom timeout for this deployment ) ] results = [] try: for i, deployment in enumerate(deployments, 1): print(f"\n{'='*50}") print(f"Starting deployment {i}: {deployment.name}") print(f"Parameters: {deployment.parameters}") try: flow_run = run_deployment( name=deployment.name, parameters=deployment.parameters, timeout=deployment.timeout ) # Wait for deployment to complete flow_run.wait_completed() state = flow_run.state_name print(f"Deployment {i} completed with state: {state}") results.append({ "deployment": deployment.name, "flow_run_id": flow_run.id, "state": state, "success": state == "COMPLETED" }) if state != "COMPLETED" and fail_fast: raise Exception(f"Deployment {deployment.name} failed with state {state}") except Exception as e: print(f"Error in deployment {deployment.name}: {e}") results.append({ "deployment": deployment.name, "error": str(e),

"success": False }) if fail_fast: raise except Exception as e: print(f"\nParent

Ah, you're right. Let me correct that. The method name is

wait_for_completion()

, not

wait_completed

. Here's the corrected code:

python
from prefect import flow
from prefect.deployments import run_deployment
from typing import List, Dict, Any

@flow
def parent_flow(log_prints: bool = True):
    # Define your deployment configurations
    deployments = [
        {
            "name": "test-prefect-flow/ingest-prefect-test-deployment",
            "parameters": {
                "script_type": "INGEST",
                "script_name": "PREFECT_TEST_1",
                "minutes": 1
            }
        },
        {
            "name": "test-prefect-flow/another-deployment",
            "parameters": {
                "script_type": "PROCESS",
                "script_name": "PREFECT_TEST_2",
                "minutes": 2
            }
        }
    ]

    try:
        for i, deployment in enumerate(deployments, 1):
            print(f"Starting deployment {i}: {deployment['name']}")
            
            flow_run = run_deployment(
                name=deployment["name"],
                parameters=deployment["parameters"],
                timeout=3600
            )
            
            # Correct method name
            flow_run.wait_for_completion()
            print(f"Deployment {i} completed with state: {flow_run.state_name}")

    except Exception as e:
        print(f"Error running deployments: {e}")
        raise

if __name__ == "__main__":
    parent_flow()

And here's the more advanced version with the correct method name: ```python from prefect import flow from prefect.deployments import run_deployment from typing import List, Dict, Any from datetime import timedelta class DeploymentConfig: def __init__(self, name: str, parameters: Dict[str, Any], timeout: int = 3600): self.name = name self.parameters = parameters self.timeout = timeout @flow def parent_flow(log_prints: bool = True, fail_fast: bool = True): # Define your deployment configurations deployments = [ DeploymentConfig( name="test-prefect-flow/ingest-prefect-test-deployment", parameters={ "script_type": "INGEST", "script_name": "PREFECT_TEST_1", "minutes": 1 } ), DeploymentConfig( name="test-prefect-flow/another-deployment", parameters={ "script_type": "PROCESS", "script_name": "PREFECT_TEST_2", "minutes": 2 }, timeout=7200 # Custom timeout for this deployment ) ] results = [] try: for i, deployment in enumerate(deployments, 1): print(f"\n{'='*50}") print(f"Starting deployment {i}: {deployment.name}") print(f"Parameters: {deployment.parameters}") try: flow_run = run_deployment( name=deployment.name, parameters=deployment.parameters, timeout=deployment.timeout ) # Correct method name flow_run.wait_for_completion() state = flow_run.state_name print(f"Deployment {i} completed with state: {state}") results.append({ "deployment": deployment.name, "flow_run_id": flow_run.id, "state": state, "success": state == "COMPLETED" }) if state != "COMPLETED" and fail_fast: raise Exception(f"Deployment {deployment.name} failed with state {state}") except Exception as e: print(f"Error in deployment {deployment.name}: {e}") results.append({ "deployment": deployment.name, "error": str(e),

"success": False }) if fail_fast: raise

I apologize for the confusion in my previous responses. Let me provide the correct way to wait for flow run completion in Prefect 3.x:

python
from prefect import flow
from prefect.deployments import run_deployment
from typing import List, Dict, Any
import time

@flow
def parent_flow(log_prints: bool = True):
    # Define your deployment configurations
    deployments = [
        {
            "name": "test-prefect-flow/ingest-prefect-test-deployment",
            "parameters": {
                "script_type": "INGEST",
                "script_name": "PREFECT_TEST_1",
                "minutes": 1
            }
        },
        {
            "name": "test-prefect-flow/another-deployment",
            "parameters": {
                "script_type": "PROCESS",
                "script_name": "PREFECT_TEST_2",
                "minutes": 2
            }
        }
    ]

    try:
        for i, deployment in enumerate(deployments, 1):
            print(f"Starting deployment {i}: {deployment['name']}")
            
            flow_run = run_deployment(
                name=deployment["name"],
                parameters=deployment["parameters"],
                timeout=3600
            )
            
            # Wait for the flow run to complete by checking its state
            while flow_run.state_type not in ["COMPLETED", "FAILED", "CANCELLED"]:
                time.sleep(1)  # Wait for 1 second before checking again
                flow_run.refresh()  # Refresh the flow run state
            
            print(f"Deployment {i} completed with state: {flow_run.state_name}")

            # Optional: raise an error if the flow run failed
            if flow_run.state_type == "FAILED":
                raise Exception(f"Deployment {deployment['name']} failed with state {flow_run.state_name}")

    except Exception as e:
        print(f"Error running deployments: {e}")
        raise

if __name__ == "__main__":
    parent_flow()

Key changes: 1. Removed the non-existent

wait_for_completion()

method 2. Added a polling loop that checks the flow run's state 3. Used

flow_run.refresh()

to get the latest state 4. Added proper state checking using

state_type

For a more robust version with additional features: ```python from prefect import flow from prefect.deployments import run_deployment from typing import List, Dict, Any import time class DeploymentConfig: def __init__(self, name: str, parameters: Dict[str, Any], timeout: int = 3600): self.name = name self.parameters = parameters self.timeout = timeout def wait_for_flow_run(flow_run, timeout_seconds: int = 3600): """Wait for a flow run to complete with timeout.""" start_time = time.time() while time.time() - start_time < timeout_seconds: if flow_run.state_type in ["COMPLETED", "FAILED", "CANCELLED"]: return True time.sleep(1) flow_run.refresh() return False @flow def parent_flow(log_prints: bool = True, fail_fast: bool = True): deployments = [ DeploymentConfig( name="test-prefect-flow/ingest-prefect-test-deployment", parameters={ "script_type": "INGEST", "script_name": "PREFECT_TEST_1", "minutes": 1 } ), DeploymentConfig( name="test-prefect-flow/another-deployment", parameters={ "script_type": "PROCESS", "script_name": "PREFECT_TEST_2", "minutes": 2 }, timeout=7200 ) ] results = [] try: for i, deployment in enumerate(deployments, 1): print(f"\n{'='*50}") print(f"Starting deployment {i}: {deployment.name}") print(f"Parameters:

You're absolutely right! I apologize for the confusion in my previous responses. In Prefect 3.x,

run_deployment

by default runs the deployment as a subflow (with

as_subflow=True

), which means it will wait for completion by default. Here's the correct, simpler code you need:

python
from prefect import flow
from prefect.deployments import run_deployment

@flow
def parent_flow(log_prints: bool = True):
    # Define your deployment configurations
    deployments = [
        {
            "name": "test-prefect-flow/ingest-prefect-test-deployment",
            "parameters": {
                "script_type": "INGEST",
                "script_name": "PREFECT_TEST_1",
                "minutes": 1
            }
        },
        {
            "name": "test-prefect-flow/another-deployment",
            "parameters": {
                "script_type": "PROCESS",
                "script_name": "PREFECT_TEST_2",
                "minutes": 2
            }
        }
    ]

    try:
        for i, deployment in enumerate(deployments, 1):
            print(f"Starting deployment {i}: {deployment['name']}")
            
            # This will wait for completion by default
            flow_run = run_deployment(
                name=deployment["name"],
                parameters=deployment["parameters"],
                timeout=3600
            )
            
            print(f"Deployment {i} completed with state: {flow_run.state_name}")

    except Exception as e:
        print(f"Error running deployments: {e}")
        raise

if __name__ == "__main__":
    parent_flow()

If you want to run the deployment without waiting for it to complete, you can set `as_subflow=False`:

python
flow_run = run_deployment(
    name=deployment["name"],
    parameters=deployment["parameters"],
    timeout=3600,
    as_subflow=False  # This will not wait for completion
)

The default behavior (with

as_subflow=True

) is typically what you want when running deployments sequentially, as it ensures each deployment completes before moving on to the next one. All the waiting logic I included in my previous responses is unnecessary - I apologize for the confusion!