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?
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