The Python equivalent of Mongoose in 2026 is Beanie — an async ODM built on Motor + Pydantic. It fits naturally into FastAPI (which is also Pydantic-based) and gives type-safe models, async I/O, validation, indexes, hooks, links (relations), aggregations, and migrations in one package. This post covers the working set: models, CRUD, indexes, links + back-links, hooks, projections, aggregations, transactions, integration patterns with FastAPI.

Series map: 17 base course → 18 Mongoose deep → 19 Beanie + FastAPI (this post) → 20 Outbox → ... → 29 K8s + mongosync + cost.

Learning Outcome

Set up Beanie + Motor with FastAPI lifespan.
Define models with Pydantic v2 + Indexed + relations (Link).
CRUD with type-safe queries.
Use projections + aggregations + bulk writes.
Hook into events (before_save, after_event).
Use transactions inside Beanie.
Integrate with FastAPI dependency injection.

Chapter 1 — Install + connect

1
2 pip install beanie fastapi uvicorn[standard] pydantic

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17 # main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from motor.motor_asyncio import AsyncIOMotorClient
from beanie import init_beanie
 
from .models import User, Order, Product
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    client = AsyncIOMotorClient(os.environ["MONGO_URI"], maxPoolSize=50, appName="shop-api")
    await init_beanie(database=client.shop, document_models=[User, Order, Product])
    yield
    client.close()
 
app = FastAPI(lifespan=lifespan)

init_beanie registers your models against the database and creates declared indexes. One call per process.

Chapter 2 — Models

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31 # models.py
from datetime import datetime
from typing import Annotated, Optional
from beanie import Document, Indexed, before_event, Replace, Insert
from pydantic import BaseModel, EmailStr, Field
 
class Address(BaseModel):
    city: str
    country: str = "UK"
 
class User(Document):
    name: str
    email: Annotated[EmailStr, Indexed(unique=True)]
    age: Optional[int] = Field(default=None, ge=0, le=150)
    role: str = "user"
    tags: list[str] = []
    address: Optional[Address] = None
    created_at: datetime = Field(default_factory=datetime.utcnow)
 
    class Settings:
        name = "users"                                  # collection name
        use_state_management = True                     # for Save / Replace
        indexes = [
            "created_at",
            [("name", 1), ("created_at", -1)],          # compound
        ]
 
    @before_event(Insert)
    async def normalise(self):
        self.email = self.email.lower()

Key elements:

Inherit Document. _id auto-managed as ObjectId.
Annotated[Type, Indexed(...)] for index declarations.
Settings inner class for collection name + index definitions.
@before_event / @after_event hooks similar to Mongoose middleware.
Pydantic v2 validators (Field(ge=0)) run on construction + assignment.

Chapter 3 — CRUD

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 # create
u = User(name="Ada", email="ada@x")
await u.insert()
 
# read
u = await User.get(some_id)
u = await User.find_one(User.email == "ada@x")
users = await User.find(User.age > 25).sort(-User.created_at).limit(20).to_list()
 
# update
u.age = 31
await u.save()
# or atomic via update operators
await User.find_one(User.email == "ada@x").update({"$inc": {"login_count": 1}})
 
# delete
await u.delete()
await User.find(User.role == "user", User.age < 18).delete()

Query expressions use the Python operators on model fields — fully type-checked.

Chapter 4 — Indexes

1
2
3
4
5
6
7
8
9
10
11
12 class Product(Document):
    name: str
    price: Annotated[float, Indexed()]
    category: Annotated[str, Indexed()]
 
    class Settings:
        indexes = [
            [("category", 1), ("price", 1)],                       # compound
            IndexModel([("name", "text")], default_language="english"),  # text
            IndexModel([("location", "2dsphere")]),                # geo
        ]

Indexes created at init_beanie. For production, disable + create via migrations (same advice as Mongoose Post 18).

Chapter 5 — Links (relations)

Beanie supports linked documents — equivalent to Mongoose populate:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 from beanie import Document, Link
 
class User(Document):
    name: str
 
