MongoDB

Overview

Singularity ships an optional MongoDB extension built on the two pieces the MongoDB team officially recommends for FastAPI in 2026:

The extension is gated behind the [mongo] extra and a pair of environment variables, so projects that don't use MongoDB pay zero install cost and zero startup cost.

Installation

pip install singularity-fm[mongo]

Then set both env vars in .env:

MONGO_URL=mongodb://localhost:27017
MONGO_DB_NAME=my_project

If either is missing, Singularity logs Mongo: skipped (...) at startup and acquire.mongo resolves to None. The framework never silently picks defaults for Mongo — that mistake is too easy to debug at 3am.

Defining a Document

# models/user.py
from beanie import Indexed

from singularity.db.mongo import TimestampedDocument


class User(TimestampedDocument):
    email: Indexed(str, unique=True)
    name: str

    class Settings:
        name = "users"  # collection name; defaults to class name lowercased

TimestampedDocument is a thin wrapper around beanie.Document that adds created_at / updated_at fields and refreshes updated_at on save_changes() and replace(). If you don't want timestamps, inherit from singularity.db.mongo.Document directly.

Auto-discovery

At startup, Singularity walks models/*.py and registers every subclass of beanie.Document it finds with init_beanie. You do not need to maintain a list of documents anywhere — drop the file in models/ and it's live.

Using It in a Service

Acquire exposes the configured AsyncDatabase as self.acquire.mongo. You can use it directly for raw queries, or use the Beanie API on your documents.

# services/users/service.py
from models.user import User


class UsersService:
    def __init__(self, acquire):
        self.acquire = acquire

    async def create(self, *, email: str, name: str) -> User:
        user = User(email=email, name=name)
        await user.insert()
        return user

    async def by_email(self, email: str) -> User | None:
        return await User.find_one(User.email == email)

    async def raw_count(self) -> int:
        # When you need to drop down to the raw driver:
        return await self.acquire.mongo.users.count_documents({})

Lifecycle

The connection pool is opened once at startup and closed on shutdown. AsyncMongoClient is thread/loop-safe and meant to be shared across the entire process — never instantiate one inside a request handler.

Error Handling

pymongo.errors.PyMongoError is automatically converted to singularity.MongoError by handle_exception(), so the same BaseCustomError machinery your SQL services use applies to Mongo services too:

from singularity import MongoError

try:
    await User.find_one(User.email == email)
except MongoError as exc:
    # exc.hint, exc.details, exc.status_code all populated
    ...

Using [sql] and [mongo] Together

The two extras are independent. A polyglot service is just:

pip install "singularity-fm[sql,mongo]"

self.acquire.db_session and self.acquire.mongo coexist; pick the right store per aggregate.

Reference

All symbols are also re-exported at the top level (e.g. from singularity import Document, get_mongo_client) via a lazy __getattr__, so you can pick whichever import style fits your codebase.