<@ULVA73B9P> I need to set up tag concurency and w...
# ask-marvin
f
@Marvin I need to set up tag concurency and would like to do that programatically each time I deploy. I am using prefect.yaml and I need to run my python code/cli once on deploy. Any supported way to do that?
m
thought for 431.0 seconds
Yes—there’s a supported way to run code once at deploy time with prefect.yaml. In Prefect 3.x, you can add “build” (or “push”) steps that run arbitrary shell commands. Use that to create/update your tag-based concurrency limits each time you run
prefect deploy
. Two common approaches: Option A: Use the Prefect CLI in a run_shell_script step - We verified these commands exist: -
prefect concurrency-limit create <TAG> <LIMIT>
-
prefect concurrency-limit inspect <TAG>
-
prefect concurrency-limit delete <TAG>
-
prefect concurrency-limit reset <TAG>
- Example prefect.yaml snippet that ensures a tag limit is set at deploy time:
Copy code
prefect-version: ">=3.0.0"

deployments:
  - name: my-deployment
    entrypoint: flows/my_flow.py:flow
    work_pool:
      name: my-pool
    build:
      - type: prefect.deployments.steps.utility.run_shell_script
        id: ensure-tag-concurrency
        script: |
          set -euo pipefail
          TAG="my-tag"
          LIMIT="5"

          # If the limit exists, delete it to enforce desired value
          if prefect concurrency-limit inspect "$TAG" >/dev/null 2>&1; then
            prefect concurrency-limit delete "$TAG"
          fi
          prefect concurrency-limit create "$TAG" "$LIMIT"
- Notes: - If you prefer “create only if missing,” replace the delete/create with:
prefect concurrency-limit inspect "$TAG" >/dev/null 2>&1 || prefect concurrency-limit create "$TAG" "$LIMIT"
- These build steps run when you execute
prefect deploy
, not during flow runs. - Ensure your deploy environment has Prefect configured (e.g.,
PREFECT_API_URL
and
PREFECT_API_KEY
if using Cloud). Option B: Run a small Python script from a run_shell_script step - If you prefer using the Python client: create file scripts/configure_concurrency.py
Copy code
import asyncio
from prefect.client.orchestration import get_client

TAG = "my-tag"
LIMIT = 5

async def main():
    async with get_client() as client:
        # Delete existing limit if present, then re-create to enforce desired value
        try:
            await client.read_concurrency_limit_by_tag(TAG)
            await client.delete_concurrency_limit_by_tag(TAG)
        except Exception:
            pass
        await client.create_concurrency_limit(tag=TAG, concurrency_limit=LIMIT)

if __name__ == "__main__":
    asyncio.run(main())
then in prefect.yaml
Copy code
build:
  - type: prefect.deployments.steps.utility.run_shell_script
    script: |
      set -euo pipefail
      python scripts/configure_concurrency.py
Notes and alternatives - There’s no “update” for tag concurrency limits—change the value by deleting and re-creating the limit. - If you actually want a per-deployment cap (not tag-based), you can set a
concurrency_limit
directly on the deployment in prefect.yaml. - For global concurrency limits, there’s a separate CLI (
prefect global-concurrency-limit ...
). Docs and references - Tag-based concurrency limits: Docs - CLI for tag concurrency limits: CLI reference - Global concurrency limits: Docs - prefect.yaml and deployment steps: Docs - run_shell_script step: API reference If you share the tags and desired limits, I can sketch the exact YAML snippet for your setup.
f
@Marvin ok, but i have 10+ deployments, the steps are run with each one. I only need it to run once
m
thought for 182.7 seconds
You’re right: in Prefect 3.x, build/push steps in prefect.yaml run per deployment. There isn’t a built-in “run once for the whole project” hook. Here are supported patterns to run it only once: Option 1: Assign the step to a single “primary” deployment - Put the concurrency-setup step only in one deployment’s
build
section. All other deployments omit or override
build
with an empty list so they don’t run it. Example:
Copy code
prefect-version: ">=3.0.0"

deployments:
  - name: primary-deployment
    entrypoint: flows/a.py:flow
    build:
      - type: prefect.deployments.steps.utility.run_shell_script
        id: ensure-tag-concurrency
        script: |
          set -euo pipefail
          for pair in "tagA:5" "tagB:2"; do
            TAG="${pair%%:*}"
            LIMIT="${pair##*:}"
            # idempotent: only create if missing
            prefect concurrency-limit inspect "$TAG" >/dev/null 2>&1 || prefect concurrency-limit create "$TAG" "$LIMIT"
          done

  - name: other-deployment-1
    entrypoint: flows/b.py:flow
    build: []  # override so nothing runs here

  - name: other-deployment-2
    entrypoint: flows/c.py:flow
    build: []
