Great question. Assuming Prefect 3.x, you have a few good options to keep dev and prod repos on the same machine using different profiles and API URLs. Recommended setup: global profiles + per-repo selection 1) Create and configure dev/prod profiles (stored in ~/.prefect/profiles.toml)

prefect profile create dev
prefect profile use dev
prefect config set PREFECT_API_URL=<http://localhost:4200/api>

prefect profile create prod
prefect profile use prod
prefect config set PREFECT_API_URL=<https://api.prefect.cloud/api/accounts/ACCOUNT_ID/workspaces/WORKSPACE_ID>
prefect config set PREFECT_API_KEY=pnu_xxxxx

2) In each repo, set the profile in a .env file so it auto-selects when you run commands there: - dev repo: create .env

PREFECT_PROFILE=dev

- prod repo: create .env

PREFECT_PROFILE=prod

3) Use normally from each repo

# In dev repo
prefect config view --show-sources
prefect deploy

# In prod repo
prefect config view --show-sources
prefect deploy

Alternative 1: set API URL per repo with prefect.toml - Put the API URL directly in each repo’s prefect.toml (instead of in profiles): dev repo prefect.toml:

[api]
url = "<http://localhost:4200/api>"

prod repo prefect.toml:

[api]
url = "<https://api.prefect.cloud/api/accounts/ACCOUNT_ID/workspaces/WORKSPACE_ID>"

You can still use .env to select the profile if you want other settings to vary:

PREFECT_PROFILE=dev   # or prod

Alternative 2: fully isolated profiles per repo - Point each repo at its own profiles file via PREFECT_PROFILES_PATH: dev repo:

# .env
PREFECT_PROFILES_PATH=./.prefect/profiles.toml

Create dev-repo/.prefect/profiles.toml:

active = "dev"
[profiles.dev]
PREFECT_API_URL = "<http://localhost:4200/api>"

prod repo:

# .env
PREFECT_PROFILES_PATH=./.prefect/profiles.toml

Create prod-repo/.prefect/profiles.toml:

active = "prod"
[profiles.prod]
PREFECT_API_URL = "<https://api.prefect.cloud/api/accounts/ACCOUNT_ID/workspaces/WORKSPACE_ID>"
PREFECT_API_KEY = "pnu_xxxxx"

Helpful tips - One-off override without switching:

prefect -p prod deploy

- Verify what’s active and where values come from:

prefect profile ls
prefect profile inspect dev
prefect config view --show-sources

If you tell me which approach you prefer (centralized profiles vs per-repo isolation), I can tailor the exact file contents/commands for your repos.

Short answer: yes. In Prefect 3.x,

.map()

schedules tasks concurrently by default, which can lead to overlapping “login” calls. You have a few ways to make mapped tasks run one-at-a-time (or just limit how many run at once). Options: - Sequential (no overlap at all) - Don’t use

.map()

. Call the task in a loop directly (blocking). - Or keep

.map()

but set the flow’s task runner to a single worker.

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

@task
def do_login_then_work(x):
    # your login + work here
    return x

@flow(task_runner=ThreadPoolTaskRunner(max_workers=1))
def my_flow():
    # .map() will run sequentially because only 1 worker is available
    futures = do_login_then_work.map([1, 2, 3])
    futures.wait()
    return futures.result()

- Limit concurrency around login (recommended if you just want to prevent simultaneous logins, but still run other work concurrently) - Use a global concurrency limit (works across all runs/workers) and wrap the login section with the

concurrency

context manager. Create the global limit once:

prefect gcl create login_service --limit 1

Guard the login code:

from prefect import flow, task
from prefect.concurrency.sync import concurrency

@task
def do_login_then_work(x):
    # Only one login section runs at a time across your whole system
    with concurrency("login_service", occupy=1, strict=True):
        # login code here
        pass
    # rest of the work here
    return x

