pytest构建和测试FastAPI CURD API

文章目录

概述
目标
FASTAPI 介绍
- CRUD API
项目设置
- - freeze
  - pipreqs
代码介绍
run API
pytest测试
F&Q
- 1.执行uvicorn app.main:app --host localhost --port 8000 --reload 报错 zsh: /usr/local/bin/uvicorn: bad interpreter
- 2.生成requirement.txt时，pip3 list pipreqs有，但是执行pipreqs提示command not found
- - ~/.bash_profile
  - ~/.zshrc
- 3.pydantic_core._pydantic_core.ValidationError: 1 validation error for UserBaseSchema

概述

API是连接前后端的前梁，当然前端既可以是UI web的前端，也可以是client端，API的测试和构建一样重要。
FastAPI 是一个用python构建API非常不错的框架，如何确保在客户端或服务器发生错误时 API 返回有用的响应？您应该针对真实数据进行测试还是模拟测试？
在本文中，介绍构建增删改查API(增加(Create)、检索(Retrieve)、更新(Update)和删除(Delete))使用FastAPI、SQLite并使用fixture等对pytest进行测试。

这里介绍构建Rest API - 从数据库创建、读取、更新、删除用户。

目标

结合本文，从以下几个方面介绍

开发一个增删改查使用FastAPI框架在Python中实现Rest API
使用SQLAIchemy ORM 工具和SQLite数据库交互
使用pytest为FastAPI 进行单测
处理错误和响应
使用 FastAPI 的内置Swagger 记录REST API

对于API这里不在赘述，我们了解下FastAPI

FASTAPI 介绍

FastAPI是一个高性能的python web框架，可以轻松构建 API，最初由Sebastian Ramirez于18年创建，并在2019年发布。它是建立在Python 库之上：Starlette和Pydantic，其中Starlette 是提供底层 Web 应用程序框架，而 Pydantic 是用于数据验证和序列化的库。FastAPI的设计注重易用性，和性能，同时还内置了async/await的支持，使其比传统的同步线程模型更高效。

CRUD API

它是包括了HTTP方法（POST、GET、PUT、DELETE）的设计原则，用于对数据库DB系统中的数据维护的基本操作。广泛用于 Web 开发，用于实现内容管理和维护

项目设置

项目中，我们创建一个增删改查使用API从的关系型数据库（使用SQLite）创建、读取、更新和删除用户，项目名称这里叫fastapi_curdapi

fastapi_curdapi
├── app
│   ├── __init__.py
│   ├── database.py
│   ├── main.py
│   ├── models.py
│   ├── schemas.py
│   └── user.py
├── pyest.ini
├── pyproject.toml
├── requirements.txt
├── tests
│   ├── __init__.py
│   ├── conftest.py
│   └── test_curd_api.py

其中app目录下包含源代码如下所示：

database.py — 创建数据库引擎和 SQLite 设置。
main.py — 创建 FastAPI 客户端、健康检查器和中间件。
models.py - 数据库架构。
schemas.py — 用户基础架构和响应架构。
user.py — API 路由和响应格式。
tests目录包含 API 的单元测试。

文件中列出了依赖项 pyproject.toml，或者说使用requirement.txt 这里使用pip3进行维护

fastapi==0.111.0
pydantic==2.7.3
SQLAlchemy==2.0.30
SQLAlchemy_Utils==0.41.2

对于项目生成requirement.txt，常用有两种方法

freeze

应用场景：在单一虚拟环境下，可以使用这种方式。
优点：这种方式会把当前环境下的所有依赖包都整理出来。
缺点：不是针对某个项目生成，如果当前环境下有多个项目，那么会生成大量无关的依赖信息。

pip freeze > requirements.txt

但是用这个方法，可以实现一个功能：删除当前环境的所有python依赖。

pip uninstall -r requirements.txt -y

pipreqs

