Minimalist
This example shows a small but complete FastAPI application built with fastapi-crudrouter-mongodb.
It keeps the setup intentionally small while still showing two common patterns:
- embedded documents with
CRUDEmbed
- nested child routes with
CRUDLookup
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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81 | from datetime import datetime
from typing import Optional, Union
from fastapi import FastAPI
from pydantic import Field
from fastapi_crudrouter_mongodb import (
ObjectIdType,
MongoModel,
CRUDRouter,
CRUDLookup,
CRUDEmbed,
)
import motor.motor_asyncio
# Database connection using motor
client = motor.motor_asyncio.AsyncIOMotorClient("mongodb://localhost:27017/local")
# store the database in a global variable
db = client.local
# Database Model
class MessageModel(MongoModel):
id: ObjectIdType | None = None
message: str
user_id: ObjectIdType
created_at: str | None = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
updated_at: str | None = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
class AddressModel(MongoModel):
id: ObjectIdType | None = None
street: str
city: str
state: str
zip: str
class UserModel(MongoModel):
id: ObjectIdType | None = None
name: str
email: str
password: str
addresses: Optional[list[AddressModel]] = Field(default_factory=list)
messages: Optional[Union[list[MessageModel], MessageModel]] = None
# Model Out -> Schema
class UserModelOut(MongoModel):
id: str
name: str
email: str
addresses: list[AddressModel] = Field(default_factory=list)
# A user embeds addresses and is linked to many messages.
addresses_embed = CRUDEmbed(model=AddressModel, embed_name="addresses")
messages_lookup = CRUDLookup(
model=MessageModel,
collection_name="messages",
prefix="messages",
local_field="_id",
foreign_field="user_id",
)
users_router = CRUDRouter(
model=UserModel,
model_out=UserModelOut,
db=db,
collection_name="users",
lookups=[messages_lookup],
embeds=[addresses_embed],
prefix="/users",
tags=["users"],
)
# Instantiating the FastAPI app
app = FastAPI()
app.include_router(users_router)
|
This generates standard user CRUD routes plus nested message routes such as:
GET /users
POST /users
GET /users/{id}
GET /users/{id}/messages
POST /users/{id}/messages
It will produce an OpenAPI schema similar to this:
