Multiple databases

[ph55]

Hi,

Does beanie supports working with multiple databases ?
I see that following example working, but is it considered "good practice" ?

        await init_beanie(database=client.db_one, document_models=[DbOneModelX, DbOneModelY])
        await init_beanie(database=client.db_two, document_models=[DbTwoModelX])

[roman-right]

Hi,
There is no "good practice" option for this now. I'm working on a new initializer atm - plan to add db swaps there.

[mg3146]

Thanks for this - I'd be keen for db swaps as well (a test, and prod)

[rollebolle123]

Hi!
But this would not work with async calls?
Consider a fastapi async endpoint, which in turn does other calls, for example HTTP calls (Which may or may not be executed, take different time to run etc).
I tried to simulate this with an endpoint with configurable sleep:

# concept
client = AsyncIOMotorClient(database_url)
@a_router.get("/", response_model=SomethingCool)
async def get_something_cool(sleep_time: int, db: str):
await init_beanie(db)
await asyncio.sleep(sleep_time)
await SomethingCool.find()

If the endpoint is being called multiple times from different consumers, the global DB instance would be updated, fetching the wrong documents for the wrong consumer.

Even if init_beanie runs precisely before every call to example fetch documents, it could still end up fetching the wrong documents? :
@a_router.get("/", response_model=SomethingCool)
async def get_something_cool(sleep_time: int, db: str):
await init_beanie(db)
await SomethingCool.find()
await init_beanie(db)
await SomethingElse.find()

[treylade]

Is there any update on the db swap feature you talked about? I would like to build a multi-tenant architecture using Beanie (actually Bunnet) using separate databases per tenant but the same Document classes.

At the moment I call init_bunnet on the current tenant db for every new request made in a Flask environment. This works. But I suppose there might be a better off the shelf solution.

[farzadsbakhtiar]

@rollebolle123 did you confirm that this can run into problems?

[rollebolle123]

Well, I decided to change to a different library called odmantic.

[Vwake04]

In the provided snippet, all instances of XX share the same document class, causing the connection session to be reinitialized each time init_beanie is called.

import asyncio
from motor.motor_asyncio import AsyncIOMotorClient
from beanie import init_beanie, Document


class XX(Document):
    v: int
    db_name: str = "test"


    class Settings:
        collection_name = "XX"


async def init_db(db_name):
    client = AsyncIOMotorClient("mongodb://localhost:27017")
    await init_beanie(database=client[db_name], document_models=[XX])
    
    xs: Document = XX(v=1, db_name=db_name)
    await xs.insert()


async def main():
    f = [
        init_db("abc#1"),
        init_db("abc#2"),
        init_db("abc#3"),
        init_db("abc#4"),
        init_db("abc#5"),
        init_db("abc#6"),
    ]
    await asyncio.gather(*f)


if __name__ == "__main__":
    asyncio.run(main())

[Yalchin403]

@roman-right I'm using multitenant-architecture and this setup you have provided is working fine, unless we have a concurrent requests, to reproduce it, you can set up two simple endpoints (FastAPI) that deal with 2 different databases, then you can try to set up postman runner to make concurrent requests. In this case we endpoint with the following error:

raise InvalidOperation("Can only use session with the MongoClient that started it")

Additionally calling init_beanie for every switch causes performance issues

[ajordanb]

@roman-right any updates on this?

[frenzyk]

I suggest Django approach could be good for adoption. I mean DB mapping and use .using('db_map_name') in queries.

[sbatdkio]

