I am trying to do it with the context manager, have a simple flow that runs some dbt operations:

from typing import List

from prefect import flow
from prefect.concurrency.sync import concurrency

from utils import dbt_core_operation_wrapper

@flow
def build_waze() -> List[str]:
    with concurrency(names='build_waze', occupy=1, timeout_seconds=3600):
        dbt_core_operation_wrapper(['dbt build --select waze'])

This created a global concurrency limit

build_waze

, as seen in the screenshot. I manually activated the limit. Now after the first next run Active Slots went to 1, but: • it stays 1 even after no flow is running • it enabled me to do a quick run of the same flow and entered the Running state (so not preventing multiple flows to be run which is my goal)

Found relevant issue on GitHub: • https://github.com/PrefectHQ/prefect/issues/12456