@Marvin within a flow, I want to limit how many subflows run at once. How can I do this without using the global concurrency controls?

To limit the concurrency of subflows within a flow without using global concurrency controls, you can utilize task concurrency limits through tags. Here are the steps you might follow: 1. Assign tags to the tasks that you want to limit and configure concurrency limits for those tags. 2. Set concurrency limits via the Prefect CLI, Prefect API using

PrefectClient

, or the Prefect server UI/Cloud. 3. When the tasks associated with the tags attempt to run, if no concurrency slots are available, the transition to a

Running

state will be delayed. Below is a high-level example of how you can set the concurrency limit by using the Prefect CLI:

shell
# Create a concurrency limit named "subflow-limit" with a limit of 3 running tasks at the same time
prefect concurrency-limit create subflow-limit 3

# To apply this limit, you would then tag your tasks with "subflow-limit"

And within your flow code, here is an example of how to use tags with your tasks:

python
from prefect import flow, task

@task(tags=["subflow-limit"])
def subflow_task():
    # Task implementation

@flow
def main_flow():
    # Flow implementation

    # When submitting tasks
    subflow_task.submit()

# Run the main flow
if __name__ == "__main__":
    main_flow()

You can find more information in the Prefect documentation for executing behavior with and configuring concurrency limits, and also using rate limits within your flows.

@Marvin Is there a way to set the controls from within the flow? I don’t like having logic outside of the flow in a CLI command. Could I use the Prefect API from within the parent flow to do an idempotent create of the limit for that specific flow?

Yes, it is possible to set concurrency limits from within a flow by using the Prefect API. You can use the Prefect API with the

PrefectClient

Python client to create concurrency limits programmatically in an idempotent manner. Here is an example of how you might use the

PrefectClient

within your flow to set a concurrency limit:

python
from prefect import flow, task
from prefect.client import get_client

@task
def subflow_task():
    # Task implementation

@flow
def main_flow():
    # Set up the concurrency limit idempotently
    with get_client() as client:
        client.create_concurrency_limit(name="subflow-limit", concurrency_limit=3)

    # Flow implementation
    subflow_task.submit()

# Run the main flow
if __name__ == "__main__":
    main_flow()

Additionally, the Prefect library provides a

concurrency

context manager, which you can use to manage concurrency within a task definition. When using the

concurrency

context manager, it will create the concurrency limit in an inactive state that does not already exist. Example with

concurrency

context manager:

python
from prefect import flow, task
from prefect.concurrency.sync import concurrency

@task
def process_data(x, y):
    with concurrency("database", occupy=1):
        return x + y

@flow
def my_flow():
    for x, y in [(1, 2), (2, 3), (3, 4), (4, 5)]:
        process_data.submit(x, y)

if __name__ == "__main__":
    my_flow()

Here's a link to the documentation section explaining this context manager in more detail. Remember that for asynchronous implementations, you should use the

prefect.concurrency.asyncio

module. For more complex use cases or dynamic concurrency limit creation, it is recommended to interact with the Prefect API, as shown in the first example.

@Marvin I looked at using the concurrency, but I didn’t see a way to specify how much concurrency is allowed for the specific limit, just how much to occupy. I did see you could set that value in the UI, but i don’t want to use the UI.

Indeed, the concurrency context manager in a task allows you to specify how much of the concurrency limit to occupy with that task run, but it does not allow you to set the total concurrency allowed for that limit directly within your flow. This means that using the

concurrency

context manager alone doesn't provide a way to define the maximum concurrency limit value. Unfortunately, the Prefect documentation excerpts above do not demonstrate creating concurrency limits with specific values directly through the

PrefectClient

. The usual approach to creating or updating concurrency limits is through the Prefect CLI or the UI/Cloud interface. However, since you would like to handle this entirely within your flow code and avoid using the UI, you might consider crafting a custom solution using the Prefect API. The Prefect API might have an endpoint allowing the idempotent creation or updating of concurrency limits, but such functionality isn’t explicitly documented or exemplified. To create or update concurrency limits through a Prefect flow, you would typically interact with the Prefect API using the