I also have multitenant setup, my solution to handle it was to create some beanie document class cache (one entry per tenant DB/Document class)
The global principle is here.
Then all my methods that want to use a beanie document are calling this method to retrieve a "specific DB bound" beanie document class

 @classmethod
    async def _init_beanie_with_document(
        cls,
        database_name: str,
        document_class: type[Document],
        retries=0,
        skip_indexes=False,
    ) -> type[Document]:
        """
        Initialize Beanie with a document class for a specific database.  Uses a cache with expiration and locking to avoid redundant initializations.

        Args:
        database_name: database name
        document_class: document class to initialize
        retries: Number of attempts already made (for recursion)

        Returns:
        The initialized document class
        """
        doc_class_name = document_class.__name__
        connection_cache = ConnectionCache.get_instance()

        # Check if the connection is in the cache
        cached_class = await connection_cache.get(database_name, doc_class_name)
        if cached_class is not None:
            return cached_class

        # Acquire a lock to avoid concurrent initialization.
        lock = await connection_cache.get_lock(database_name, doc_class_name)
        async with lock:
            # Vérifier à nouveau après avoir acquis le verrou (double-check locking pattern)
            cached_class = await connection_cache.get(database_name, doc_class_name)
            if cached_class is not None:
                return cached_class

            # Initialize the Beanie connection
            try:
                logger.debug(
                    f"#######################  INIT BEANIE WITH DOCUMENT ->>>>> {database_name,doc_class_name}",
                )
                await init_beanie(
                    database=client.get_database(database_name),
                    document_models=[document_class],
                    multiprocessing_mode=False,
                    skip_indexes=skip_indexes,
                )

                motor = document_class.get_motor_collection()
                if motor is None:
                    if retries == 5:
                        logger.warning(
                            f"Init failure after 5 tries for {database_name}.{doc_class_name}"
                        )
                        return document_class

                    logger.debug(
                        f"MOTOR EMPTY pour db: {database_name} - document: {doc_class_name}, réessai {retries + 1}"
                    )
                    await asyncio.sleep(0.05)
                    initialized_beanie = await cls._init_beanie_with_document(
                        database_name, document_class, retries=retries + 1
                    )
                    return initialized_beanie

                # Cache the connection
                if not skip_indexes:
                    await connection_cache.set(
                        database_name, doc_class_name, document_class
                    )
                return document_class

            except Exception as e:
                logger.error(
                    f"Beanie Initialization error for {database_name}.{doc_class_name}: {e}"
                )
                raise
       `

[tfglasswing]

Having the same feature request. In our case we are "polluting" one client database by another which is critical.

A simple example of the issue looks like this:

async def v1_mongo_delayed():
  await init_beanie(database=client.db_one, document_models=[SomeDocument])
  await asyncio.sleep(1)
  SomeDocument.insert()

async def v1_mongo_fast():
  await init_beanie(database=client.db_two, document_models=[SomeDocument])
  SomeDocument.insert()

calling both endpoints will write both documents to "client.db_two".

[farzadsbakhtiar]

Not sure about the state of this. My setting is like the OP, i.e. n-to-1 mapping from documents to database clients.

await init_beanie(database=clientOne, document_models=[docOne])
await init_beanie(database=clientTwo, document_models=[docTwo])

Things seem to be working fine. I only call init_beanie once for each database (and associated docs) at startup. Will there be a problem for simultaneous requests?

[tfglasswing]

yeah, but we did not intend to ..

[farzadsbakhtiar]

not sure what you mean

[farzadsbakhtiar]

what i see is you want to persist one document to two dbs and that's not working. My goal is different

[farzadsbakhtiar]

@roman-right could you please pitch in about whether this will be a problem in rollebolle123's scenario above?
I'm using multiple DB clients for multiple DBs. But each document model is only persisted to one DB.

[farzadsbakhtiar]

UPDATE: I took a look into the code to see if this could be a problem. It wouldn't. The db connection is set on a class level property on the document class. As long as you're not doing weird subclassing things with your documents and persisting parent and child classes to different dbs, you should be fine. You could be fine even if you are doing weird stuff.

[QSHolzner]

Hi everyone,

we've been dealing with the same multi-tenant challenge (FastAPI + Beanie,
database-per-tenant isolation) and wanted to share an approach we've been
running in production testing. We'd appreciate any feedback on whether
this holds up or if we're missing something important.

We noticed that @roman-right mentioned a new initializer with DB switch
support. We're not sure if that's still on the roadmap.
If not, the pattern below might help others who are dealing with
the same problem in the meantime.

What we found in the source code

Looking at beanie/odm/interfaces/getters.py, all collection access in
Beanie flows through a single method chain:

# beanie/odm/interfaces/getters.py (OtherGettersInterface)
@classmethod
def get_pymongo_collection(cls) -> AsyncCollection:
    return cls.get_settings().pymongo_collection

Every Beanie operation -- find, insert, save, replace, delete,
aggregate -- ultimately calls get_pymongo_collection(), which goes
through get_settings(). We couldn't find any code path that bypasses
this and accesses _document_settings directly.

This means get_settings() is the single point of control for all
database access in Beanie. Overriding this one method on a base Document
class is enough to redirect all operations to a different database.

The implementation

Override get_settings() on a base Document class. Use a
contextvars.ContextVar to carry the active tenant per request.
Return a copy of the original settings with pymongo_collection
and pymongo_db pointing to the target database. Cache the copies
per (class, tenant) so they are only created once per combination.

from __future__ import annotations

from contextvars import ContextVar, Token
from contextlib import contextmanager
from typing import Any, ClassVar

from beanie import Document
from motor.motor_asyncio import AsyncIOMotorDatabase

# -- Tenant context --------------------------------------------------------

_current_tenant: ContextVar[str | None] = ContextVar(
    "current_tenant", default=None
)

@contextmanager
def tenant_context(tenant_id: str):
    """Set the active tenant for the current async context."""
    token: Token = _current_tenant.set(tenant_id)
    try:
        yield
    finally:
        _current_tenant.reset(token)  # restores previous value, not None

def get_current_tenant() -> str | None:
    return _current_tenant.get()

# -- Tenant DB registry ----------------------------------------------------

_tenant_dbs: dict[str, AsyncIOMotorDatabase] = {}

def register_tenant_db(tenant_id: str, db: AsyncIOMotorDatabase) -> None:
    if not isinstance(db, AsyncIOMotorDatabase):
        raise TypeError(
            f"Expected AsyncIOMotorDatabase, got {type(db).__name__}"
        )
    _tenant_dbs[tenant_id] = db

def get_tenant_db(tenant_id: str) -> AsyncIOMotorDatabase:
    try:
        return _tenant_dbs[tenant_id]
    except KeyError:
        raise ValueError(f"Unknown tenant: {tenant_id!r}")

# -- Settings cache --------------------------------------------------------

_settings_cache: dict[tuple, Any] = {}

def clear_settings_cache() -> None:
    _settings_cache.clear()

# -- Base document ---------------------------------------------------------

class TenantDocument(Document):
    """
    Base class for all documents that should be tenant-aware.
    Inherit from this instead of Document.
    """

    class Settings:
        is_root = True   # so Beanie doesn't treat subclasses as union types

    @classmethod
    def get_settings(cls):
        tenant = get_current_tenant()
        if tenant is None:
            return super().get_settings()

        cache_key = (cls, tenant)    # cls itself, not cls.__name__
        cached = _settings_cache.get(cache_key)
        if cached is not None:
            return cached

        original = super().get_settings()
        db = get_tenant_db(tenant)

        settings = original.model_copy()
        settings.pymongo_collection = db[original.name]
        settings.pymongo_db = db

        _settings_cache[cache_key] = settings
        return settings

init_beanie() runs exactly once at startup with a default database.
After that, additional databases are simply registered in the dict.
No further init_beanie() calls needed.

Usage with FastAPI

Tenant detection happens in raw ASGI middleware (not Starlette's
BaseHTTPMiddleware, which runs the downstream app in a separate
task and can cause subtle issues with ContextVar propagation in
certain scenarios). The middleware reads the tenant ID from a
JWT claim and sets the context:

class TenantMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] not in ("http", "websocket"):
            await self.app(scope, receive, send)
            return

        tenant_id = extract_tenant_from_jwt(scope)  # your JWT logic
        if tenant_id:
            with tenant_context(tenant_id):
                await self.app(scope, receive, send)
        else:
            await self.app(scope, receive, send)

What we've tested

We tried this in a FastAPI application where the tenant ID is carried in
a JWT claim. Write isolation, read isolation, parallel async requests --
everything worked as expected so far. Documents land in the correct
tenant database, the default database stays empty, and we couldn't
produce any cross-tenant data leaks in our tests.

Known limitation: indexes

Since init_beanie() only runs once on the default database, the other
databases don't get the indexes that init_beanie() would normally
create. Queries still work (MongoDB creates collections on first insert),
but without the expected indexes, which could become a performance issue
as data grows -- and unique constraints would be missing entirely.

Our workaround is to create indexes when registering a new tenant database:

async def ensure_tenant_indexes(
    document_models: list[type[TenantDocument]],
    tenant_id: str,
) -> None:
    """Create indexes on a tenant database for all document models."""
    with tenant_context(tenant_id):
        for model in document_models:
            try:
                await model._ensure_indexes()
            except AttributeError:
                # Method name may differ across Beanie versions
                await model.ensure_indexes()

This runs once per tenant at registration time, not per request.

We're wondering if there's a cleaner way -- maybe by calling parts
of the Beanie initializer selectively. Running init_beanie() once per
database at startup would overwrite the settings from the first call,
so that's not an option.

Feedback welcome

We've tested this extensively and it works well so far, but we're sure
there are edge cases and Beanie internals we haven't considered. If anyone
spots something that could break, we'd really appreciate a heads-up:

1. Are there Beanie operations that bypass get_settings()?
   We couldn't find any code path that accesses _document_settings
   directly, but we may have missed something.

2. Is model_copy() on ItemSettings safe?
   We only replace pymongo_collection and pymongo_db. The remaining
   fields (indexes, bson_encoders, etc.) are shared between original
   and copy but are only ever read, never mutated at runtime. If Beanie
   ever mutates these after init, this could break.

3. How stable is the get_settings() -> get_pymongo_collection()
   chain across Beanie versions?
   This is a public API, but we couldn't find a guarantee that it won't
   change. If it does, this approach breaks silently.

4. ContextVar propagation in edge cases?
   asyncio.create_task() copies the current context (documented
   behavior since Python 3.7). But for run_in_executor(), context is
   only propagated in Python 3.12+. We don't use sync executors, but
   others might.

5. Is there a safe way to run index creation selectively?
   Calling _ensure_indexes() works in our Beanie version but it's
   a private method. A public API for "create indexes for this model
   on this database" would be very helpful.

Any other thoughts or "have you thought about X?" equally welcome.

Related: #249, #1072