@flow
def my_flow():
    futures = do_login_then_work.map([1, 2, 3])
    futures.wait()
    return futures.result()

- Tag-based limit (simpler, per-tag) - Tag the task and cap concurrent runs with that tag:

prefect concurrency-limit create login 1

from prefect import flow, task

@task(tags=["login"])
def do_login_then_work(x):
    # login + work
    return x

@flow
def my_flow():
    futures = do_login_then_work.map([1, 2, 3])
    futures.wait()
    return futures.result()

- Login once, then map - If possible, do a single login upstream and pass the session/token to the mapped tasks:

@task
def login():
    # return token/session
    return "token"

@task
def work(x, token):
    # use token
    return x

@flow
def my_flow():
    token = login()
    futures = work.map([1, 2, 3], token=token)
    futures.wait()
    return futures.result()

Note: this works best with thread-based runners; with process-based runners the token must be picklable. A couple clarifying questions so I can recommend the best pattern: - Do you want strictly one-at-a-time execution, or just a small cap (e.g., 2-3 at once)? - Are you running on Prefect 3.x? - Is the “double login” coming from the same run on one machine, or from multiple workers? If you share a small snippet of your flow and login task, I can tailor this to your setup.

Yes—this is a great fit for a single Prefect flow. You can orchestrate multiple “download” tasks (that may run at different times), then combine their outputs and write the final result to your database, all in one flow run. Two common patterns: 1) Single long-running flow run (waits for data as it becomes available) - Each “download” is a task. - If a field isn’t ready yet, that task can wait/sleep/retry until its window opens. - Combine task takes futures from the download tasks; Prefect will wait for them automatically. - Insert task writes the combined result to your DB. Example:

from time import sleep
from prefect import flow, task

@task(retries=3, retry_delay_seconds=60)
def download_field_a():
    # e.g., wait until a specific time or condition
    # sleep(600)
    return {"a": 1}

@task(retries=3, retry_delay_seconds=60)
def download_field_b():
    # maybe this field is available later
    # sleep(3600)
    return {"b": 2}

@task
def combine(a: dict, b: dict) -> dict:
    return {*a, *b}

@task
def insert_into_db(record: dict):
    # write to your DB here
    pass

@flow
def pipeline():
    a_fut = download_field_a.submit()
    b_fut = download_field_b.submit()
    combined = combine(a_fut, b_fut)  # Prefect waits for both
    insert_into_db(combined)

2) Multiple short runs (recommended if fields arrive hours apart) - Create separate subflows or deployments that fetch each field when it becomes available (e.g., scheduled at specific times). - Persist each intermediate output (e.g., S3, filesystem, or a staging table). - A final “combine” flow reads those persisted pieces and inserts the result. - You can trigger the final step with a schedule or an Automation when both parts exist. Example skeleton:

from prefect import flow, task

@task
def write_field_a():
    # fetch A and write to storage (e.g., <s3://bucket/a.json>)
    pass

@task
def write_field_b():
    # fetch B and write to storage (e.g., <s3://bucket/b.json>)
    pass

@task
def read_and_combine():
    # read a.json + b.json, combine
    return {"a": 1, "b": 2}

@task
def insert_into_db(record):
    pass

@flow
def fetch_a_flow():
    write_field_a()

@flow
def fetch_b_flow():
    write_field_b()

@flow
def combine_and_insert_flow():
    combined = read_and_combine()
    insert_into_db(combined)

Deploy and schedule

fetch_a_flow

and

fetch_b_flow

at their respective times, then trigger

combine_and_insert_flow

when both are ready. Notes for your “double login” concern: - If the downloads share an authentication step, you can: - Login once in the flow and pass the token/session to the tasks, or - Wrap the login in a global concurrency limit so it never overlaps, even across workers: - Create once:

prefect gcl create login_service --limit 1

- In your task:

from prefect.concurrency.sync import concurrency

    @task
    def download_field_a():
        with concurrency("login_service", occupy=1, strict=True):
            # login here
            pass
        # download after login
        return {"a": 1}

