如何在从 Pydantic 的 BaseModel 创建的模型中访问 FastAPI 的请求对象和依赖项

Question

我正在使用堆栈 FastAPI、Pydantic 和 SQL Alchemy 编写 API，我遇到过很多情况，我必须查询数据库才能对有效负载值执行验证。让我们考虑一个示例 API，/forgot-password。此 API 将在有效负载中接受 email，我需要验证数据库中是否存在该电子邮件。如果电子邮件存在于数据库中，则将执行创建令牌和发送邮件等必要操作，否则 Pydantic 应针对该字段引发错误响应。错误响应必须是标准的PydanticValueError响应。这是因为所有验证错误都会有一致的响应，因为它变得易于为消费者处理。

有效载荷 -

{
    "email": "example@gmail.com"
}

在 Pydantic 中，此模式和电子邮件验证实现为 -

class ForgotPasswordRequestSchema(BaseModel):
    email: EmailStr
    
    @validator("email")
    def validate_email(cls, v):
        # this is the db query I want to perform but 
        # I do not have access to the active session of this request.
        user = session.get(Users, email=v) 
        if not user:
            raise ValueError("Email does not exist in the database.")

        return v

现在，如果我们像这样在 pydantic 模型中简单地创建一个 Alchemy 会话，这就很容易处理了。

class ForgotPasswordRequestSchema(BaseModel):
    email: EmailStr
    _session = get_db() # this will simply return the session of database.
    _user = None
    
    @validator("email")
    def validate_email(cls, v):
        # Here I want to query on Users's model to see if the email exist in the 
        # database. If the email does. not exist then I would like to raise a custom 
        # python exception as shown below.

        user = cls._session.get(Users, email=v) # Here I can use session as I have 
        # already initialised it as a class variable.

        if not user:
            cls.session.close()
            raise ValueError("Email does not exist in the database.")

        cls._user = user # this is because we want to use user object in the request 
        # function.

        cls.session.close()

        return v

但这不是一种正确的方法，因为在整个请求过程中只应使用一个会话。正如您在上面的示例中看到的那样，我们正在关闭会话，因此我们将无法在请求函数中将用户对象用作user = payload._user。这意味着我们将不得不再次查询请求函数中的同一行。如果我们不关闭会话，那么我们会看到像这样的炼金术异常 - sqlalchemy.exc.PendingRollbackError。

现在，最好的方法是能够在请求开始时创建并在请求结束时关闭的 Pydantic 模型中使用相同的会话。

所以，我基本上是在寻找一种方法将该会话作为上下文传递给 Pydantic。我的请求函数的会话作为依赖项提供。

Answer 1

不要那样做！

pydantic 类的目的是以合法的方式存储字典，因为它们具有 IDE 支持并且不易出错。验证器用于非常简单的东西，不会触及系统的其他部分（例如正整数或电子邮件是否满足正则表达式）。

这么说，你应该use the dependencies。这样你就可以确保在处理所有请求期间只有一个会话，并且由于上下文管理器，会话将在任何情况下关闭。

最终解决方案可能如下所示：

from fastapi import Body, Depends
from fastapi.exceptions import HTTPException

def get_db():
    db = your_session_maker
    try:
        yield db
    finally:
        db.close()

@app.post("/forgot-password/")
def forgot_password(email: str = Body(...), db: Session = Depends(get_db)):
    user = db.get(Users, email=email)
    if not user:
        # If you really need to, you can for some reason raise pydantic exception here
        raise HTTPException(status_code=400, detail="No email")