Thanks for the detailed context — that helps a lot. Here’s a robust, automation-first architecture for your use case (Prefect 3.4.15, self-hosted, Postgres, process worker) that will let “service X” provision a new store by creating a deployment and kicking off an initial run to create tables. High-level architecture - Keep one reusable, parameterized dbt flow (clickstream_dbt_run). - For each new store, programmatically create a new deployment from that flow with store-specific defaults. - Enforce per-deployment concurrency (limit=1, CANCEL_NEW) to avoid overlapping runs for the same store. - After creating the deployment, trigger a “bootstrap” run that sets

full_refresh=true

(or similar) to create/refresh tables. - Back the deployment from source (your Git repo) so the worker always pulls latest code. - Expose an internal Provisioning API (FastAPI or similar) that service X calls; it uses Prefect’s Python API to create/update deployments and trigger the first run. Why this works well - It’s fully API-driven (no manual

prefect deploy

runs needed). - It scales to many stores (1 deployment per store). - It respects your current infra (EC2, Prefect Server w/ Postgres, process worker). - It preserves your dbt conventions and lets you keep using a single flow with parameters. Step-by-step setup 0) Prereqs (confirm these once) - Prefect Server configured to use Postgres (already done) - One work pool for your process worker, e.g.

local-work-pool

- Create if needed:

prefect work-pool create local-work-pool --type process

- A worker running against that pool on your EC2 instance:

prefect worker start -p local-work-pool

- Secret block for your Git access token (the same one you reference in prefect.yaml): - In Python once:

from prefect.blocks.system import Secret
    Secret(value="<YOUR_GIT_TOKEN>").save("dbt-git-repo-token", overwrite=True)

- Or create it in the UI. 1) Make sure your dbt flow is parameterized Your flow already takes params like

database

,

models

,

full_refresh

,

store_timezone_offset_minutes

, etc. Perfect. Keep that entrypoint: -

flows/clickstream/clickstream.py:clickstream_dbt_run

2) Provisioning service (API) to create deployments per store Deploy this (FastAPI example) alongside your EC2 services. This endpoint: - Creates/updates a deployment for a given store - Enforces per-deployment concurrency (limit=1, CANCEL_NEW) - Triggers a bootstrap run with

full_refresh=True

on first provision ``` from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional from prefect.client.orchestration import get_client from prefect.runner.storage import GitRepository from prefect.blocks.system import Secret from prefect import flow from prefect.client.schemas.objects import ConcurrencyLimitConfig, ConcurrencyLimitStrategy # IMPORTANT: Ensure ENV is set where this service runs: # - PREFECT_API_URL=http//<your prefect server host>4200/api # If you use Prefect Cloud, also set PREFECT_API_KEY. app = FastAPI(title="Prefect Provisioning API") class ProvisionRequest(BaseModel): store_id: str database: str models: Optional[str] = None # e.g., "tag:clickstream" branch: Optional[str] = "dev" store_timezone_offset_minutes: Optional[str] = "330" is_complete_clickstream_run: Optional[bool] = True default_full_refresh: Optional[bool] = False repo_url: str = "https://github.com/xyzorg/abcrepo.git" def _build_flow_from_source(repo_url: str, branch: str): # Use a GitRepository with Secret block for private repo access git_token = Secret.load("dbt-git-repo-token") # created earlier return flow.from_source( source=GitRepository( url=repo_url, branch=branch, credentials={"access_token": git_token} ), entrypoint="flows/clickstream/clickstream.py:clickstream_dbt_run", ) @app.post("/provision-store")

async def provision_store(req: ProvisionRequest): # 1) Reference the flow from source (repo + entrypoint) try: store_flow = _build_flow_from_source(req.repo_url, req.branch) except Exception as exc: raise HTTPException(status_code=400, detail=f"Failed to load flow from source: {exc}") # 2) Create or update a store-specific deployment deployment_name = f"clickstream-{req.store_id}" try: deployment_id = store_flow.deploy( name=deployment_name, work_pool_name="local-work-pool", tags=["clickstream-aggregation", "prod", f"store:{req.store_id}"], description=( "Clickstream Aggregation\n\n" "Args:\n" f" database: {req.database}\n" f" models: {req.models}\n" f" full_refresh (default): {req.default_full_refresh}\n" f" store_timezone_offset_minutes: {req.store_timezone_offset_minutes}\n" ), parameters={ "database": req.database, "models": req.models or "tag:clickstream", "full_refresh": req.default_full_refresh, "store_timezone_offset_minutes": req.store_timezone_offset_minutes, "is_complete_clickstream_run": req.is_complete_clickstream_run, }, concurrency_limit=ConcurrencyLimitConfig( limit=1, collision_strategy=ConcurrencyLimitStrategy.CANCEL_NEW ), ) except Exception as exc: raise HTTPException(status_code=500, detail=f"Deploy failed: {exc}") # 3) Trigger a bootstrap run that forces full table creation/refresh once # (overrides the deployment default full_refresh=False) try: async with get_client() as client: flow_run = await client.create_flow_run_from_deployment( deployment_id=deployment_id, parameters={ "database": req.database, "models": req.models or "tag:clickstream", "store_timezone_offset_minutes": req.store_timezone_offset_minutes, "is_complete_clickstream_run": req.is_complete_clickstream_run, "full_refresh": True, # override to create/refresh tables on first run }, name=f"bootstrap-{req.store_id}", tags=[f"store:{req.store_id}", "bootstrap"], ) return { "deployment_id": str(deployment_id), "bootstrap_flow_run_id": str(flow_run.id), "message": f"Provisioned {deployment_name} and started bootstrap run." } except Exception as exc: # Deployment is created; report only the run failure raise HTTPException( status_code=500, detail=f"Deployment created but failed to start bootstrap run: {exc}" ) ``` Notes - This is idempotent for deployment name. Calling it again with same