class Order(Document):
    user: Link[User]      # stored as { id, collection }
    total: float
 
# Insert
ada = User(name="Ada"); await ada.insert()
o   = Order(user=ada, total=99); await o.insert()
 
# Fetch with link
o = await Order.get(order_id, fetch_links=True)
print(o.user.name)        # populated
 
# or one at a time
await o.fetch_link(Order.user)

Multi-level fetch supported. Like Mongoose populate, this issues additional queries — for hot endpoints prefer aggregation $lookup.

BackLink[T] for reverse references (also fetched on demand).

Chapter 6 — Projections

1
2
3
4
5
6
7 class UserProjection(BaseModel):
    name: str
    email: str
 
users = await User.find().project(UserProjection).to_list()
# Returns UserProjection instances (slim), not full User docs

Use for list endpoints — server returns only projected fields + Pydantic constructs the typed slim object.

Chapter 7 — Aggregations

1
2
3
4
5
6 result = await User.aggregate([
    {"$match": {"role": "user"}},
    {"$group": {"_id": "$country", "count": {"$sum": 1}}},
    {"$sort": {"count": -1}},
]).to_list()

Returns raw dicts. To get typed results, pass a projection_model:

1
2
3
4
5
6
7
8
9 class CountryCount(BaseModel):
    country_id: str = Field(alias="_id")
    count: int
 
result = await User.aggregate(
    [...],
    projection_model=CountryCount,
).to_list()

For aggregation builders + type-safe pipeline construction, the community pyodmongo and helpers exist. Most teams just use dict pipelines as in mongosh.

Chapter 8 — Hooks (events)

1
2
3
4
5
6
7
8
9
10
11
12
13 from beanie import Document, before_event, after_event, Insert, Save, Replace, Delete, Validate
 
class User(Document):
    name: str
 
    @before_event(Insert)
    async def normalise(self):
        self.name = self.name.strip()
 
    @after_event(Save, Replace)
    async def publish_event(self):
        await publish({"type": "user.saved", "id": str(self.id)})

Events: Insert, Replace, Update, Delete, Save, Validate. Multiple events per decorator. Async or sync function bodies.

Chapter 9 — Transactions

1
2
3
4
5
6
7 from beanie import Transaction
 
async with await client.start_session() as session:
    async with session.start_transaction():
        await User.find_one(User.id == 1, session=session).update({"$inc": {"balance": -100}}, session=session)
        await User.find_one(User.id == 2, session=session).update({"$inc": {"balance": 100}}, session=session)

Or with retry helper from Motor (Post 09 patterns apply).

session= must be passed to every op — same rule as Mongoose + the official driver.

Chapter 10 — Bulk writes

1
2
3
4
5
6
7
8 from beanie import BulkWriter
 
async with BulkWriter() as bulk:
    await User(name="A").insert(bulk_writer=bulk)
    await User(name="B").insert(bulk_writer=bulk)
    await User.find_one(User.name == "A").update({"$set": {"role": "admin"}}, bulk_writer=bulk)
# committed on exit

Batches the operations into a single bulkWrite call — much faster than per-doc round-trips.

Chapter 11 — FastAPI integration patterns

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31 from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
 
class UserCreate(BaseModel):
    name: str
    email: EmailStr
 
class UserOut(BaseModel):
    id: str
    name: str
    email: str
 
    @classmethod
    def from_doc(cls, u: User):
        return cls(id=str(u.id), name=u.name, email=u.email)
 
app = FastAPI(lifespan=lifespan)
 
@app.post("/users", response_model=UserOut, status_code=201)
async def create_user(input: UserCreate):
    u = User(**input.model_dump())
    await u.insert()
    return UserOut.from_doc(u)
 
@app.get("/users/{user_id}", response_model=UserOut)
async def get_user(user_id: str):
    u = await User.get(PydanticObjectId(user_id))
    if not u:
        raise HTTPException(404)
    return UserOut.from_doc(u)

Separation of concerns:

