PyMongo With FastAPI

Introduction

FastAPI excels at JSON APIs; MongoDB stores flexible documents— a natural pair for content, catalogs, and event APIs. This chapter wires PyMongo through lifespan, dependency injection, Pydantic schemas, and a small posts CRUD router—following patterns from FastAPI project structure and PyMongo basics.

Prerequisites

Project Layout

requirements.txt:

text
fastapi[standard]>=0.115.0
pymongo[srv]>=4.6
pydantic-settings>=2.0
python-dotenv>=1.0

Settings

app/core/config.py:

.env:

env
MONGODB_URI=mongodb://127.0.0.1:27017
MONGODB_DB_NAME=hello_demo

Atlas:

env
MONGODB_URI=mongodb+srv://app_user:PASSWORD@cluster0.xxxxx.mongodb.net/?retryWrites=true&w=majority

MongoClient Lifespan

app/db/mongo.py:

app/main.py:

Code explanation:

  • init_mongo / close_mongo mirror Redis lifespan
  • ping on startup fails fast if DB is down

Pydantic Schemas

app/schemas/post.py:

Document ↔ Schema Helpers

app/api/v1/endpoints/posts.py (top):

Ensure index once at startup (optional in init_mongo):

python
get_posts_collection().create_index("slug", unique=True)

CRUD Endpoints

Code explanation:

  • Depends(get_collection) injects the same collection per request
  • Pydantic validates input—prevents NoSQL injection via typed fields — Security
  • 409 on duplicate slug from unique index

Run and Test

bash
uvicorn app.main:app --reload
bash
curl -s http://127.0.0.1:8000/health
curl -s -X POST http://127.0.0.1:8000/api/v1/posts/ \
  -H "Content-Type: application/json" \
  -d '{"title":"API Post","slug":"api-post","published":true,"body":"Hello"}'
curl -s "http://127.0.0.1:8000/api/v1/posts/?published=true"

Open /docs for Swagger UI — OpenAPI chapter.

MongoDB + SQLAlchemy Split (Architecture Note)

Some products use MySQL/PostgreSQL for billing (ACID, reports) and MongoDB for content or logs. Two connections in lifespan:

python
# Concept only — two resources in lifespan
init_mongo()
init_sqlalchemy_engine()

Do not duplicate transactional data in both without a sync strategy.

Optional: Redis Cache Layer

After get_post, add Redis cache-aside — Performance and FastAPI Redis.

Testing With TestClient

Use mongomock or a disposable Docker MongoDB in CI:

python
# pytest fixture sketch — override get_collection to test DB
@pytest.fixture
def client():
    os.environ["MONGODB_DB_NAME"] = "hello_demo_test"
    with TestClient(app) as c:
        yield c
    get_database().client.drop_database("hello_demo_test")

See FastAPI testing.

Post-Chapter Checklist

  • MongoClient initialized in lifespan
  • Posts CRUD works via curl or /docs
  • ObjectId exposed as string in JSON
  • Unique slug index returns 409 on conflict

FAQ

Async routes with PyMongo?

Blocking PyMongo in async def blocks the event loop—use def routes or Motor — chapter 21.

Beanie / ODM?

Beanie adds async document models; this track stays driver-level for clarity.

JWT auth on writes?

Add Depends(get_current_user) from FastAPI JWT.

CORS for SPA?

CORSMiddleware — FastAPI middleware.

Deploy?

Docker Compose with mongo + api — Project deploy.