@Marvin How can I get one flow to wait for another flow.

To make one flow wait for another flow to complete in Prefect, you can utilize subflows, where a parent flow orchestrates the execution of child flows (known as subflows). ### Using

run_deployment

for Subflows You can utilize the function

run_deployment

to run a deployment within a parent flow, which essentially waits for the child flow to complete. Here's a basic example to illustrate this concept:

python
from prefect import flow
from prefect.deployments import run_deployment

@flow
def parent_flow():
    # Run the child flow deployment
    deployment_id = "child-flow-deployment"
    print(f"Running child flow deployment: {deployment_id}")
    flow_run = run_deployment(name=deployment_id)
    # Flow run will wait here until the child flow completes
    result = flow_run.result()
    print(f"Child flow completed with result: {result}")
    
    # Continue with the rest of the parent flow
    print("Parent flow is now complete.")

if __name__ == "__main__":
    parent_flow()

### Cancellation Considerations In some cases, you might want to ensure that if the parent flow fails or is canceled, all subflows are also canceled. Future improvements might offer more streamlined solutions for such scenarios, as referenced in one of the GitHub issues. ### Documentation and Further Reading Unfortunately, direct links to documentation could not be fetched, but you can refer to the Prefect documentation for more details on flow orchestration and deployment execution: - Prefect Documentation This should help you set up dependencies between flows effectively. If you need further help or more specific examples, feel free to ask!

@Marvin Will this dependency be reflected in the ui

No, sub-flows executed within the same flow are not currently reflected as dependencies in the Prefect UI. This is a known issue among users, and solutions to ensure that sub-flows are accurately represented and managed within the UI are an area of active development. Would you like help exploring workarounds some people are using?

@Marvin If I'm running multiple run_deployment statements in parallel then want to use the results once available for another run_deployment statement, how would I do so

To run multiple

run_deployment

statements in parallel and then use their results for another

run_deployment

, you can use Prefect tasks and flows to handle this asynchronously. Here’s an example to illustrate this concept:

python
from prefect import flow, task
from prefect.deployments import run_deployment

# Fetch results in parallel
@task
def run_parallel_deployments(deployment_name, params):
    return run_deployment(name=deployment_name, parameters=params)

# Main flow
@flow
def parent_flow():
    # Run multiple deployments in parallel
    deployments = [
        {"name": "deployment-1", "params": {"param1": "value1"}},
        {"name": "deployment-2", "params": {"param1": "value2"}},
        {"name": "deployment-3", "params": {"param1": "value3"}},
    ]

    # Submit run_deployment tasks
    futures = [run_parallel_deployments.submit(deployment["name"], deployment["params"]) for deployment in deployments]

    # Wait for all deployments to complete and gather results
    results = [future.result() for future in futures]
    
    # Use results for another deployment
    print(f"All deployments completed with results: {results}")
    another_deployment_name = "another-deployment"
    another_params = {"previous_results": results}
    
    final_run = run_deployment(name=another_deployment_name, parameters=another_params)
    
    print(f"Final deployment completed with result: {final_run.result()}")

if __name__ == "__main__":
    parent_flow()

### Explanation: 1. Parallel Deployment Execution: Use

task

to wrap the

run_deployment

calls and then use

submit

to run them asynchronously. 2. Waiting for All Results: Collect the results using the

.result()

method on each future. 3. Use Results for Next Deployment: Once all previous deployments are complete, use their results for the next deployment. ### Prefect Code Example Reference: For a deeper understanding, you can view a similar example that handles task submission and managing task futures: Prefect Task Submission Example This way, you can orchestrate multiple deployments in parallel and use their results for subsequent tasks dynamically. If you need any further assistance or specific adjustments, feel free to ask!

@Marvin This solutions waits for all of the results to be aggregated, how would I go about using the result of each run, immediately after being available

To use the result of each

run_deployment

immediately after it becomes available, you can use asynchronous tasks and the

await

mechanism. This allows you to process each result as it completes. Here’s an updated example utilizing asynchronous functions:

python
import asyncio
from prefect import flow, task, get_run_logger
from prefect.deployments import run_deployment

@task
async def run_parallel_deployment(deployment_name, params):
    return await run_deployment(name=deployment_name, parameters=params)

@task
def process_result(result):
    logger = get_run_logger()
    # Process your result here
    <http://logger.info|logger.info>(f"Processing result: {result}")

@flow
async def parent_flow():
    # List of deployment details to run
    deployments = [
        {"name": "deployment-1", "params": {"param1": "value1"}},
        {"name": "deployment-2", "params": {"param1": "value2"}},
        {"name": "deployment-3", "params": {"param1": "value3"}},
    ]
    
    # Run deployments in parallel and process each result as it completes
    deployment_tasks = [run_parallel_deployment.submit(deployment["name"], deployment["params"]) for deployment in deployments]
    
    for deployment_task in asyncio.as_completed(deployment_tasks):
        result = await deployment_task
        process_result.submit(result)
    
    # Optionally: You can also chain another deployment using results as they complete
    final_results = await asyncio.gather(*deployment_tasks)
    another_deployment_name = "another-deployment"
    another_params = {"previous_results": final_results}
    final_run = await run_deployment(name=another_deployment_name, parameters=another_params)
    
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Final deployment completed with result: {final_run}")

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

### Explanation: 1. Async Task: Each

run_parallel_deployment

task is awaited individually. 2. Handling Results as They Complete: Using

asyncio.as_completed

, we can handle each completed task's result immediately. 3. Processing Results: The

process_result

task processes each result as it becomes available. 4. Final Deployment: Optionally, you can chain another deployment once all initial deployments are complete. ### Note: Async functionalities are properly handled in modern

Prefect

versions. Ensure your environment and version are set up to support asyncio. Feel free to ask if you have further questions or need specific adjustments!