@Marvin Is there a way from within a task to get the count of how many other tasks are running?

To get the count of how many other tasks are running within a task in Prefect 3.x, you would typically interact with the Prefect API. Here's a general approach using Prefect's client: 1. Get the Client: Use

prefect.client.get_client()

to create a client instance. 2. Read Task Runs: Use the client's

read_task_runs()

method to retrieve information about task runs. You can filter these by state to count the running tasks. Here's a simple example:

python
from prefect import task, flow
from prefect.client import get_client
from prefect.states import StateType
import asyncio

@task
async def count_running_tasks():
    async with get_client() as client:
        # Retrieve all task runs
        runs = await client.read_task_runs()
        
        # Count the number of tasks currently running
        running_count = sum(1 for run in runs if run.state.type == StateType.RUNNING)
        
        print(f"Number of running tasks: {running_count}")
        return running_count

@flow
def my_flow():
    count_running_tasks()

my_flow()

This example assumes that you have access to the Prefect API and the necessary permissions to query task runs. You might need to adjust the filtering logic based on your specific requirements. For additional details, you can refer to the Prefect documentation and explore the client API. If you want to explore more about Prefect's API offerings, you can check the Prefect API documentation.