应用场景：针对单个项目生成 requirements.txt
优点：使用 pipreqs 可以自动检索到当前项目下的所有组件及其版本，并生成 requirements.txt 文件，极大方便了项目迁移和部署的包管理。
缺点：相比直接用 freeze 命令，能直接隔离其它项目的包生成。

pipreqs ./ --encoding=utf-8
#强制执行命令 --force ，覆盖原有的 requirements.txt 文件
pipreqs ./ --encoding=utf-8 --force

所以这里使用pipreqps

代码介绍

database.py是创建数据库引擎和SQLite设置的代码

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# This module is ***
# @Time : 2024/6/6 09:54
# @Author :
# function :
# @File : database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLITE_DATABASE_URL = "sqlite:///./user.db"

engine = create_engine(SQLITE_DATABASE_URL, echo=True, connect_args={"check_same_thread": False})

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

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

SQLite作为内存数据库，使用起来较为简便，当然还可以使用PostgreSQL、MySQL等数据库。而get_db函数是一个依赖项，将会为注入的每个请求创建一个新的数据库会话session。

models.py-数据库架构

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# This module is ***
# @Time : 2024/6/6 09:54
# @Author : 
# function :
# @File : models.py
import uuid

from sqlalchemy import Column, String, Boolean, TIMESTAMP
from sqlalchemy_utils import UUIDType
from sqlalchemy.sql import func

from app.database import Base


class User(Base):
    __tablename__ ="users"

    # Primary key and GUID type
    id = Column(UUIDType(binary=False), primary_key=True, default=uuid.uuid4())

    # String types with appropriate non0null constraints
    first_name = Column(
        String(255), nullable=False, index=True
    )  # Index for faster searches

    last_name = Column(
        String(255), nullable=False, index=True
    )  # Index for faster searches

    address = Column(String(255), nullable=True)

    # Boolean type with a deafult value
    activated = Column(Boolean, nullable=False, default=True)

    # Timestamp with timezone support
    createAt = Column(
        TIMESTAMP(timezone=True), nullable=False, server_default=func.now()
    )

    udpatedAt = Column(TIMESTAMP(timezone=True), default=None, onupdate=func.now())

本例中的数据模型非常简单 - 一个包含列的User表- id、first_name、last_name、address、activated、createdAt、updatedAt

使用SQLAlchemy ORM，我们定义表模式和列。

现在我们有了数据库模型，让我们使用Pydantic User创建和响应模型。
schemas.py — 用户基础架构和响应架构

from enum import Enum
from datetime import datetime
from typing import List
from pydantic import BaseModel, Field
from uuid import UUID


class UserBaseSchema(BaseModel):

    id: UUID = None
    first_name: str = Field(
        ..., description="The first name of the user", example="John"
    )
    last_name: str = Field(..., description="The last name of the user", example="Doe")
    address: str = None
    activated: bool = False
    createdAt: datetime = None
    # updatedAt: datetime = None
    updatedAt: datetime = Field(default_factory=datetime.utcnow)

    class Config:
        from_attributes = True
        populate_by_name = True
        arbitrary_types_allowed = True


class Status(Enum):
    Success = "Success"
    Failed = "Failed"


class UserResponse(BaseModel):
    Status: Status
    User: UserBaseSchema


class GetUserResponse(BaseModel):
    Status: Status
    User: UserBaseSchema


class ListUserResponse(BaseModel):
    status: Status
    results: int
    users: List[UserBaseSchema]


class DeleteUserResponse(BaseModel):
    Status: Status
    Message: str

上面的用法中使用到了 Pydantic 模型，用于在API路由中验证请求和响应负载信息

user.py- API路由和响应信息

import app.schemas as schemas
import app.models as models
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from fastapi import Depends, HTTPException, status, APIRouter
from app.database import get_db

router = APIRouter()


@router.post(
    "/", status_code=status.HTTP_201_CREATED, response_model=schemas.UserResponse
)
def create_user(payload: schemas.UserBaseSchema, db: Session = Depends(get_db)):
    try:
        # Create a new user instance from the payload
        new_user = models.User(**payload.model_dump())
        db.add(new_user)
        db.commit()
        db.refresh(new_user)

    except IntegrityError as e:
        db.rollback()
        # Log the error or handle it as needed
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="A user with the given details already exists.",
        ) from e
    except Exception as e:
        db.rollback()
        # Handle other types of database errors
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="An error occurred while creating the user.",
        ) from e

    # Convert the SQLAlchemy model instance to a Pydantic model
    user_schema = schemas.UserBaseSchema.from_orm(new_user)
    # Return the successful creation response
    return schemas.UserResponse(Status=schemas.Status.Success, User=user_schema)


@router.get(
    "/{userId}", status_code=status.HTTP_200_OK, response_model=schemas.GetUserResponse
)
def get_user(userId: str, db: Session = Depends(get_db)):
    user_query = db.query(models.User).filter(models.User.id == userId)
    db_user = user_query.first()

    if not db_user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"No User with this id: `{userId}` found",
        )

    try:
        return schemas.GetUserResponse(
            Status=schemas.Status.Success, User=schemas.UserBaseSchema.model_validate(db_user)
        )
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="An unexpected error occurred while fetching the user.",
        ) from e


@router.patch(
    "/{userId}",
    status_code=status.HTTP_202_ACCEPTED,
    response_model=schemas.UserResponse,
)
def update_user(
    userId: str, payload: schemas.UserBaseSchema, db: Session = Depends(get_db)
):
    user_query = db.query(models.User).filter(models.User.id == userId)
    db_user = user_query.first()

    if not db_user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"No User with this id: `{userId}` found",
        )

    try:
        update_data = payload.dict(exclude_unset=True)
        user_query.update(update_data, synchronize_session=False)
        db.commit()
        db.refresh(db_user)
        user_schema = schemas.UserBaseSchema.model_validate(db_user)
        return schemas.UserResponse(Status=schemas.Status.Success, User=user_schema)
    except IntegrityError as e:
        db.rollback()
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="A user with the given details already exists.",
        ) from e
    except Exception as e:
        db.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="An error occurred while updating the user.",
        ) from e


@router.delete(
    "/{userId}",
    status_code=status.HTTP_202_ACCEPTED,
    response_model=schemas.DeleteUserResponse,
)
def delete_user(userId: str, db: Session = Depends(get_db)):
    try:
        user_query = db.query(models.User).filter(models.User.id == userId)
        user = user_query.first()
        if not user:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"No User with this id: `{userId}` found",
            )
        user_query.delete(synchronize_session=False)
        db.commit()
        return schemas.DeleteUserResponse(
            Status=schemas.Status.Success, Message="User deleted successfully"
        )
    except Exception as e:
        db.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="An error occurred while deleting the user.",
        ) from e


@router.get(
    "/", status_code=status.HTTP_200_OK, response_model=schemas.ListUserResponse
)
def get_users(
    db: Session = Depends(get_db), limit: int = 10, page: int = 1, search: str = ""
):
    skip = (page - 1) * limit

    users = (
        db.query(models.User)
        .filter(models.User.first_name.contains(search))
        .limit(limit)
        .offset(skip)
        .all()
    )
    return schemas.ListUserResponse(
        status=schemas.Status.Success, results=len(users), users=users
    )

上述代码定了 C R U D用户的API路由信息，可以更具需要再处理错误信息，日志记录，响应格式等方面的复杂功能。主要围绕4条信息开展的，也是我么日常使用较多的错误码类型。

create_user- 创建新用户。201成功或409发生冲突时返回状态代码。
get_user- 通过 ID 获取用户。如果200成功则返回状态代码，404如果未找到则返回状态代码。
update_user- 通过 ID 更新用户。202成功或409发生冲突时返回状态代码。
delete_user- 根据 ID 删除用户。如果202成功则返回状态代码，404如果未找到则返回状态代码。
get_users- 获取用户列表。200成功时返回状态代码。

