Yes! There's a much simpler way to get results from

run_deployment

without setting up storage. You can use the

as_subflow=True

parameter when calling

run_deployment

. This will make the deployment run as part of your current flow's execution context, allowing you to directly access the return value. Here's an example:

python
from prefect import flow
from prefect.deployments import run_deployment

@flow
def parent_flow():
    # Run the deployment as a subflow and get the result directly
    result = run_deployment(
        name="your-deployment/your-deployment-name",
        as_subflow=True  # This is the key!
    )
    print(f"Got result: {result}")  # The actual return value from your subflow

This approach has several benefits: 1. No storage configuration needed 2. Direct access to return values 3. Maintains flow run hierarchy (the subflow will appear nested under your parent flow in the UI) 4. Preserves proper error propagation The only caveat is that your parent flow needs to be running for this to work - you can't use

as_subflow=True

from a Python REPL or script that isn't part of a flow. In those cases, you would need either storage or to use the async API to wait for the result. This is actually the recommended approach in Prefect 3.x for handling subflow results when you don't need persistent storage of the results.

After researching this thoroughly, I need to correct my previous response with an important caveat: when using

as_subflow=True

, the subflow will actually run in the same context as the parent flow, NOT in the work pool specified in the deployment. This means: 1. If your parent flow is running locally, the subflow will also run locally 2. If your parent flow is running in a work pool (e.g., on a worker), the subflow will run in that same worker This behavior makes

as_subflow=True

unsuitable if you specifically need the subflow to run in a different work pool or infrastructure context than the parent flow. For cases where you need the subflow to run in its specified work pool, you should: 1. Use

as_subflow=False

(the default) 2. Either: - Set up result storage to get the results - Use the async API to wait for the result:

python
from prefect import flow
from prefect.deployments import run_deployment

@flow
async def parent_flow():
    # This will run the deployment in its configured work pool
    deployment_result = await run_deployment(
        name="your-deployment/your-deployment-name",
        timeout=300  # Optional: specify how long to wait for the result
    )
    return deployment_result.state.result()

So to directly answer your question: No, using

as_subflow=True

will not run the subflow in its configured work pool - it will run in the same context as the parent flow. If you need the subflow to run in its specific work pool, you'll need to either set up result storage or use the async API approach shown above.

After researching this, I need to clarify an important point: when using

as_subflow=True

, the subflow will actually execute in the same worker process as the parent flow. This is because

as_subflow=True

makes the subflow run as part of the parent flow's execution context. If you want subflows to run on different workers within the same pool, you have two options: 1. Use

as_subflow=False

(default behavior):

python
from prefect import flow
from prefect.deployments import run_deployment

@flow
async def parent_flow():
    # These will be distributed across available workers in the pool
    results = []
    for i in range(5):
        deployment_result = await run_deployment(
            name="your-deployment/your-deployment-name",
            timeout=300
        )
        results.append(deployment_result.state.result())
    return results

2. Use parallel task runners with subflows:

python
from prefect import flow
from prefect.tasks import task_input_kwargs
from prefect.runners import RayTaskRunner

@flow(task_runner=RayTaskRunner())
def parent_flow():
    # These will run in parallel in separate processes
    results = []
    for i in range(5):
        results.append(subflow.submit())
    return results

@flow
def subflow():
    return "some result"

The key differences are: - `as_subflow=True`: Runs in the same worker process as the parent - `as_subflow=False`: Can be distributed across different workers in the pool - Using a parallel task runner: Allows parallel execution in different processes, but still within the same worker If you need the results without setting up storage and want the flows to run on different workers, I'd recommend using the async approach with

as_subflow=False

and waiting for the results. This gives you both distribution across workers and access to results. Remember that if you're running multiple flows concurrently, you might want to consider setting appropriate concurrency limits on your deployment to manage resource usage.

You can start multiple workers for your work pool using:

bash
prefect worker start -p "your-work-pool-name"

Run this command multiple times (in different terminals or processes) to start multiple workers that will share the workload of your parallel subflows.

Based on the research, I can provide a nuanced answer about using blocks for state sharing: For simple string data, there are several lightweight alternatives that are more appropriate than setting up S3: 1. SQLite Block (Simple and Local):

python
from prefect.blocks.system import JSON

@flow
async def producer_flow():
    # Save data
    json_block = await JSON.load("my-json-block")
    await json_block.write_value({"my_data": "some result"})

@flow
async def consumer_flow():
    # Read data
    json_block = await JSON.load("my-json-block")
    data = json_block.value

2. Redis Block (If you have Redis available):

python
from prefect_redis import RedisCache

@flow
async def producer_flow():
    redis_block = RedisCache.load("my-redis-block")
    redis_block.set("my_key", "my_value")

@flow
async def consumer_flow():
    redis_block = RedisCache.load("my-redis-block")
    value = redis_block.get("my_key")

However, there are some important considerations: Pros of using blocks for state sharing: - Simple to set up and use - No need for complex infrastructure - Works well for simple data types - Built into Prefect's security model Cons and potential issues: - Not designed primarily for state sharing - Could lead to race conditions if multiple flows try to update the same data - May not scale well with large amounts of data or high concurrency - Could create hidden dependencies between flows Better alternatives depending on your specific needs: 1. If you need proper persistence:

python
from prefect.blocks.system import Secret

# Use the Secret block (it's designed for small strings)
@flow
async def producer_flow():
    secret_block = await Secret.load("my-secret")
    await secret_block.write_value("my_result")

2. If you need high performance:

python
# Use a proper message queue or database
from prefect_redis.tasks import redis_get, redis_set

@flow
def producer_flow():
    redis_set("my_key", "my_value", redis_block_name="my-redis")

@flow
def consumer_flow():
    value = redis_get("my_key", redis_block_name="my-redis")

3. If you need temporary sharing during a parent-child flow relationship:

python
from prefect import flow
from prefect.deployments import run_deployment

@flow
async def parent_flow():
    # Use flow parameters to pass data
    result = await run_deployment(
        "child-flow/deployment",
        parameters={"shared_data": "my_value"},
        timeout=300
    )

The recommended approach depends on your specific needs: - For truly temporary data between related flows: Use flow parameters - For small configuration values: Use Secret blocks - For larger datasets or high concurrency: Use a proper database or message queue - For simple prototypes or low-stakes data: JSON or SQLite blocks can work, but be aware of the limitations If you're just dealing with simple strings and don't want to set up S3, using a JSON or Secret block is a reasonable compromise, but make sure to document this pattern clearly in your codebase and be aware of its limitations.