store_id

updates the deployment defaults (parameters, branch, description, tags, etc.) and you can choose whether to start another run. - You can add schedules later by passing

cron=...

,

interval=...

, or

schedules=[...]

into

.deploy()

if/when you want per-store scheduling. - If you want to keep some deployments defined in prefect.yaml for “known stores” and only use the API for new stores, that also works — both methods happily coexist in Prefect 3.x. 3) Wire service X to your Provisioning API - Service X calls POST /provision-store with the new store’s info as soon as integration completes. - The worker polls

local-work-pool

, pulls the code from your repo (auth via Secret block), and executes the bootstrap run. 4) Optional hardening and ops - Run Prefect Server and worker as systemd services on EC2 so they auto-restart. - Keep

PREFECT_API_URL

exported in the Provisioning API environment (e.g., http://localhost:4200/api).

- If you ever move to Docker or Kubernetes workers, swap the work pool type and pass appropriate

job_variables

(env, memory, cpu, etc.) in

.deploy()

. 5) Concurrency behavior You mentioned: - limit: 1 - collision_strategy: CANCEL_NEW In Prefect 3.x Python, that maps to:

from prefect.client.schemas.objects import ConcurrencyLimitConfig, ConcurrencyLimitStrategy

concurrency_limit=ConcurrencyLimitConfig(
    limit=1,
    collision_strategy=ConcurrencyLimitStrategy.CANCEL_NEW
)

This ensures only one run per store is active at a time; excess runs are immediately canceled. 6) Triggering additional runs From anywhere in your stack, you can later trigger parameterized runs of an existing deployment:

from prefect.client.orchestration import get_client

async def trigger_run(flow_name, deployment_name, params: dict):
    async with get_client() as client:
        deployment = await client.read_deployment_by_name(f"{flow_name}/{deployment_name}")
        flow_run = await client.create_flow_run_from_deployment(
            deployment_id=deployment.id,
            parameters=params,
            name="on-demand",
            tags=["manual"],
        )
        return flow_run.id

Trade-offs and an alternative - Per-store deployments are great for isolation and visibility. If you expect thousands of stores, consider a single shared deployment and just trigger runs with a

store_id

parameter — it’ll be lighter weight operationally. Your current pattern (one deployment per store) is totally fine at moderate scale and matches your preference. Quick checklist to validate the end-to-end path - Worker is polling

local-work-pool

and has network access to your Git repo and databases. - Secret block

dbt-git-repo-token

exists. - Provisioning API has

PREFECT_API_URL

set correctly. - Repo branch contains the flow at

flows/clickstream/clickstream.py:clickstream_dbt_run

. Questions to tailor the final setup - Do you want the provisioning to also attach schedules immediately (e.g., interval every 30 minutes), or do you prefer fully on-demand runs? - Is your repo always private and branch

dev

, or should the API accept a per-store branch? - Any store-specific dbt target/schema naming conventions we should encode (e.g., schema per store)? - Do you need to toggle

full_refresh