main.py — 创建 FastAPI 客户端、健康检查器和中间件

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# This module is ***
# @Time : 2024/6/6 09:54
# @Author : 
# function :
# @File : main.py
from app import models, user
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.database import engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

origins = [
    "http://localhost:3000",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


app.include_router(user.router, tags=["Users"], prefix="/api/users")


@app.get("/api/healthchecker")
def root():
    return {"message": "The API is LIVE!!"}

上面的信息创建了FastAPI client，设置了 CORS中间件并定义了API路由user.py

run API

要拉起服务，执行如下cli命令

uvicorn app.main:app --host localhost --port 8000 --reload

在这里插入图片描述
在浏览器输入：http://localhost:8000/docs

通过使用不同接口完成测试。

pytest测试

tests使用conftest.py完成封装，使用test_curd_api完成业务测试。在conftest.py中

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# This module is ***
# @Time : 2024/6/6 09:54
# @Author : 
# function :
# @File : conftest.py
import pytest
import uuid
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from fastapi.testclient import TestClient
from app.main import app
from app.database import Base, get_db

# SQLite database URL for testing
SQLITE_DATABASE_URL = "sqlite:///./test_db.db"

# Create a SQLAlchemy engine
engine = create_engine(
    SQLITE_DATABASE_URL,
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)

# Create a sessionmaker to manage sessions
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Create tables in the database
Base.metadata.create_all(bind=engine)


@pytest.fixture(scope="function")
def db_session():
    """Create a new database session with a rollback at the end of the test."""
    connection = engine.connect()
    transaction = connection.begin()
    session = TestingSessionLocal(bind=connection)
    yield session
    session.close()
    transaction.rollback()
    connection.close()


@pytest.fixture(scope="function")
def test_client(db_session):
    """Create a test client that uses the override_get_db fixture to return a session."""

    def override_get_db():
        try:
            yield db_session
        finally:
            db_session.close()

    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as test_client:
        yield test_client


# Fixture to generate a random user id
@pytest.fixture()
def user_id() -> uuid.UUID:
    """Generate a random user id."""
    return str(uuid.uuid4())


# Fixture to generate a user payload
@pytest.fixture()
def user_payload(user_id):
    """Generate a user payload."""
    return {
        "id": user_id,
        "first_name": "John",
        "last_name": "Doe",
        "address": "123 Farmville",
    }


@pytest.fixture()
def user_payload_updated(user_id):
    """Generate an updated user payload."""
    return {
        "first_name": "Jane",
        "last_name": "Doe",
        "address": "321 Farmville",
        "activated": True,
    }

接着进行测试用例的编写

tests/test_crud_api.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# This module is ***
# @Time : 2024/6/6 09:55
# @Author : 
# function :
# @File : test_curd_api.py
import time


def test_root(test_client):
    response = test_client.get("/api/healthchecker")
    assert response.status_code == 200
    assert response.json() == {"message": "The API is LIVE!!"}


def test_create_get_user(test_client, user_payload):
    response = test_client.post("/api/users/", json=user_payload)
    response_json = response.json()
    assert response.status_code == 201

    # Get the created user
    response = test_client.get(f"/api/users/{user_payload['id']}")
    assert response.status_code == 200
    response_json = response.json()
    assert response_json["Status"] == "Success"
    assert response_json["User"]["id"] == user_payload["id"]
    assert response_json["User"]["address"] == "123 Farmville"
    assert response_json["User"]["first_name"] == "John"
    assert response_json["User"]["last_name"] == "Doe"


def test_create_update_user(test_client, user_payload, user_payload_updated):
    response = test_client.post("/api/users/", json=user_payload)
    response_json = response.json()
    assert response.status_code == 201

    # Update the created user
    time.sleep(
        1
    )  # Sleep for 1 second to ensure updatedAt is different (datetime precision is low in SQLite)
    response = test_client.patch(
        f"/api/users/{user_payload['id']}", json=user_payload_updated
    )
    response_json = response.json()
    assert response.status_code == 202
    assert response_json["Status"] == "Success"
    assert response_json["User"]["id"] == user_payload["id"]
    assert response_json["User"]["address"] == "321 Farmville"
    assert response_json["User"]["first_name"] == "Jane"
    assert response_json["User"]["last_name"] == "Doe"
    assert response_json["User"]["activated"] is True
    assert (
        response_json["User"]["updatedAt"] is not None
        and response_json["User"]["updatedAt"] > response_json["User"]["createdAt"]
    )


def test_create_delete_user(test_client, user_payload):
    response = test_client.post("/api/users/", json=user_payload)
    response_json = response.json()
    assert response.status_code == 201

    # Delete the created user
    response = test_client.delete(f"/api/users/{user_payload['id']}")
    response_json = response.json()
    assert response.status_code == 202
    assert response_json["Status"] == "Success"
    assert response_json["Message"] == "User deleted successfully"

    # Get the deleted user
    response = test_client.get(f"/api/users/{user_payload['id']}")
    assert response.status_code == 404
    response_json = response.json()
    assert response_json["detail"] == f"No User with this id: `{user_payload['id']}` found"


def test_get_user_not_found(test_client, user_id):
    response = test_client.get(f"/api/users/{user_id}")
    assert response.status_code == 404
    response_json = response.json()
    assert response_json["detail"] == f"No User with this id: `{user_id}` found"


def test_create_user_wrong_payload(test_client):
    response = test_client.post("/api/users/", json={})
    assert response.status_code == 422


def test_update_user_wrong_payload(test_client, user_id, user_payload_updated):
    user_payload_updated["first_name"] = (
        True  # first_name should be a string not a boolean
    )
    response = test_client.patch(f"/api/users/{user_id}", json=user_payload_updated)
    assert response.status_code == 422
    response_json = response.json()
    assert response_json == {
        "detail": [
            {
                "type": "string_type",
                "loc": ["body", "first_name"],
                "msg": "Input should be a valid string",
                "input": True,
            }
        ]
    }


def test_update_user_doesnt_exist(test_client, user_id, user_payload_updated):
    response = test_client.patch(f"/api/users/{user_id}", json=user_payload_updated)
    assert response.status_code == 404
    response_json = response.json()
    assert response_json["detail"] == f"No User with this id: `{user_id}` found"

执行测试用例的时候，可以直接点击ide，这里使用的是macOs pycharm中的三角，也可以执行

pytest

F&Q

执行为macOS Sonoma 14.0 python3.9.6

# python3 -V
Python 3.9.6

1.执行uvicorn app.main:app --host localhost --port 8000 --reload 报错 zsh: /usr/local/bin/uvicorn: bad interpreter

查看python路径

% which python3
/usr/bin/python3

检修uvicorn脚本的Shebang行

which uvicorn
/Python/3.9/bin/uvicorn

既然报错解释器相关的问题，进入到文件中，将第一行解释器的路径修改为

#!/usr/bin/python3

保存退出即可。

2.生成requirement.txt时，pip3 list pipreqs有，但是执行pipreqs提示command not found

查询pipreqs命令

ffastapi_curdapi % which pipreqs                    
pipreqs not found
fastapi_curdapi % pip3 list| grep pipreqs
pipreqs                   0.5.0
fastapi_curdapi % pipreqs
zsh: command not found: pipreqs
fastapi_curdapi % 
bfastapi_curdapi % which python3
/usr/bin/python3

如果which中找不到，则说明不在$PATH中，这里看到Python位于/usr/bin/下，而macOS中一般客户手动安装的路径位于/usr/local/bin/下，查询pipreqs的安装路径

fastapi_curdapi % pip3 show pipreqs | grep Location
Location: ~/Library/Python/3.9/lib/python/site-packages

重点关注输出中的Location字段，它会显示该包的安装路径。全局安装的包一般位于/usr/local/lib/pythonX.Y/site-packages（X.Y代表你的Python版本），而用户级安装的包通常位于~/Library/Python/X.Y/lib/python/site-packages
在用户级目录下发现

bin % pwd
***/Library/Python/3.9/bin
bin % ls | grep pipreqs
pipreqs

需要将对应的路径添加到PATH中，添加到用户PATH中，一般有~/.bash_profile和~/.zshrc两个文件，
~/.bash_profile 和 ~/.zshrc 是两种不同Shell的配置文件，它们分别用于Bash和Zsh这两种Unix/Linux shell环境中。它们存储着用户自定义的设置和环境变量，使得用户可以在启动shell时应用自己的个性化配置。下面是两者的具体区别：

~/.bash_profile

用途: 主要用于Bash shell。它是Bash在登录（login）模式下读取的第一个配置文件。当用户通过终端登录或者使用ssh远程登录时，Bash会执行这个文件来设置环境变量（如PATH）、别名（aliases）以及其他一些初始化设置。
特点: 适合放置那些只需要在登录时执行一次的配置，比如环境变量设置、启动程序等。
兼容性: 由于Bash是大多数Linux发行版和macOS的默认shell，所以这个文件在这些系统中非常常见。

~/.zshrc

用途: 专用于Zsh shell。无论是在交互式登录还是非登录shell中，Zsh都会读取这个文件。这意味着每次你打开一个新的Zsh终端窗口或tab时，~/.zshrc中的配置都会被应用。
特点: 由于Zsh提供了比Bash更多的功能和定制性，~/.zshrc可以包含更复杂的配置，如插件管理（如oh-my-zsh）、主题设置、自定义函数、别名等。
兼容性: 自macOS Catalina开始，Apple将Zsh设为默认shell，因此在较新的macOS系统中，~/.zshrc成为用户配置shell环境的主要文件
如果你正在使用macOS并且主要使用的是Zsh（尤其是在Catalina及以后的版本），那么你应当关注_{/.zshrc的配置。如果你在某些场景下仍然使用Bash（或者系统默认是Bash），则需要维护}/.bash_profile。两者都非常重要，因为它们决定了你的shell环境如何响应你的命令以及提供哪些功能。在一些情况下，用户可能需要在这两个文件中都做一些配置，特别是当他们同时在Bash和Zsh环境中工作时。而且macOS Catalina是Mac的第16个主要版本，版本号为macOS 10.15,从 2019.10.8 开始，所以可以认为只需要维护~/.zshrc即可。

# 使用的macos的默认python，位于/usr/bin下，而用户级一般位于/usr/local/bin下，因此需要将用户级安装目录加入到PATH中
export PATH="自己的目录地址/Library/Python/3.9/bin:$PATH"

保存退出后，执行

source ~/.zshrc

这里大家会好奇机器明明Sonoma 14.0为什么要修改 .zshrc呢，当然是确定本机的默认Shell已经从Bash是否切换到了Zsh。执行命令

% echo $SHELL
/bin/zsh

发现这里已经切换到zsh，所以上面需要修改~/.zshrc

3.pydantic_core._pydantic_core.ValidationError: 1 validation error for UserBaseSchema

在执行pytest全量的case时报错如下：

E       pydantic_core._pydantic_core.ValidationError: 1 validation error for UserBaseSchema
E       updatedAt
E         Input should be a valid datetime [type=datetime_type, input_value=None, input_type=NoneType]
E           For further information visit https://errors.pydantic.dev/2.7/v/datetime_type

../../../../Library/Python/3.9/lib/python/site-packages/pydantic/main.py:1220: ValidationError

在这里插入图片描述
这个错误信息表明你在使用Pydantic模型（UserBaseSchema）验证数据时遇到了问题，具体来说，错误发生在updatedAt字段，它期望一个有效的datetime类型值，但实际上收到了None，所以发现在schemas.py中定义updatedAt和createdAt都使用了初始化值为None的情况

...
    address: str = None
    activated: bool = False
    createdAt: datetime = None
    # updatedAt: datetime = None
    updatedAt: datetime = Field(default_factory=datetime.utcnow)
...