2

I am writing APIs using stack FastAPI, Pydantic & SQL Alchemy and I have come across many cases where I had to query database to perform validations on payload values. Let's consider one example API, /forgot-password. This API will accept email in the payload and I need to validate the existence of the email in database. If the email exist in the database then necessary action like creating token and sending mail would be performed or else an error response against that field should be raise by Pydantic. The error responses must be the standard PydanticValueError response. This is because all the validation errors would have consistent responses as it becomes easy to handle for the consumers.

Payload -

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

In Pydantic this schema and the validation for email is implemented as -

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

Now this can be easily handled if the we simple create an Alchemy session in the pydantic model like this.

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

But it is not a right approach as through out the request only one session should be used. As you can see in above example we are closing the session so we won't be able to use the user object in request function as user = payload._user. This means we will have to again query for the same row in request function. If we do not close the session then we are seeing alchemy exceptions like this - sqlalchemy.exc.PendingRollbackError.

Now, the best approach is to be able to use the same session in the Pydantic model which is created at the start of request and is also closing at the end of the request.

So, I am basically looking for a way to pass that session to Pydantic as context. Session to my request function is provided as dependency.

snakecharmerb
  • 47,570
  • 11
  • 100
  • 153
Jeet Patel
  • 1,140
  • 18
  • 51
  • 2
    Usually you'd use a dependency in FastAPI to fetch any user, instead of doing that inside a pydantic validator; generally a Pydantic validator shouldn't have business logic (in my view); that belongs to a service or other part of your application. Which means that you'd have something like `@app.get, async def reset_password_from_email(user: User = Depends(get_valid_user_from_email):` - `get_valid_user_from_email` would then have the signature and be responsible for fetching anything from the current db (through a service) and generate the proper error code if necessary. – MatsLindh Feb 03 '23 at 14:13
  • That way the service is only concerned with fetching and handling of users, while the application-dependency is concerned with getting the parameter, fetching the user and generating any errors, while your controller is concerned with "what does this endpoint actually do". – MatsLindh Feb 03 '23 at 14:14

2 Answers2

1

Don't do that!

The purpose of pydantic classes is to store dictionaries in a legit way, as they have IDE support and are less error prone. The validators are there for very simple stuff that doesn't touch other parts of system (like is integer positive or does email satisfy the regex).

Saying that, you should use the dependencies. That way you can be sure you have single session during processing all request and because of context manager the session will be closed in any case.

Final solution could look like this:

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")
kosciej16
  • 6,294
  • 1
  • 18
  • 29
1

It is not recommended to query the database in pydantic schema. Instead use session as a dependency.

If you want to raise errors like pydantic validation error you might need this:

def raise_custom_error(exc: Exception, loc: str, model: BaseModel, status_code=int, **kwargs):
    """
    This method will return error responses using pydantic error wrapper (similar to pydantic validation error).
    """
    raise HTTPException(
        detail=json.loads(ValidationError([ErrorWrapper(exc(**kwargs), loc=loc)], model=model).json()),
        status_code=status_code,
    )

Usage 

class PayloadSchema(BaseModel):
    email: EmailStr

@app_router.post('/forgot-password')
def forgot_password(
    payload: PayloadSchema,
    session: Session = Depends(get_db),
    background_tasks: BackgroundTasks
):
    
    existing_user = db.get(Users, email=payload.email)
    if(existing_user):
        raise_custom_error(
        PydanticValueError, "email", PayloadSchema, status.HTTP_400_BAD_REQUEST
    )
    background_tasks(send_email, email=payload.email)
Anubhav Gupta
  • 74
  • 6