for periodic re-initialization, or is it strictly a one-time bootstrap? If you share those details, I can refine the provisioning endpoint (and the deployment defaults) to exactly fit your workflow.

Great question. In Prefect there are two main ways to talk to the orchestration API: 1) Python SDK (Client + high-level helpers) - What it is: The Python package you’re already using (

prefect

). It includes: - A typed client for API calls - High-level helpers like

flow.deploy()

,

flow.from_source(...)

,

Secret.save()

,

client.create_flow_run_from_deployment(...)

, etc. - How it’s used: - From your Python code (flows, provisioning scripts/services) - The Prefect CLI also uses this under the hood - Why use it: - Convenience: fewer lines of code, type-checked models, retries, and sensible defaults - Features: helpers for Git-backed code via

flow.from_source(...)

, block management (

.save()

), schedules, concurrency, etc. - Returns Python objects you can work with immediately - When to choose it: - Your provisioning logic is in Python (easiest/most ergonomic) - You want to create/update deployments from a Git repo with a Secret block - You need to trigger runs, manage blocks, or update schedules from code 2) REST API (HTTP/JSON endpoints) - What it is: The raw HTTP endpoints exposed by your Prefect server: - CRUD for deployments, flows, blocks, flow runs, schedules, etc. - Examples:

POST /deployments

,

POST /deployments/{id}/create_flow_run

,

GET /deployments/name/{flow}/{deployment}

- How it’s used: - From any language or system (curl, Postman, Go/Java/Node, etc.) - Webhooks or platforms that can only make HTTP calls - Why use it: - Language-agnostic - Integrate Prefect directly into non-Python systems - When to choose it: - Service X is not Python and you don’t want to add a thin Python service - You need lowest-level control and are comfortable crafting JSON payloads Key differences - Ergonomics: SDK is higher-level and safer; REST is lower-level and more manual. - Language: SDK = Python only; REST = any language that can call HTTP. - Features: Everything the SDK does eventually hits REST under the hood, but the SDK adds powerful convenience (e.g.,

flow.from_source(...).deploy(...)

and block serialization). How this applies to your setup - You’re already in Python and using Prefect/DBT on EC2. The simplest, most reliable path is to build a small Python “Provisioning API” (e.g., FastAPI) that: - Uses the Python SDK to create per-store deployments via

flow.from_source(...).deploy(...)

- Enforces per-deployment concurrency (limit=1, CANCEL_NEW) - Immediately triggers a first run to create tables - If service X is not Python, it can either: - Call your Provisioning API (recommended), or - Call Prefect’s REST endpoints directly (you’ll need to craft the JSON and handle auth) Minimal examples - Python SDK: trigger a run from an existing deployment

from prefect.client.orchestration import get_client

async def trigger():
    async with get_client() as client:
        dep = await client.read_deployment_by_name("clickstream_dbt_run/clickstream-2427b5-28")
        fr = await client.create_flow_run_from_deployment(
            deployment_id=dep.id,
            parameters={"full_refresh": True}
        )
        print(fr.id)

- REST API: trigger a run (curl)

# For self-hosted Prefect Server OSS, typically no API key required by default.
# Adjust URL to your server and supply auth headers if applicable.

curl -X POST http://<your-prefect-server>:4200/api/deployments/<DEPLOYMENT_ID>/create_flow_run \
  -H "Content-Type: application/json" \
  -d '{
        "parameters": {"full_refresh": true},
        "name": "bootstrap-run",
        "tags": ["bootstrap"]
      }'

- REST API: read deployment by name (useful to get the ID)

curl http://<your-prefect-server>:4200/api/deployments/name/<FLOW_NAME>/<DEPLOYMENT_NAME>

Notes on auth/config - Self-hosted Prefect Server (OSS): usually no API key; set

PREFECT_API_URL

for SDK usage (e.g., http://host:4200/api).

- Prefect Cloud: set both

PREFECT_API_URL

and

PREFECT_API_KEY

for SDK and add Authorization headers for REST. Docs - SDK overview: Prefect Docs - Deployments: Deployments - API reference: API Reference Recommendation for you - Implement the provisioning step in Python using the SDK. It’s the fastest way to: - Create per-store deployments from your Git repo - Apply concurrency = 1 with CANCEL_NEW - Trigger the first run for table creation - If service X can’t call Python, expose a small internal HTTP endpoint that wraps the SDK (thin FastAPI service). This keeps all logic consistent and avoids hand-crafting REST payloads.