https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> - are there any provisions in prefect...
# ask-marvin
k
@Marvin - are there any provisions in prefect v3 for managing shared but controlled resources? I seem to recall some sort of context manager thing in prefect v1. I want to simplify the creation and sharing of a DB connection pool between tasks?
m
✅ thought for 124.4 seconds
Short answer: not in the old Prefect 1 “resource/context” sense. In Prefect 3, there isn’t a built-in “shared resource manager” primitive for things like “create one DB pool and automatically inject/share it across tasks.” The main built-ins nearby are: - Blocks for storing configuration/credentials, not live in-memory resources - Concurrency limits for controlling access rates / parallelism, not sharing objects - normal Python context managers inside flows/tasks So for a DB connection pool, the recommended pattern is usually: 1. Store connection config in a Block if you want central config/credentials 2. Create the pool in flow scope 3. Pass it to tasks only if those tasks run in the same process/thread model and the client is safe to share 4. Otherwise, create a client/pool per task and use Prefect limits to control contention A few important caveats: - If your tasks are executed in separate infrastructure/processes, you cannot share an in-memory pool between them. - Even in one flow run, tasks may run concurrently, so the pool/client must be thread-safe / async-safe for your task runner and DB library. - Prefect won’t manage the lifecycle of that pool for you beyond normal Python code. A simple pattern looks like this:
Copy code
python
from prefect import flow, task

class DBPool:
    def __init__(self, dsn: str):
        self.dsn = dsn
        self._pool = None

    async def open(self):
        # create underlying pool here
        self._pool = ...
        return self

    async def close(self):
        # close underlying pool here
        ...

@task
async def use_db(pool: DBPool, item: int):
    # use shared pool here
    ...

@flow
async def my_flow():
    pool = await DBPool("postgresql://...").open()
    try:
        await use_db.submit(pool, 1)
        await use_db.submit(pool, 2)
    finally:
        await pool.close()
That said, this is only appropriate if the tasks truly share memory. If your tasks are not colocated, do this instead:
Copy code
python
from prefect import flow, task

@task
async def use_db(dsn: str, item: int):
    pool = ...
    try:
        ...
    finally:
        await pool.close()

@flow
async def my_flow():
    await use_db.submit("postgresql://...", 1)
    await use_db.submit("postgresql://...", 2)
If your actual goal is controlled access to the DB rather than literal object sharing, Prefect’s concurrency controls are often the better fit than a shared pool. Those let you cap how many tasks hit the DB at once. Relevant docs: - Blocks - Global concurrency limits - Task runners So: yes for controlled access, no for a first-class shared resource abstraction. If you want, I can show the best pattern for your specific DB stack (for example 
asyncpg
, SQLAlchemy, psycopg3, Snowflake, etc.), since the right answer depends a lot on whether your tasks are async and where they run.
k
@Marvin - 
Pass it to tasks only if those tasks run in the same process/thread model and the client is safe to share
is there no way to bind it to the flow run context or something so I don't have to add an argument to every single task?
m
✅ thought for 145.3 seconds
Yes, you can do something like that, but with an important catch: You can bind a live object to a custom Prefect context, but it only works when tasks execute in the same Python process. It is not a general flow-run-scoped resource mechanism that works across workers/processes/infrastructure boundaries. So the answer is: - Same-process / thread-based execution: yes, a custom context can avoid passing the pool argument everywhere - Distributed or separate-process execution: no, because the live pool object cannot be propagated via Prefect context A sketch of the same-process pattern:
Copy code
python
from contextvars import ContextVar

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

class DBContext(ContextModel):
    __var__ = ContextVar("db_context")
    pool: object

@task
def task_a():
    pool = DBContext.get().pool
    # use pool here
    ...

@task
def task_b():
    pool = DBContext.get().pool
    # use pool here
    ...

@flow(task_runner=ThreadPoolTaskRunner())
def my_flow():
    pool = create_pool()
    try:
        with DBContext(pool=pool):
            task_a.submit()
            task_b.submit()
    finally:
        close_pool(pool)
Why I’d be cautious recommending this: - it couples your tasks to ambient runtime state - it only works for same-process runners like 
ThreadPoolTaskRunner
- it will not work reliably if tasks run in another process / worker / container - it’s less explicit, so testing and reuse get harder So if you want ergonomics without passing the pool into every task, there are two better patterns depending on your setup: 1. Use a helper function instead of task args Tasks call a shared accessor that reads from custom context when available.
Copy code
python
def get_pool():
    ctx = DBContext.get()
    if ctx is None:
        raise RuntimeError("No DBContext is active")
    return ctx.pool

@task
def task_a():
    pool = get_pool()
    ...
That cleans up signatures, but still has the same same-process limitation. 2. Pass config, not the pool If your tasks may ever leave the process boundary, pass a DSN / block / settings object and lazily create connections in task code. That’s the most portable pattern. Also relevant if your main concern is controlled usage rather than literal sharing: use concurrency limits to cap simultaneous DB access: - Global concurrency limits Useful references: - Task runners - Blocks So: yes, with custom 
ContextModel
, but only as a local same-process convenience hack/pattern — not as a universal Prefect resource binding feature. If you want, I can show you a clean production-safe pattern for: - same-process shared SQLAlchemy engine/sessionmaker, or - distributed-safe per-task DB access with minimal boilerplate
Open in Slack
PreviousNext
Powered by