UserCreate — request body.
User (Beanie) — database document.
UserOut — response body (no internal fields).

Helps avoid leaking server-side fields (__v, internal_flag, etc.) into responses.

One POST /users request through FastAPI and Beanie, from the request model to the stored document to the response model

Dependency injection for repositories

1
2
3
4
5
6
7
8
9
10
11
12
13 from fastapi import Depends
 
class UserRepo:
    async def by_email(self, email: str):
        return await User.find_one(User.email == email.lower())
 
def repo() -> UserRepo:
    return UserRepo()
 
@app.get("/by-email/{email}")
async def by_email(email: str, repo: UserRepo = Depends(repo)):
    return await repo.by_email(email)

Useful when you want to mock the repository in tests.

Chapter 12 — Migrations

Beanie has beanie migrate CLI:

1
2
3
4 beanie new-migration --name add_email_index
# edit the generated python file with up() / down()
beanie migrate -uri "mongodb://..." -db shop

Migration files apply schema-evolution + index changes idempotently. State stored in migrations_log collection.

For complex schema versioning (Post 07), pair with _schema_v on documents + lazy migration in app code.

Chapter 13 — Common pitfalls

Pitfall	Cause	Fix
`init_beanie` called per request	overhead, leak risk	call once in lifespan
Links populate causing N+1	many docs each `fetch_links`	aggregation `$lookup`
Projection model field name mismatch	aliases needed	use `Field(alias="_id")`
`EmailStr` install error	missing optional dep	`pip install pydantic[email]`
Auto-index blocks on production startup	huge collection	disable + migrate manually
Mixed sync + async code	accidental blocking call	wrap with `asyncio.to_thread`
Old Pydantic v1 examples	breaking changes	always use Pydantic v2
`id` vs `_id` confusion	Beanie exposes `.id` (the str-friendly view)	match docs

Practice Exercises

Exercise 1 — Hello Beanie + FastAPI

Build the FastAPI skeleton in Chapter 1 + 2. Create User model. POST + GET endpoint. Test with curl.

Exercise 2 — Aggregation

Build an endpoint /stats/users-per-country that returns count per country via an aggregation projected to a Pydantic model.

Exercise 3 — Links

Create Order linking to User. Insert a few. Endpoint /orders/{id} returns the order with the linked user fetched.

Exercise 4 — Hooks

Add a before_event(Insert) hook to Order that sets created_at and validates total > 0.

Exercise 5 — Bulk + transaction

Endpoint that inserts a batch of 100 users in a single bulkWrite. Wrap in a transaction (overkill for inserts; just to practice).

Exercise 6 — Migration

Add an email_index migration. Run beanie migrate. Verify getIndexes() from mongosh shows it.

Cheat-sheet

What	How
Install	`pip install beanie motor fastapi`
Init	`await init_beanie(db, document_models=[...])`
Model	`class X(Document): ...`
Index	`Annotated[T, Indexed(unique=True)]` or `Settings.indexes`
Query	`Model.find(Model.field == v)` (type-safe)
Insert	`await m.insert()` or `Model.insert_one(m)`
Update	`await m.save()` (state-managed) or `find(...).update({$set: ...})`
Delete	`await m.delete()`
Project	`find().project(SlimModel)`
Aggregate	`Model.aggregate([...], projection_model=Out)`
Link	`Link[User]` field; `fetch_links=True`
Hook	`@before_event(Insert)` / `@after_event(...)`
Transaction	`session.start_transaction()`
Bulk	`async with BulkWriter() as bulk: m.insert(bulk_writer=bulk)`
FastAPI	lifespan to init/close; Pydantic models for request/response
Migration	`beanie new-migration` / `beanie migrate`
Async only	use Motor under the hood

What's next

Post 20 covers the outbox pattern + transactional event publishing via change streams — the canonical way to publish reliable events when mutating MongoDB inside a service.

References

Note: this post is in progress. Planned additions: Beanie + Strawberry GraphQL integration; ODMantic vs Beanie comparison; testing Beanie with Testcontainers (Post 25).