https://prefect.io
Join Slack
<@ULVA73B9P> how does the `with tags(...)` work is relation to independent deployment flows?
# ask-marvin
s
@Marvin how does the 
with tags(...)
work is relation to independent deployment flows?
m
✅ thought for 120.2 seconds
with tags(...)
adds runtime tags to the current execution context. So for independent deployments: - it does not link deployments together - it does not modify the deployment definition itself - it does tag the specific flow run and any task runs created inside that context Example:
Copy code
from prefect import flow, task, tags

@task
def my_task():
    pass

@flow
def my_flow():
    with tags("customer-a", "nightly"):
        my_task()
When that flow runs: - the flow run gets 
customer-a
and 
nightly
- 
my_task()
also inherits those tags for that run If you have two deployments of the same flow, they are still independent. But if both deployments execute code under the same 
with tags("x")
, then runs from both deployments will carry that tag. So the relationship is: - deployment tags = metadata configured on the deployment - 
with tags(...)
= metadata applied at runtime to a specific run/context The important caveat is tag-based concurrency limits. If you use tags that have concurrency limits attached, then those limits can affect runs across deployments. For example, if tasks in deployment A and deployment B both end up with the tag 
database
, they may contend for the same tag-based concurrency limit. Example idea:
Copy code
@task(tags=["database"])
def query_db():
    pass

@flow
def flow_a():
    with tags("team-a"):
        query_db()

@flow
def flow_b():
    with tags("team-b"):
        query_db()
Here: - 
flow_a
and 
flow_b
deployments are independent - but 
query_db()
in both can still share the same concurrency limit for 
database
So the short version: - 
with tags(...)
is per-run context - it affects flow runs and child task runs - deployments remain independent - except shared tags can matter for things like tag-based concurrency limits, which can apply across deployments Docs and references: - Prefect tags concept: https://docs.prefect.io - Prefect Python API reference: https://reference.prefect.io If you want, I can also show: 1. how 
with tags(...)
differs from deployment tags, or 2. a concrete example with two deployments and a shared concurrency limit
s
@Marvin if there are two deployment dep_a and dep_b that each have flow_a. In flow_a there is a concurrency tag called "database", both dep_a.flow_a and dep_b.flow_a are running. dep_a.flow_a gets a hold on the concurrency tag slot first, does that block dep_b.flow_a until the slot is released by dep_a.flow_a?
m
✅ thought for 9.6 seconds
Yes — if both runs are trying to execute a task/run carrying the same concurrency-limited tag and the limit is already full, then 
dep_b.flow_a
will wait until 
dep_a.flow_a
releases the slot. The key nuance is: - the blocking is not because they are separate deployments - the blocking is because they share the same tag-based concurrency limit - this usually applies at the task run level for tagged tasks So if you have something like:
Copy code
from prefect import flow, task

@task(tags=["database"])
def query_db():
    ...
and you configured a concurrency limit of 
1
for 
database
, then: - 
dep_a.flow_a
starts 
query_db()
and acquires the 
database
slot - 
dep_b.flow_a
reaches 
query_db()
while that slot is still occupied - 
dep_b.flow_a
must wait until 
dep_a.flow_a
finishes and releases it In effect: yes, one can block the other. A few important clarifications: 1. It blocks the tagged work, not necessarily the whole flow immediately If the flow has work before the 
database
-tagged task, that earlier work can still run. The waiting happens when execution reaches the constrained task. 2. This is shared across deployments The concurrency limit is attached to the tag, not to the deployment. So all task runs with tag 
database
compete for the same slots, regardless of whether they came from: - 
dep_a
- 
dep_b
- another flow entirely 3. If the limit is greater than 1, multiple runs can proceed If 
database
has limit 
3
, then up to 3 matching task runs can run concurrently across all deployments. Example mental model: - concurrency limit on 
database
= 1 - 
dep_a.flow_a -> query_db()
acquires slot - 
dep_b.flow_a -> query_db()
waits - when 
dep_a
finishes 
query_db()
, slot is released - 
dep_b
can then proceed If helpful, I can draw a timeline for `dep_a`/`dep_b` showing exactly when the wait happens.
s
@Marvin this is for Prefect 2.20.16 does that change how the concurrency slots work?
m
✅ thought for 127.7 seconds
Yes — for Prefect 2.20.16, the behavior you described is still basically yes. If both deployments run tasks tagged 
database
, and there is a tag concurrency limit on 
database
, then they share that same pool of slots across deployments. So if: - 
dep_a.flow_a
reaches a task tagged 
database
- it acquires the only available slot - 
dep_b.flow_a
then reaches a task tagged 
database
then `dep_b.flow_a`'s task run will wait until the slot is released. A few important details for Prefect 2.20.16: What is being limited? - In 2.20.16, tag concurrency limits apply to task runs, not flow runs. - So 
dep_b.flow_a
itself is not globally blocked from starting. - It will only pause when it tries to run the tagged task. Is the limit shared across deployments? - Yes. - The concurrency limit is keyed by the tag, not the deployment. - So all tasks with tag 
database
compete for the same limit, no matter which deployment they came from. Does Prefect 2 differ from Prefect 3 here? - For this specific behavior, not in the way that matters to your example. - In both versions, shared tagged work can contend across deployments. - The main practical point remains: a 
database
-tagged task in one deployment can make another deployment's 
database
-tagged task wait. Example:
Copy code
from prefect import flow, task

@task(tags=["database"])
def query_db():
    ...
    
@flow
def flow_a():
    query_db()
If you deploy 
flow_a
twice as 
dep_a
and 
dep_b
, and set:
Copy code
prefect concurrency-limit create database 1
then the sequence is: 1. 
dep_a.flow_a
starts 2. 
dep_a
reaches 
query_db()
and gets the 
database
slot 3. 
dep_b.flow_a
starts 4. 
dep_b
reaches 
query_db()
and cannot get the slot 5. 
dep_b
waits/retries until 
dep_a
finishes 
query_db()
So the short answer is: Yes — in Prefect 2.20.16, if both deployments run tasks sharing the same constrained tag, one deployment's tagged task can block the other's tagged task until the slot is freed. If you want, I can also show you how this differs from: - work queue / worker concurrency - deployment-level concurrency - task tag concurrency Those three are easy to mix up in Prefect 2.x.
s
@Marvin what is the prefect cli command to check whether a concurrency-limit exists or not?
m
✅ thought for 16.7 seconds
Use:
Copy code
prefect concurrency-limit inspect <tag>
For example:
Copy code
prefect concurrency-limit inspect database
If it exists, Prefect will show its details. If it does not exist, you'll get an error indicating no limit was found for that tag. If you want to list all existing concurrency limits first, use:
Copy code
prefect concurrency-limit ls
Verified CLI help: - 
prefect concurrency-limit inspect TAG
- 
prefect concurrency-limit ls
If you want, I can also show a small shell snippet to test existence programmatically in bash.
s
@Marvin does 
prefect concurrency-limit ls
only show a subset of tags?
m
✅ thought for 12.6 seconds
Yes — by default, 
prefect concurrency-limit ls
only shows a subset. The CLI help shows:
Copy code
prefect concurrency-limit ls --help
with options: - 
--limit INTEGER  [default: 15]
- 
--offset INTEGER [default: 0]
So by default it shows only the first 15 concurrency limits. To see more, increase the limit, for example:
Copy code
prefect concurrency-limit ls --limit 100
You can also paginate with offset:
Copy code
prefect concurrency-limit ls --limit 100 --offset 100
So if you're checking whether a specific tag exists, 
inspect
is usually more reliable:
Copy code
prefect concurrency-limit inspect database
s
@Marvin is a concurrency-limit slot created in python using 
with tag("name-of-tag")
?
m
✅ thought for 7.8 seconds
No. 
with tags("name-of-tag")
does not create a concurrency limit. It only adds a tag to the current runtime context. A concurrency limit must be created separately, typically with the CLI:
Copy code
prefect concurrency-limit create name-of-tag 1
Then, if a task run has that tag, Prefect will enforce the limit. So the distinction is: - 
with tags("database")
- adds the tag 
database
to the current flow/task run context - does not create any limit by itself - 
prefect concurrency-limit create database 1
- creates an actual concurrency rule for that tag Example:
Copy code
from prefect import flow, task, tags

@task
def work():
    ...

@flow
def my_flow():
    with tags("database"):
        work()
This alone just tags the run. There is no concurrency control unless a matching limit already exists for 
database
. One more important Prefect 2.x nuance: - tag-based concurrency limits are enforced on task runs - so using 
with tags("database")
matters when tasks executed in that context inherit that tag - but the tag itself is still just metadata unless a concurrency limit for 
database
exists Related CLI commands: - create limit: 
prefect concurrency-limit create <tag> <limit>
- inspect limit: 
prefect concurrency-limit inspect <tag>
- list limits: 
prefect concurrency-limit ls
If you want, I can show a complete Prefect 2.20.16 example proving when 
with tags(...)
does and does not affect concurrency.
s
@Marvin can a tag be created in python and assigned a concurrency limit and if the concurrency limit tag already exists, it is just updated?
m
✅ thought for 5.9 seconds
Not directly with 
with tags(...)
. In Prefect 2.20.16: - Python code can apply tags to runs/tasks - but creating or updating the concurrency limit record for a tag is generally done through the API/CLI, not by the tag context manager itself So:
Copy code
with tags("database"):
    ...