A few questions to tailor this: - Are the fields arriving minutes or hours apart? Is a single long-running flow okay, or do you prefer separate runs? - Where do you want to keep intermediate results (files, object storage, staging DB)? - Which database are you inserting into (so I can provide a minimal insert example)? If you share those details, I can sketch the exact flow structure and code for your case.

Yes—do this with multiple deployments on their own schedules, then a final deployment that’s triggered only after all the “part” deployments complete. High-level pattern: - Create one flow per part (each with its own deployment and schedule). - Create a final “combine + insert” flow. - Add a compound trigger to the final deployment that waits for completion events from all part deployments within a time window. Example (2 parts shown, extend to N parts):

from datetime import timedelta
from prefect import flow, serve
from prefect.events import DeploymentCompoundTrigger

@flow
def download_part_a():
    # fetch/save part A
    return "a"

@flow
def download_part_b():
    # fetch/save part B
    return "b"

@flow
def combine_and_insert():
    # read A+B from where they were saved (S3/files/DB), combine, insert to DB
    return "done"

if __name__ == "__main__":
    dep_a = download_part_a.to_deployment(
        name="part-a",
        cron="0 9 * * *"  # e.g., daily 09:00
    )

    dep_b = download_part_b.to_deployment(
        name="part-b",
        interval=3600  # e.g., hourly
    )

    # Final runs after BOTH part-a and part-b complete within 24h
    dep_final = combine_and_insert.to_deployment(
        name="final-insert",
        triggers=[
            DeploymentCompoundTrigger(
                name="wait-for-all-parts",
                enabled=True,
                require="all",         # wait for all child triggers
                within=86400,          # seconds; adjust to your schedule spread
                triggers=[
                    {
                        "type": "event",
                        "posture": "Reactive",
                        "expect": ["prefect.flow-run.Completed"],
                        "match_related": {
                            "prefect.resource.name": "part-a",
                            "prefect.resource.role": "deployment",
                        },
                    },
                    {
                        "type": "event",
                        "posture": "Reactive",
                        "expect": ["prefect.flow-run.Completed"],
                        "match_related": {
                            "prefect.resource.name": "part-b",
                            "prefect.resource.role": "deployment",
                        },
                    },
                ],
            )
        ],
    )

    serve(dep_a, dep_b, dep_final)

Notes: - Set the window (

within

) large enough to cover the gap between the slowest and fastest schedules. - If you need the final run to pair runs by a specific partition/date, schedule each part deployment with the same parameter (e.g.,

run_date

) and we can template the final trigger to pass or match that parameter. If you want that, tell me the parameter you use and I’ll show the exact trigger config. - If parts are producing files/assets, you can also trigger the final flow off asset materialization events instead of flow-completion events. Can you share: - How many parts and their schedules? - Do they share a common partition key (e.g., a date) that final should use to join them? - Where the parts are persisted (S3/filesystem/DB) so I can tailor the combine/insert step?

Short answer: subflows are great for organizing logic within a single run, but they’re not the right tool when each part needs its own schedule. For different schedules and dependencies, use separate flows with their own deployments and coordinate them with event-based triggers. Why: - Calling a subflow from a parent runs it when the parent runs (no independent schedule). - If parts arrive hours apart, a parent + subflows design forces one long-running run. - Deployments let each part run on its own schedule; the final deployment can be triggered only after all parts complete. Recommended pattern: - One flow per part → one deployment per flow, each with its own schedule. - A final “combine + insert” flow → a deployment with a compound trigger that requires completion events from all part deployments within a window. Concise example:

from datetime import timedelta
from prefect import flow, serve
from prefect.events import DeploymentCompoundTrigger

@flow
def part_a(): ...
@flow
def part_b(): ...
@flow
def final_insert(): ...

dep_a = part_a.to_deployment(name="part-a", cron="0 9 * * *")
dep_b = part_b.to_deployment(name="part-b", interval=3600)

