pythonfastapipydanticsqlmodel

SQLModel and Pydantic data validation


I am learning FastAPI and SQLmodel on a simple CRUD app. I am trying to validate the "name" field to ensure that it is not empty when a new entry is created in the database.

# Define the SQLModel for the User table
class UserBase(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(..., min_length=1, nullable=False)
    email: str = Field(min_length=1, default=None, unique=True)
    password: str = Field(min_length=1)
    date_created: datetime = Field(default=datetime.now(), nullable=False)

    @validator('name')
    def name_must_not_be_empty(cls, v):
        if v.strip() == '':
            raise ValueError('Name cannot be an empty string')
        return v

# Define the schema for a response upon user creation
class CreateUser(BaseModel):
    message: str = 'Successfully created user'
    id: int
    name: str
    email: str

# Create a new user
@app.post('/users/', response_model=CreateUser)
def create_user(user: UserBase, session: Session = Depends(get_session)):
    session.add(user)
    session.commit()
    return user

I made the field required and added a validator, yet an empty string can still be passed. What am I missing here?

POST new user with empty name


Solution

  • The solution suggested in the comments; more details available in issue discussion on Github
    I will leave an example of the corrected code that performs the required validation.

    Models:

    
    # Define the User model; it is only Pydantic data model
    class UserBase(SQLModel):
        name: str = Field(nullable=False)
        email: EmailStr = Field(sa_column=Column("email", VARCHAR, unique=True))
    
        @validator('name')
        def name_must_not_be_empty(cls, v):
            if v.strip() == '':
                raise ValueError('Name cannot be an empty string')
            return v
    
    
    # Define the User model that declares the data in the database
    # Represents a table; it is both Pydantic model and SQLAlchemy model
    class User(UserBase, table=True):
        id: Optional[int] = Field(default=None, primary_key=True)
        date_created: datetime = Field(default=datetime.now(), nullable=False)
    
    
    # Define the schema for Creating a new User; it is only Pydantic data model
    # Declares required fields in addition to fields from Base model
    class UserCreate(UserBase):
        password: str = Field(nullable=False, min_length=6)
    
        @validator('password')
        def validate_password(cls, v):
            if len(v) < 6:
                raise ValueError('Password must be at least 6 characters long')
            if not re.search(r'\W', v):
                raise ValueError(
                    'Password must contain at least one special character')
            if not re.search(r'[A-Z]', v):
                raise ValueError(
                    'Password must contain at least one uppercase letter')
            return v
    
    
    # Define the schema for Reading a User; it is only Pydantic data model
    # These additional fields will shape the response model when requeuing a user data
    class UserRead(UserBase):
        id: int
        date_created: datetime
    
    
    # Define the schema for Updating a User; independent Pydantic data model
    # We create an independent model since the same fields are required in Base
    class UserUpdate(SQLModel):
        name: Optional[str] = None
        email: Optional[EmailStr] = None
    
    

    Path operation:

    
    # Get all users
    @app.get('/users/', response_model=list[UserRead])
    def read_users(session: Session = Depends(get_session)):
        db_users = session.exec(select(User)).all()
        return db_users
    
    
    # Get a specific user
    @app.get('/users/{user_id}', response_model=UserRead)
    def read_user(user_id: int, session: Session = Depends(get_session)):
        statement = select(User).where(User.id == user_id)
        db_user = session.exec(statement).first()
        if not db_user:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                                detail='User not found')
        return db_user
    
    
    # Create a new user
    @app.post('/users/', response_model=UserRead)
    def create_user(user: UserCreate, session: Session = Depends(get_session)):
        db_user = User.model_validate(user)
        session.add(db_user)
        session.commit()
        session.refresh(db_user)
        return db_user
    
    
    # Update a user
    @app.patch('/users/{user_id}', response_model=UserRead)
    def update_user(user_id: int,
                    user: UserUpdate,
                    session: Session = Depends(get_session)):
        db_user = session.get(User, user_id)
        if not db_user:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                                detail='User not found')
        user_data = user.model_dump(exclude_unset=True)
        db_user.sqlmodel_update(user_data)
        session.add(db_user)
        session.commit()
        session.refresh(db_user)
        return db_user
    
    
    # Delete a user
    @app.delete('/users/{user_id}', status_code=status.HTTP_204_NO_CONTENT)
    def delete_user(user_id: int, session: Session = Depends(get_session)):
        db_user = session.get(User, user_id)
        if not db_user:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                                detail='User not found')
        session.delete(db_user)
        session.commit()
        return None