目录:
ODMantic 模块是一个文档映射库，专门用于操作 MongoDB 数据库。
安装依赖:
$ pip install odmantic -i https://pypi.tuna.tsinghua.edu.cn/simple
只需继承 Model 类，可以使用 Field 函数为字段添加额外的约束和元数据。
这个 User 类不仅定义了文档的结构，还通过 Field 指定了业务规则（如 age > 0）和数据库约束（如 email 唯一）。
from odmantic import Field, Model
# 定义一个用户模型
class User(Model):
name: str
age: int = Field(gt=0) # 年龄必须大于0
email: str = Field(unique=True) # 邮箱必须唯一
创建引擎实例来连接 MongoDB 数据库。
from odmantic import AIOEngine
# 创建异步引擎，连接到本地 MongoDB
engine = AIOEngine(database="my_database")
使用引擎进行基本的增删改查操作:
# 创建一个用户实例
user = User(name="张三", age=30, email="zhangsan@example.com")
# 保存到数据库 (异步操作)
await engine.save(user)
# 查询用户 (异步操作)
found_user = await engine.find_one(User, User.name == "张三")
print(found_user)
查询示例:
async def get_all_users():
# 使用异步生成器遍历所有用户
async for user in engine.find(User):
print(user.name, user.email)
# 带条件的查询
async def get_adult_users():
users = await engine.find(User, User.age >= 18)
return users
from fastapi import FastAPI, HTTPException
from odmantic import AIOEngine, Model
app = FastAPI()
engine = AIOEngine(database="fastapi_demo")
class Item(Model):
name: str
price: float
is_offer: bool = False
@app.post("/items/")
async def create_item(item: Item):
await engine.save(item)
return item
@app.get("/items/{item_name}")
async def read_item(item_name: str):
item = await engine.find_one(Item, Item.name == item_name)
if item is None:
raise HTTPException(status_code=404, detail="Item not found")
return item
↶ 返回首页 ↶