项目架构
Conda 隔离环境
# 创建环境conda create-nreport-pythonpython=3.13# 激活环境conda activate report-python项目结构
report-python/ ├── .env # 环境变量 ├── requirements.txt ├── alembic.ini # Alembic 配置 ├── alembic/ │ ├── env.py # Alembic 异步 env │ ├── script.py.mako │ └── versions/ # 迁移版本文件 ├── src/ │ ├── __init__.py │ ├── main.py # 应用入口 │ ├── core/ │ │ ├── __init__.py │ │ ├── config.py # 全局配置 │ │ ├── logger.py # Loguru 配置 │ │ ├── base_model.py # ORM 基类 │ │ ├── base_repository.py # 通用 Repository 基类 │ │ ├── base_schema.py # 通用响应 Schema │ │ └── exceptions.py # 全局异常 & Handler │ ├── infra/ │ │ ├── database.py # 异步引擎 & Session │ ├── middlewares/ │ │ ├── __init__.py │ │ └── logging.py # 请求日志中间件 │ └── modules/ │ ├── __init__.py │ └── user/ # 示例：用户模块 │ ├── __init__.py │ ├── model.py # ORM Model │ ├── schema.py # Pydantic Schema │ ├── repository.py # 数据访问 │ ├── service.py # 业务逻辑 │ └── api.py # 路由 │ ├── utils/ # 公共方法每个业务模块的职责：
model.py— SQLAlchemy ORM 模型schema.py— Pydantic 请求/响应模型repository.py— 数据库 CRUD 操作service.py— 业务逻辑编排api.py— FastAPI 路由定义
搭建脚手架
搭建脚手架
步骤 1：安装依赖
# fastapi： web开发# sqlalchemy： 数据库orm框架# asyncmy： mysql异步驱动# cryptography： mysql密码套件# alembic： 数据库迁移工具# loguru：日志工具pipinstall"fastapi[standard]==0.135.1"sqlalchemy==2.0.48 asyncmy loguru alembic更新 requirements.txt：
pip freeze > requirements.txt步骤 2：创建 .env
# 应用APP_NAME=MyApp APP_ENV=development APP_DEBUG=true# MySQLDB_HOST=127.0.0.1 DB_PORT=3306 DB_USER=root DB_PASSWORD=your_password DB_NAME=myapp# 日志LOG_LEVEL=DEBUG LOG_DIR=logs步骤 3：创建所有目录
mkdir-p src/core src/middlewares src/modules/user步骤 4：全局配置 —src/core/config.py
frompydantic_settingsimportBaseSettingsfromfunctoolsimportlru_cacheclassSettings(BaseSettings):APP_NAME:str="MyApp"APP_ENV:str="development"APP_DEBUG:bool=TrueDB_HOST:str="127.0.0.1"DB_PORT:int=3306DB_USER:str="root"DB_PASSWORD:str=""DB_NAME:str="myapp"LOG_LEVEL:str="DEBUG"LOG_DIR:str="logs"@propertydefDATABASE_URL(self)->str:return(f"mysql+asyncmy://{self.DB_USER}:{self.DB_PASSWORD}"f"@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"f"?charset=utf8mb4")# 指定环境变量文件model_config={"env_file":".env","env_file_encoding":"utf-8"}# 保存到内存缓存中。以后直接获取。这是一种单例的实现@lru_cachedefget_settings()->Settings:returnSettings()步骤 5：Loguru 日志配置 —src/core/logger.py
importsysfrompathlibimportPathfromloguruimportloggerfromsrc.core.configimportget_settingsdefsetup_logger()->None:settings=get_settings()logger.remove()# 控制台logger.add(sys.stdout,level=settings.LOG_LEVEL,format=("<green>{time:YYYY-MM-DD HH:mm:ss}</green> | ""<level>{level: <8}</level> | ""<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - ""<level>{message}</level>"),colorize=True,)# 文件log_dir=Path(settings.LOG_DIR)log_dir.mkdir(parents=True,exist_ok=True)logger.add(str(log_dir/"{time:YYYY-MM-DD}.log"),level=settings.LOG_LEVEL,format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}",rotation="00:00",retention="30 days",compression="gz",encoding="utf-8",)步骤 6：数据库引擎 & Session —src/infra/database.py
Infra：属于基础设施层整合。未来redis、mysql、**minio **都在这
fromsqlalchemy.ext.asyncioimportcreate_async_engine,async_sessionmaker,AsyncSessionfromsrc.core.configimportget_settings settings=get_settings()engine=create_async_engine(settings.DATABASE_URL,echo=settings.APP_DEBUG,pool_size=10,max_overflow=20,pool_recycle=3600,pool_pre_ping=True,)AsyncSessionLocal=async_sessionmaker(bind=engine,class_=AsyncSession,expire_on_commit=False,)asyncdefget_db()->AsyncSession:"""FastAPI Depends 注入, 自动提交和异常回滚"""asyncwithAsyncSessionLocal()assession:try:yieldsessionawaitsession.commit()exceptException:awaitsession.rollback()raise步骤 7：ORM 基类 —src/core/base_model.py
fromdatetimeimportdatetimefromsqlalchemyimportBigInteger,DateTime,funcfromsqlalchemy.ormimportDeclarativeBase,Mapped,mapped_columnclassBase(DeclarativeBase):"""所有 Model 继承此类"""pass# 创建时间和更新时间由数据库自动维护classTimestampMixin:created_at:Mapped[datetime]=mapped_column(DateTime,server_default=func.now(),comment="创建时间")updated_at:Mapped[datetime]=mapped_column(DateTime,server_default=func.now(),onupdate=func.now(),comment="更新时间")# 所有的数据表都有 id, created_at, updated_at 字段classBaseModel(Base,TimestampMixin):__abstract__=Trueid:Mapped[int]=mapped_column(BigInteger,primary_key=True,autoincrement=True)步骤 8：通用 Repository 基类 —src/core/base_repository.py
封装常用 CRUD，各模块 Repository 继承即可：
fromtypingimportTypeVar,Generic,Type,Sequencefromsqlalchemyimportselectfromsqlalchemy.ext.asyncioimportAsyncSessionfromsrc.core.base_modelimportBaseModel T=TypeVar("T",bound=BaseModel)classBaseRepository(Generic[T]):def__init__(self,model:Type[T],db:AsyncSession):self.model=model self.db=dbasyncdefget_by_id(self,id:int)->T|None:returnawaitself.db.get(self.model,id)asyncdefget_all(self,offset:int=0,limit:int=100)->Sequence[T]:stmt=select(self.model).offset(offset).limit(limit)result=awaitself.db.execute(stmt)returnresult.scalars().all()asyncdefcreate(self,obj:T)->T:self.db.add(obj)awaitself.db.flush()awaitself.db.refresh(obj)returnobjasyncdefupdate(self,obj:T)->T:awaitself.db.flush()awaitself.db.refresh(obj)returnobjasyncdefdelete(self,obj:T)->None:awaitself.db.delete(obj)awaitself.db.flush()步骤 9：通用响应 Schema —src/core/base_schema.py
fromtypingimportTypeVar,Generic,OptionalfrompydanticimportBaseModel T=TypeVar("T")classResponseSchema(BaseModel,Generic[T]):code:int=200message:str="success"data:Optional[T]=None步骤 10：全局异常处理 —src/core/exceptions.py
fromfastapiimportFastAPI,Requestfromfastapi.responsesimportJSONResponsefromloguruimportloggerclassBizException(Exception):"""业务异常"""def__init__(self,code:int=400,message:str="业务异常"):self.code=code self.message=messagedefregister_exception_handlers(app:FastAPI)->None:@app.exception_handler(BizException)asyncdefbiz_exception_handler(request:Request,exc:BizException):returnJSONResponse(status_code=200,content={"code":exc.code,"message":exc.message,"data":None},)@app.exception_handler(Exception)asyncdefglobal_exception_handler(request:Request,exc:Exception):logger.exception(f"Unhandled exception:{exc}")returnJSONResponse(status_code=500,content={"code":500,"message":"服务器内部错误","data":None},)步骤 11：请求日志中间件 —src/middlewares/logging.py
importtimefromstarlette.middleware.baseimportBaseHTTPMiddlewarefromstarlette.requestsimportRequestfromstarlette.responsesimportResponsefromloguruimportloggerclassLoggingMiddleware(BaseHTTPMiddleware):asyncdefdispatch(self,request:Request,call_next)->Response:start=time.perf_counter()logger.info(f"-->{request.method}{request.url.path}")response=awaitcall_next(request)elapsed=(time.perf_counter()-start)*1000logger.info(f"<--{request.method}{request.url.path}"f"status={response.status_code}{elapsed:.2f}ms")returnresponse步骤 12：应用入口 —src/main.py
fromcontextlibimportasynccontextmanagerfromfastapiimportFastAPIfromloguruimportloggerfromsrc.core.configimportget_settingsfromsrc.core.loggerimportsetup_loggerfromsrc.infra.databaseimportenginefromsrc.core.exceptionsimportregister_exception_handlersfromsrc.middlewares.loggingimportLoggingMiddleware@asynccontextmanagerasyncdeflifespan(app:FastAPI):setup_logger()settings=get_settings()logger.info(f"{settings.APP_NAME}starting | env={settings.APP_ENV}")yieldawaitengine.dispose()logger.info(f"{settings.APP_NAME}shutdown")defcreate_app()->FastAPI:settings=get_settings()app=FastAPI(title=settings.APP_NAME,debug=settings.APP_DEBUG,lifespan=lifespan,)# 异常处理register_exception_handlers(app)# 中间件app.add_middleware(LoggingMiddleware)# 注册模块路由# app.include_router(user_router, prefix="/api/v1")returnapp app=create_app()# 健康检查端点@app.get("/health")asyncdefroot():return{"status":"ok"}新增模块时，只需：
- 在
src/modules/下新建模块目录 - 在
src/main.py中导入并注册路由
步骤 13：配置 Alembic 数据库迁移
🚨 永远不要在迁移脚本中直接改数据
**Alembic是用来改表结构的，不是用来改数据的。如果你需要数据迁移（比如把用户名从两列合并成一列），请在upgrade函数里用op.execute()执行原生 SQL，并且务必写好downgrade **回退脚本。
初始化 Alembic
alembic init-t async alembic这会生成alembic.ini和alembic/目录。
修改alembic.ini
找到sqlalchemy.url行，清空它（我们在env.py中动态设置）：
sqlalchemy.url =修改alembic/env.py
importasynciofromlogging.configimportfileConfigfromsqlalchemyimportpoolfromsqlalchemy.engineimportConnectionfromsqlalchemy.ext.asyncioimportasync_engine_from_configfromalembicimportcontext# 加载 .env 配置fromsrc.core.configimportget_settings# 导入 Base 和所有 Model（确保 Alembic 能发现表结构）fromsrc.core.base_modelimportBase# import src.modules.user.model # noqa: F401 每新增模块在此导入# this is the Alembic Config object, which provides# access to the values within the .ini file in use.config=context.config settings=get_settings()# 动态设置数据库 URLconfig.set_main_option("sqlalchemy.url",settings.DATABASE_URL)# Interpret the config file for Python logging.# This line sets up loggers basically.ifconfig.config_file_nameisnotNone:fileConfig(config.config_file_name)# add your model's MetaData object here# for 'autogenerate' support# from myapp import mymodel# target_metadata = mymodel.Base.metadatatarget_metadata=Base.metadata# other values from the config, defined by the needs of env.py,# can be acquired:# my_important_option = config.get_main_option("my_important_option")# ... etc.defrun_migrations_offline()->None:"""Run migrations in 'offline' mode. This configures the context with just a URL and not an Engine, though an Engine is acceptable here as well. By skipping the Engine creation we don't even need a DBAPI to be available. Calls to context.execute() here emit the given string to the script output. """url=config.get_main_option("sqlalchemy.url")context.configure(url=url,target_metadata=target_metadata,literal_binds=True,dialect_opts={"paramstyle":"named"},)withcontext.begin_transaction():context.run_migrations()defdo_run_migrations(connection:Connection)->None:context.configure(connection=connection,target_metadata=target_metadata)withcontext.begin_transaction():context.run_migrations()asyncdefrun_async_migrations()->None:"""In this scenario we need to create an Engine and associate a connection with the context. """connectable=async_engine_from_config(config.get_section(config.config_ini_section,{}),prefix="sqlalchemy.",poolclass=pool.NullPool,)asyncwithconnectable.connect()asconnection:awaitconnection.run_sync(do_run_migrations)awaitconnectable.dispose()defrun_migrations_online()->None:"""Run migrations in 'online' mode."""asyncio.run(run_async_migrations())ifcontext.is_offline_mode():run_migrations_offline()else:run_migrations_online()生成首次迁移（生成变更脚本）
alembic revision--autogenerate-m"init"执行迁移
alembic upgrade head后续迁移流程
每次修改 Model 后：
# 1. 生成迁移文件alembic revision--autogenerate-m"描述本次变更"# 2. 检查生成的迁移文件（在 alembic/versions/ 下）# 3. 执行迁移alembic upgrade head# 其他常用命令alembic downgrade-1# 回退一个版本alembic current# 查看当前版本alembic history# 查看迁移历史步骤 14：启动项目
# 开发环境快速启动； src 包下必须有 __init__.py 文件fastapi dev src/main.py# 生产环境部署uvicorn src.main:app--host 0.0.0.0--port 8000--workers 4接口文档
http://127.0.0.1:8000/docs
开源项目地址
vue版本：https://gitee.com/belief-team/report
react版本：https://gitee.com/qlsgr/DataReport/