- Benefit: runs once when you deploy the primary deployment (e.g., with
prefect deploy --all
or when you include the primary one). Others won’t run it. Option 2: Use a top-level build step and override it away in others - Define the concurrency step in top-level
build:
and then add
build: []
in every deployment except one that should inherit it. This yields the same effect as Option 1 but keeps the logic in a single place. Option 3: Run once in your CI/CD before
prefect deploy
- Add a pipeline step that runs:
Copy code
prefect concurrency-limit inspect my-tag >/dev/null 2>&1 || prefect concurrency-limit create my-tag 5
- This is cleanest if you have many deployments and you always deploy via CI. Option 4: Make the step idempotent - Even if the step executes multiple times, keep it a quick no-op when already configured:
Copy code
prefect concurrency-limit inspect "$TAG" >/dev/null 2>&1 || prefect concurrency-limit create "$TAG" "$LIMIT"
- This ensures it only “does work” once, even if evaluated per deployment. CLI we verified: -
prefect concurrency-limit create <TAG> <LIMIT>
-
prefect concurrency-limit inspect <TAG>
-
prefect concurrency-limit delete <TAG>
-
prefect concurrency-limit reset <TAG>
Reference: CLI reference, Tag-based concurrency limits, prefect.yaml docs Quick question to tailor the setup: Do you deploy via a single
prefect deploy --all
in CI, or do you deploy individual deployments ad-hoc? Based on that, I can share the exact YAML layout that will run it exactly once in your flow.
f
@Marvin I do both, which makes this unnecessary complex. There really should be a global build step that is always ran..
m
thought for 84.3 seconds
I hear you—today there’s no “global build step” that runs once per project in prefect.yaml. Build/push steps are evaluated per deployment. I agree a project-level pre-deploy hook would reduce this friction. Practical ways to run it exactly once while supporting both --all and ad‑hoc deploys: Option 1: Wrapper script that always runs before deploy (works for both modes) - Put your concurrency setup in a script and wrap
prefect deploy
with a small shell function. scripts/setup_concurrency.sh
Copy code
#!/usr/bin/env bash
set -euo pipefail

# Define your desired limits here
declare -A LIMITS=(
  [tagA]=5
  [tagB]=2
  # add more tags:limits here
)

for TAG in "${!LIMITS[@]}"; do
  LIMIT="${LIMITS[$TAG]}"
  # Idempotent: only create if missing
  if ! prefect concurrency-limit inspect "$TAG" >/dev/null 2>&1; then
    prefect concurrency-limit create "$TAG" "$LIMIT"
  fi
done
Shell helper (put in your rc file or a small runner script)
Copy code
pdeploy() {
  ./scripts/setup_concurrency.sh
  prefect deploy "$@"
}
- Then use
pdeploy --all
in CI and
pdeploy -n my-deployment
for ad‑hoc. Setup runs once per invocation and is a no-op after the first time. Option 2: One “primary” deployment runs the step; others override build to empty - Keep the concurrency step only in one deployment’s
build:
and set
build: []
on all others. When you need to deploy all, ensure the primary deployment is included; for ad‑hoc, deploy the primary first (or rely on the idempotent wrapper above to avoid ordering requirements). Option 3: CI-only setup before any deploy - In your pipeline:
Copy code
./scripts/setup_concurrency.sh
prefect deploy --all
- Locally, use the wrapper function from Option 1 so you don’t have to think about it. Notes - Verified CLI: -
prefect concurrency-limit create <TAG> <LIMIT>
-
prefect concurrency-limit inspect <TAG>
-
prefect concurrency-limit delete <TAG>
-
prefect concurrency-limit reset <TAG>
Docs: CLI reference, Tag-based concurrency limits - If you ever need to enforce updated limits, switch the script to delete/recreate for each tag:
Copy code
prefect concurrency-limit inspect "$TAG" >/dev/null 2>&1 && prefect concurrency-limit delete "$TAG"
prefect concurrency-limit create "$TAG" "$LIMIT"
Feature request - I can open a feature request for a project-level “pre-deploy” step that runs once per
prefect deploy
invocation, regardless of the number of deployments. Want me to file that for you? If yes, any specifics you want included (e.g., should it run before any build steps, how it should behave with
--name
filters, etc.)?