dep_final = final_insert.to_deployment(
    name="final-insert",
    triggers=[
        DeploymentCompoundTrigger(
            name="wait-for-both",
            enabled=True,
            require="all",
            within=86400,
            triggers=[
                {
                    "type": "event",
                    "posture": "Reactive",
                    "expect": ["prefect.flow-run.Completed"],
                    "match_related": {
                        "prefect.resource.name": "part-a",
                        "prefect.resource.role": "deployment",
                    },
                },
                {
                    "type": "event",
                    "posture": "Reactive",
                    "expect": ["prefect.flow-run.Completed"],
                    "match_related": {
                        "prefect.resource.name": "part-b",
                        "prefect.resource.role": "deployment",
                    },
                },
            ],
        )
    ],
)

serve(dep_a, dep_b, dep_final)

If you prefer to coordinate by data instead of runs, you can have each part materialize an asset and trigger the final flow off asset materialization events instead. Docs: - Automations and event triggers: Automations (events) - Assets and materializations: Assets If you share how many parts you have, their schedules, and whether you join by a partition/date, I can sketch the exact trigger configuration (including pairing runs by the same date).

Short answer: it’s not the version warning. Your flow is hanging because you’re wrapping Prefect’s async Slack task with

asyncio.run(...)

inside a running Prefect flow/task. What’s happening -

prefect_slack.messages.send_chat_message

is a Prefect task (async). When you call it from within a flow/task, Prefect manages its event loop and state. - Wrapping it with

asyncio.run(...)

inside the flow can deadlock or stall, especially when there’s already an event loop and task context. Your log shows Prefect started the task (“Task run 'send_chat_message-…' - Posting chat message...”) and then stalls. The version mismatch warning - “Your Prefect server is running an older version…” is a warning only. It does not stop execution. Still, you should upgrade your server to match your client (3.4.14+) to avoid edge-case issues. How to fix Remove the

asyncio.run

wrapper and let Prefect run the Slack task directly. Two clean options: Option A: Call the Prefect Slack task directly (blocking)

from prefect_slack.credentials import SlackCredentials
from prefect_slack.messages import send_chat_message

def send_slack_notification(message: str):
    creds = SlackCredentials(token=SLACK_BOT_TOKEN)
    # This will run as a Prefect task within the current flow/task context
    return send_chat_message(
        slack_credentials=creds,
        channel=SLACK_FAILED_CHANNEL,
        text=f">{message}\n`Source Workflow: stock_eod_stats_daily_download`",
    )

Option B: Submit it so it runs asynchronously and doesn’t block your task

from prefect_slack.credentials import SlackCredentials
from prefect_slack.messages import send_chat_message

def send_slack_notification(message: str):
    creds = SlackCredentials(token=SLACK_BOT_TOKEN)
    return send_chat_message.submit(
        slack_credentials=creds,
        channel=SLACK_FAILED_CHANNEL,
        text=f">{message}\n`Source Workflow: stock_eod_stats_daily_download`",
    )

Then, in your

download_eod_md_fields

task, notify and re-raise:

@task
def download_eod_md_fields(trade_date, stock_lists):
    try:
        # ...do work...
        raise Exception("Simulated failure after notifying")
    except Exception as e:
        send_slack_notification(f"download_eod_md_fields failed for {trade_date}: {e}")
        raise

A couple of extra notes - If you need a sync wrapper outside of Prefect (e.g., a plain Python script),

asyncio.run

is fine. Just don’t use it inside Prefect flows/tasks. - If you expect the flow to fail when

download_eod_md_fields

fails, make sure you propagate the exception (as above). Using

.submit(...)

is fine; downstream tasks that depend on it will be cancelled, and the flow will finish once all futures settle. - Strongly recommended: upgrade your server from 3.1.6.dev9 to 3.4.14+ to match your client. References - Prefect Slack collection: prefect-slack docs - Prefect docs (3.x): Prefect 3.x docs If this change doesn’t unstick the run, can you share: - Your Prefect versions (client/worker/server) - Where

