项目架构Conda 隔离环境# 创建环境conda create-nreport-pythonpython3.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日志工具pipinstallfastapi[standard]0.135.1sqlalchemy2.0.48 asyncmy loguru alembic更新 requirements.txtpip freeze requirements.txt步骤 2创建 .env# 应用APP_NAMEMyApp APP_ENVdevelopment APP_DEBUGtrue# MySQLDB_HOST127.0.0.1 DB_PORT3306 DB_USERroot DB_PASSWORDyour_password DB_NAMEmyapp# 日志LOG_LEVELDEBUG LOG_DIRlogs步骤 3创建所有目录mkdir-p src/core src/middlewares src/modules/user步骤 4全局配置 —src/core/config.pyfrompydantic_settingsimportBaseSettingsfromfunctoolsimportlru_cacheclassSettings(BaseSettings):APP_NAME:strMyAppAPP_ENV:strdevelopmentAPP_DEBUG:boolTrueDB_HOST:str127.0.0.1DB_PORT:int3306DB_USER:strrootDB_PASSWORD:strDB_NAME:strmyappLOG_LEVEL:strDEBUGLOG_DIR:strlogspropertydefDATABASE_URL(self)-str:return(fmysqlasyncmy://{self.DB_USER}:{self.DB_PASSWORD}f{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}f?charsetutf8mb4)# 指定环境变量文件model_config{env_file:.env,env_file_encoding:utf-8}# 保存到内存缓存中。以后直接获取。这是一种单例的实现lru_cachedefget_settings()-Settings:returnSettings()步骤 5Loguru 日志配置 —src/core/logger.pyimportsysfrompathlibimportPathfromloguruimportloggerfromsrc.core.configimportget_settingsdefsetup_logger()-None:settingsget_settings()logger.remove()# 控制台logger.add(sys.stdout,levelsettings.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),colorizeTrue,)# 文件log_dirPath(settings.LOG_DIR)log_dir.mkdir(parentsTrue,exist_okTrue)logger.add(str(log_dir/{time:YYYY-MM-DD}.log),levelsettings.LOG_LEVEL,format{time:YYYY-MM-DD HH:mm:ss} | {level: 8} | {name}:{function}:{line} - {message},rotation00:00,retention30 days,compressiongz,encodingutf-8,)步骤 6数据库引擎 Session —src/infra/database.pyInfra属于基础设施层整合。未来redis、mysql、**minio **都在这fromsqlalchemy.ext.asyncioimportcreate_async_engine,async_sessionmaker,AsyncSessionfromsrc.core.configimportget_settings settingsget_settings()enginecreate_async_engine(settings.DATABASE_URL,echosettings.APP_DEBUG,pool_size10,max_overflow20,pool_recycle3600,pool_pre_pingTrue,)AsyncSessionLocalasync_sessionmaker(bindengine,class_AsyncSession,expire_on_commitFalse,)asyncdefget_db()-AsyncSession:FastAPI Depends 注入, 自动提交和异常回滚asyncwithAsyncSessionLocal()assession:try:yieldsessionawaitsession.commit()exceptException:awaitsession.rollback()raise步骤 7ORM 基类 —src/core/base_model.pyfromdatetimeimportdatetimefromsqlalchemyimportBigInteger,DateTime,funcfromsqlalchemy.ormimportDeclarativeBase,Mapped,mapped_columnclassBase(DeclarativeBase):所有 Model 继承此类pass# 创建时间和更新时间由数据库自动维护classTimestampMixin:created_at:Mapped[datetime]mapped_column(DateTime,server_defaultfunc.now(),comment创建时间)updated_at:Mapped[datetime]mapped_column(DateTime,server_defaultfunc.now(),onupdatefunc.now(),comment更新时间)# 所有的数据表都有 id, created_at, updated_at 字段classBaseModel(Base,TimestampMixin):__abstract__Trueid:Mapped[int]mapped_column(BigInteger,primary_keyTrue,autoincrementTrue)步骤 8通用 Repository 基类 —src/core/base_repository.py封装常用 CRUD各模块 Repository 继承即可fromtypingimportTypeVar,Generic,Type,Sequencefromsqlalchemyimportselectfromsqlalchemy.ext.asyncioimportAsyncSessionfromsrc.core.base_modelimportBaseModel TTypeVar(T,boundBaseModel)classBaseRepository(Generic[T]):def__init__(self,model:Type[T],db:AsyncSession):self.modelmodel self.dbdbasyncdefget_by_id(self,id:int)-T|None:returnawaitself.db.get(self.model,id)asyncdefget_all(self,offset:int0,limit:int100)-Sequence[T]:stmtselect(self.model).offset(offset).limit(limit)resultawaitself.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.pyfromtypingimportTypeVar,Generic,OptionalfrompydanticimportBaseModel TTypeVar(T)classResponseSchema(BaseModel,Generic[T]):code:int200message:strsuccessdata:Optional[T]None步骤 10全局异常处理 —src/core/exceptions.pyfromfastapiimportFastAPI,Requestfromfastapi.responsesimportJSONResponsefromloguruimportloggerclassBizException(Exception):业务异常def__init__(self,code:int400,message:str业务异常):self.codecode self.messagemessagedefregister_exception_handlers(app:FastAPI)-None:app.exception_handler(BizException)asyncdefbiz_exception_handler(request:Request,exc:BizException):returnJSONResponse(status_code200,content{code:exc.code,message:exc.message,data:None},)app.exception_handler(Exception)asyncdefglobal_exception_handler(request:Request,exc:Exception):logger.exception(fUnhandled exception:{exc})returnJSONResponse(status_code500,content{code:500,message:服务器内部错误,data:None},)步骤 11请求日志中间件 —src/middlewares/logging.pyimporttimefromstarlette.middleware.baseimportBaseHTTPMiddlewarefromstarlette.requestsimportRequestfromstarlette.responsesimportResponsefromloguruimportloggerclassLoggingMiddleware(BaseHTTPMiddleware):asyncdefdispatch(self,request:Request,call_next)-Response:starttime.perf_counter()logger.info(f--{request.method}{request.url.path})responseawaitcall_next(request)elapsed(time.perf_counter()-start)*1000logger.info(f--{request.method}{request.url.path}fstatus{response.status_code}{elapsed:.2f}ms)returnresponse步骤 12应用入口 —src/main.pyfromcontextlibimportasynccontextmanagerfromfastapiimportFastAPIfromloguruimportloggerfromsrc.core.configimportget_settingsfromsrc.core.loggerimportsetup_loggerfromsrc.infra.databaseimportenginefromsrc.core.exceptionsimportregister_exception_handlersfromsrc.middlewares.loggingimportLoggingMiddlewareasynccontextmanagerasyncdeflifespan(app:FastAPI):setup_logger()settingsget_settings()logger.info(f{settings.APP_NAME}starting | env{settings.APP_ENV})yieldawaitengine.dispose()logger.info(f{settings.APP_NAME}shutdown)defcreate_app()-FastAPI:settingsget_settings()appFastAPI(titlesettings.APP_NAME,debugsettings.APP_DEBUG,lifespanlifespan,)# 异常处理register_exception_handlers(app)# 中间件app.add_middleware(LoggingMiddleware)# 注册模块路由# app.include_router(user_router, prefix/api/v1)returnapp appcreate_app()# 健康检查端点app.get(/health)asyncdefroot():return{status:ok}新增模块时只需在src/modules/下新建模块目录在src/main.py中导入并注册路由步骤 13配置 Alembic 数据库迁移 永远不要在迁移脚本中直接改数据**Alembic是用来改表结构的不是用来改数据的。如果你需要数据迁移比如把用户名从两列合并成一列请在upgrade函数里用op.execute()执行原生 SQL并且务必写好downgrade **回退脚本。初始化 Alembicalembic init-t async alembic这会生成alembic.ini和alembic/目录。修改alembic.ini找到sqlalchemy.url行清空它我们在env.py中动态设置sqlalchemy.url 修改alembic/env.pyimportasynciofromlogging.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.configcontext.config settingsget_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 models MetaData object here# for autogenerate support# from myapp import mymodel# target_metadata mymodel.Base.metadatatarget_metadataBase.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 dont even need a DBAPI to be available. Calls to context.execute() here emit the given string to the script output. urlconfig.get_main_option(sqlalchemy.url)context.configure(urlurl,target_metadatatarget_metadata,literal_bindsTrue,dialect_opts{paramstyle:named},)withcontext.begin_transaction():context.run_migrations()defdo_run_migrations(connection:Connection)-None:context.configure(connectionconnection,target_metadatatarget_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. connectableasync_engine_from_config(config.get_section(config.config_ini_section,{}),prefixsqlalchemy.,poolclasspool.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-minit执行迁移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/reportreact版本https://gitee.com/qlsgr/DataReport/