does not: - create the tag in the backend as a concurrency object - create a concurrency limit - update an existing concurrency limit It only uses the string tag at runtime. If you want the actual concurrency limit behavior, you need something like:
Copy code
prefect concurrency-limit create database 1
If 
database
already exists, 
create
is not an "upsert" command. In practice, you would inspect/delete/recreate or use the underlying API. Typical CLI workflow:
Copy code
prefect concurrency-limit inspect database
prefect concurrency-limit ls
prefect concurrency-limit delete database
prefect concurrency-limit create database 5
So the answer to your exact question is: - Can a tag be created in Python? Yes, as a runtime tag string on flow/task runs. - Can it be assigned a concurrency limit just by doing that? No. - If the concurrency-limit tag already exists, is it just updated automatically? No, not via 
with tags(...)
. If you're asking whether you can programmatically manage concurrency limits from Python using the Prefect client/API in 2.20.16, that's a slightly different question — and the answer is generally yes, via the API, but not via the tag context manager itself. If you want, I can give you: - the CLI-safe way to "create or update" a concurrency limit, or - a Python API example for Prefect 2.20.16.
s
@Marvin give both - the CLI-safe way to "create or update" a concurrency limit, or - a Python API example for Prefect 2.20.16.
m
✅ thought for 64.6 seconds
Yep — here are both. 1. CLI-safe “create or update” pattern In Prefect 2.20.16, the CLI has 
create
, 
inspect
, 
delete
, 
ls
, and 
reset
for task tag concurrency limits. Verified command:
Copy code
prefect concurrency-limit create <tag> <limit>
A safe shell pattern is:
Copy code
TAG="database"
LIMIT="5"

if prefect concurrency-limit inspect "$TAG" >/dev/null 2>&1; then
  echo "Concurrency limit exists for $TAG; recreating it with new value $LIMIT"
  prefect concurrency-limit delete "$TAG"
fi

prefect concurrency-limit create "$TAG" "$LIMIT"
prefect concurrency-limit inspect "$TAG"
That works reliably from the CLI. If you want it as a one-liner-ish pattern:
Copy code
prefect concurrency-limit inspect database >/dev/null 2>&1 && \
prefect concurrency-limit delete database

prefect concurrency-limit create database 5
Important note: the CLI 
create
command is not presented as an explicit 
update
command, so the safest CLI workflow is: - inspect - delete if present - create Useful commands:
Copy code
prefect concurrency-limit inspect database
prefect concurrency-limit ls --limit 100
prefect concurrency-limit delete database
prefect concurrency-limit create database 5
2. Python API example for Prefect 2.20.16 In Python, you can manage concurrency limits programmatically via the Prefect client. The relevant method is effectively an upsert: - if the tag does not exist, it creates it - if it already exists, it updates the concurrency limit value Example:
Copy code
import asyncio
from prefect import get_client

async def main():
    async with get_client() as client:
        # Create or update the concurrency limit
        limit_id = await client.create_concurrency_limit(
            tag="database",
            concurrency_limit=5,
        )
        print(f"limit id: {limit_id}")

        # Read it back
        limit = await client.read_concurrency_limit_by_tag("database")
        print(f"tag={limit.tag}, limit={limit.concurrency_limit}, active_slots={limit.active_slots}")

asyncio.run(main())
If you want a more explicit helper:
Copy code
import asyncio
from prefect import get_client

async def create_or_update_concurrency_limit(tag: str, value: int):
    async with get_client() as client:
        await client.create_concurrency_limit(
            tag=tag,
            concurrency_limit=value,
        )
        return await client.read_concurrency_limit_by_tag(tag)

async def main():
    limit = await create_or_update_concurrency_limit("database", 3)
    print(limit)

asyncio.run(main())
Delete example:
Copy code
import asyncio
from prefect import get_client

async def main():
    async with get_client() as client:
        await client.delete_concurrency_limit_by_tag("database")

asyncio.run(main())
So the practical summary is: - CLI-safe approach: inspect/delete/create - Python client approach: 
create_concurrency_limit(...)
behaves like an upsert If you want, I can also give you a synchronous wrapper function for use inside normal Python scripts so you don’t have to deal with 
asyncio
directly.
s
@Marvin, can a concurrency limit and tag be created in the Prefect UI?
m
✅ thought for 5.4 seconds
Yes — the concurrency limit can be created in the UI, but the tag itself is just a string used by runs/tasks, not a separate object you pre-create in the UI first. For Prefect 2.20.16 tag-based concurrency: - you create a concurrency limit for a tag in the UI - the tag name is entered as text, like 
database
- Prefect then stores the concurrency rule for that tag - tasks/runs using that tag will be subject to the limit So practically: - Can you create the tag by itself in the UI? Not really as a standalone managed entity in the way you might create a work pool or block. - Can you create a concurrency limit for a tag in the UI? Yes. Once you type a tag name into the concurrency limit form, that tag is effectively established for concurrency purposes. If you don’t already see it, check the Concurrency / Limits area in Prefect Cloud or your server UI. General references: - Prefect docs: https://docs.prefect.io - Prefect Cloud app: Prefect Cloud If you want, I can describe exactly where to click in Prefect 2 Cloud/Server UI versus Prefect 3 UI, since the layout differs a bit.
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by