最近想居于Python、MongoDB这两个大的方向,做一个API接口的项目,方向大体确定,根据一贯作风,去谷歌检索
"python mongo best practices"
这其中需要解决一些问题:
- ORM框架选择(MyBatis)
- WEB选择(类似SpringMVC、SpringBootstart)
- ...
ORM框架先后把MongoEngine、Beanie作为候选,WEB端历史记忆中对Flask比较熟悉;但后来得到MongoDB官方文档提示,FastAPI可能更合适当前,尝试新框架磨合。
慢慢调试出下列框架
- FastAPI (APIRouter按角色分组路由)
- Motor (MongoDB驱动框架)
- pydantic 实体注解规范,如Field、
id: Optional[str] = Field(default_factory=str, alias="_id", validation_alias=AliasChoices('id', '_id', 'note_id'))
注解实践
枚举的注解示例:
class NoteType(str, Enum):
""" 笔记类型 """
VIDEO = 'V'
NOTE = 'N'
@classmethod
def of(cls, value: str) -> Any:
if value is None or value == '': return None
for k in NoteType:
if value == k.lower() or value == k.title() or value == k.name or value == k.name.lower(): return k
return value
FastAPI处理数据,并入库
@router.get("/{username}")
async def read(username: str, request: Request):
resp = await request.app.state.user_repository.find_one({'nickname': username})
if resp is None or resp.data is None:
return HTTPException(status_code=404, detail="User not found")
return User.model_validate(resp.data).model_dump(exclude_none=True)
在使用类型注解时,为了快速对象与dict间转换即
User.model_validate(data)
# 多个不同的key对应到一个属性
class User(BaseModel):
id: Optional[str] = Field(default_factory=str, alias="_id",
validation_alias=AliasChoices('id', '_id', 'user_id', 'userid'))
service层实践
user_service.py
class UserService:
...
try:
# 更新用户基础数据
if item is not None and item.t is not None:
await app.state.user_repository.upsert(item.model_dump(exclude_none=True, by_alias=True))
else:
await app.state.user_repository.insert_one(item.model_dump(exclude_none=True, by_alias=True))
except BulkWriteError:
# 忽略错误
pass