Python新项目最佳实践

最近想居于Python、MongoDB这两个大的方向,做一个API接口的项目,方向大体确定,根据一贯作风,去谷歌检索

"python mongo best practices"

这其中需要解决一些问题:

  • ORM框架选择(MyBatis)
  • WEB选择(类似SpringMVC、SpringBootstart)
  • ...

ORM框架先后把MongoEngine、Beanie作为候选,WEB端历史记忆中对Flask比较熟悉;但后来得到MongoDB官方文档提示,FastAPI可能更合适当前,尝试新框架磨合。

8 Best Practices for Building FastAPI and MongoDB Applications | MongoDB
FastAPI seamlessly integrates with MongoDB through the Motor library, enabling asynchronous database interactions.

MongoDB官方文档(最佳实践)

慢慢调试出下列框架

  • 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
展示评论