PrefectClient

methods—assuming such methods exist in the client library. This interaction would be similar to how you would use other Prefect settings or resources programmatically. Given that Prefect's API and client library are subject to change, and the documentation doesn't provide a direct example, the best course of action might be to inquire directly with Prefect's support or development community, or to explore the

PrefectClient

source code or API documentation to see if the capability exists or if it should be requested as a feature. Here is an example pattern of how this might look within your flow if such an API method existed. Please note that this is speculative and might not reflect the actual PrefectClient API:

python
from prefect import flow, task
from prefect.client import get_client

@task
def subflow_task():
    # Task implementation

@flow
def main_flow():
    with get_client() as client:
        # Hypothetical API method to create or update the concurrency limit
        client.create_or_update_concurrency_limit("subflow-limit", limit=3)

    # Flow implementation
    subflow_task.submit()

# Run the main flow
if __name__ == "__main__":
    main_flow()

If this functionality does not currently exist in Prefect, you might need to continue using the Prefect CLI or the Prefect UI/Cloud until such a feature is implemented in Prefect Client or exposed via the API.

@Marvin

from prefect.client import get_client

isn’t valid, it should be

from prefect import get_client

My apologies for the confusion. The correct import statement for

get_client

from the Prefect library is indeed

from prefect import get_client

. Thank you for pointing that out. With that corrected, here's how you might use

get_client

within your flow to interact with Prefect's API. This script outlines the concept behind using the

get_client()

function, although it does not directly demonstrate setting a concurrency limit because such a method wasn't found in the provided documentation excerpts:

python
from prefect import flow, task, get_client

@task
def subflow_task():
    # Task implementation

@flow
def main_flow():
    with get_client() as client:
        # Pseudo-code for creating/updating concurrency limit; this is not a real method
        # You would need to look at the PrefectClient documentation or source code
        # to find the correct method to create or update concurrency limits via the API
        client.create_or_update_concurrency_limit(name="subflow-limit", limit=3)

    # Flow implementation
    subflow_task.submit()

# Run the main flow
if __name__ == "__main__":
    main_flow()

Remember that the function

create_or_update_concurrency_limit

is hypothetical and serves as a placeholder for the appropriate API call necessary to create or update concurrency limits programmatically. You would need actual API documentation or additional research to determine the precise method for this action. If such a method does not exist or is undocumented, consider reaching out to Prefect support or checking Prefect's API reference for more information.

@Marvin are all concurrency controls global? How do I keep the controls to my flow?

No, not all concurrency controls in Prefect are global. Prefect provides the ability to manage concurrency within the context of a specific flow using the

concurrency

context manager. Although the term "global concurrency limits" is used, this does not necessarily mean the limits are globally enforced across all flows. Instead, it refers to a mechanism that you can reuse across different parts of your workflows. You can use the

concurrency

context manager within your flow to limit the number of task runs that can execute concurrently, and such limits are effectively local to the tasks where they are applied. Here is an example showing how you might use the

concurrency

context manager within a flow:

python
from prefect import flow, task
from prefect.concurrency.sync import concurrency

@task
def process_data(x, y):
    with concurrency("data-processing-limit", occupy=1):
        # ... process data

@flow
def my_flow():
    for x, y in [(1, 2), (2, 3), (3, 4), (4, 5)]:
        process_data.submit(x, y)

if __name__ == "__main__":
    my_flow()

In this example, a concurrency limit named

data-processing-limit

is used locally within the

process_data

task to ensure that no more than a certain number of instances of this task run at the same time. If you have a use case where you want to make sure that the concurrency limit is only applied within a particular flow, you can ensure the limit name is unique to the flow, thereby avoiding effects on other flows. However, keep in mind that the concurrency context manager does not enable you to actually set the maximum number of concurrent tasks from within the flow. Instead, it creates the limit in an inactive state if it does not already exist. To specify the limit itself, you would still need to use the Prefect UI/Cloud or CLI to set the value for that limit. For further information, consult the Prefect documentation on global concurrency limits.