send_slack_notification

is called from (flow vs task), and your task runner setup - Any additional logs after “Posting chat message…” (timeouts, network/proxy)

Good data point — thanks. The fact that your test flow runs fine suggests the hang you saw earlier isn’t caused by the version warning, but by differences between how you’re sending Slack messages in the two cases and/or the event loop on Windows. Key differences I see - In your test, you set the Windows event loop policy:

if platform.system() == 'Windows':
      asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

If your “stuck” flow runs on Windows without this, an asyncio interaction can stall. Prefect’s engine is async; mixing

asyncio.run(...)

inside it can be especially brittle on Windows unless you set this policy. - Your test imports

send_slack_message

from

phoenix.notify.slack_notifier

. That function may not be the same implementation you used in the “stuck” flow. If your production wrapper calls

prefect_slack.messages.send_chat_message

via

asyncio.run(...)

, it can hang when called from inside a Prefect task/flow. In the test, your notifier might be using the Slack SDK synchronously or otherwise avoiding the nested-event-loop problem. What I recommend 1) In your production flow, don’t wrap Prefect’s Slack task with

asyncio.run(...)

. Call the Prefect task directly or submit it:

from prefect_slack.credentials import SlackCredentials
from prefect_slack.messages import send_chat_message

def send_slack_notification(message: str):
    creds = SlackCredentials(token=SLACK_BOT_TOKEN)
    # Option A: run inline (blocking in task context)
    return send_chat_message(
        slack_credentials=creds,
        channel=SLACK_FAILED_CHANNEL,
        text=f">{message}\n`Source Workflow: stock_eod_stats_daily_download`",
    )
    # Option B: fire-and-forget in background
    # return send_chat_message.submit(...same args...)

2) If you must keep a synchronous wrapper for Slack, avoid

asyncio.run(...)

inside Prefect code paths. Either: - Use the Slack SDK directly in a normal sync function (no Prefect task), or - Run the async call from a separate thread to avoid a nested loop:

import anyio

  def send_slack_message_sync(...):
      def _run():
          asyncio.run(send_chat_message(...))  # your async call
      return anyio.run_sync_in_worker_thread(_run)

3) On Windows, keep the WindowsSelectorEventLoopPolicy at the entrypoint (as in your test) if your production environment is Windows:

if platform.system() == 'Windows':
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

4) It’s still a good idea to upgrade your Prefect server from 3.1.6.dev9 to 3.4.14+ to match your client. The mismatch is a warning only, but matching versions avoids edge-case behaviors. 5) Please rotate the Slack bot token you pasted publicly. Treat it as compromised. To help pinpoint the delta, could you share: - The implementation of

phoenix.notify.slack_notifier.send_slack_message

used in your production flow vs the test flow? - OS/Python versions for the environment where the hang occurs - Whether adding the WindowsSelectorEventLoopPolicy in your production entrypoint resolves the hang With that, I can give you an exact, minimal change for your “stuck” flow.

Yes — they’re different things: - Work pool: A logical resource in the Prefect API that holds configuration and serves as the queue a worker polls. You don’t “restart” a work pool; you can pause/resume it or update its settings. - Useful commands: - Pause:

prefect work-pool pause <pool-name>

- Resume:

prefect work-pool resume <pool-name>

- Inspect:

prefect work-pool inspect <pool-name>

- Worker: A process you run that polls a work pool for flow runs and executes them on some infrastructure. Restarting a worker means stopping and starting that process. - Start:

prefect worker start --pool <pool-name> [--name <worker-name>] [--work-queue <queue>]

So: - Restarting a worker = restarting a process on your machine/cluster. - “Restarting” a work pool isn’t a thing; you manage its state (pause/resume) or config via the API/CLI.