SQLModel 和 Pydantic 数据验证

Question

我正在学习FastAPI和SQLmodel，做一个简单的增删改查应用。

我想验证一下“名字”这个字段，确保在数据库中创建新条目时，它不是空的。

# 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

我把这个字段设置为必填，并添加了一个验证器，但我发现还是可以传递一个空字符串。我漏掉了什么呢？

Answer 1

评论中提到的解决方案；更多细节可以在GitHub上的讨论中找到。
我会提供一个修正后的代码示例，这段代码可以进行所需的验证。

模型：

# 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

路径操作：

# 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