1. 为什么选择FastAPIMySQL组合我第一次用FastAPI连接MySQL做项目是在两年前的一个文件管理系统上。当时团队需要快速搭建一个支持高并发的API服务经过技术选型对比最终选择了这个组合。两年用下来这套技术栈确实没让我失望。FastAPI的异步特性是最大亮点。传统同步框架比如Flask在处理数据库请求时每个请求都要等待MySQL返回结果后才能处理下一个。而FastAPI配合async/await语法可以在等待数据库响应时先去处理其他请求。实测下来同样的服务器配置QPS每秒查询数能提升3-5倍。举个例子我们有个文件列表接口在同步模式下峰值只能扛住800请求/秒改造成异步后轻松突破3000。MySQL的优势则在于它的全能性。不像某些NoSQL数据库只擅长特定场景MySQL既能满足我们存储文件元数据名称、类型、大小等结构化数据的需求又支持事务保证数据一致性。有一次我们系统遇到批量插入时部分失败的情况正是靠MySQL的事务回滚功能避免了数据错乱。安装过程简单到令人发指。只需要5个命令就能跑起来pip install fastapi uvicorn sqlalchemy pymysql mysql -u root -p -e CREATE DATABASE file_manager2. 从零搭建项目骨架2.1 项目结构设计我习惯用分层架构组织代码推荐这样安排目录/file_manager ├── api/ # 路由层 │ └── v1/ # 版本控制 ├── models/ # 数据模型 ├── schemas/ # Pydantic模型 ├── services/ # 业务逻辑 ├── database.py # 数据库连接 └── main.py # 启动入口这种结构在后期功能扩展时会非常清爽。比如要加个用户系统只需要在api/v1下新建user.py在models添加User类即可。2.2 数据库连接最佳实践新手最容易犯的错误是直接在路由里写SQL。正确的做法是用SQLAlchemy这样的ORM工具管理连接。这是我的database.py配置from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker SQLALCHEMY_DATABASE_URL mysqlpymysql://user:passwordlocalhost:3306/file_manager engine create_engine( SQLALCHEMY_DATABASE_URL, pool_size20, # 连接池大小 max_overflow50, # 最大溢出连接数 pool_pre_pingTrue # 自动检测连接有效性 ) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) Base declarative_base()关键参数说明pool_size建议设置为最大并发量的1.5倍max_overflow突发流量时的缓冲连接数pool_pre_ping避免拿到已断开的连接3. 实现核心CRUD功能3.1 文件上传接口设计文件上传要考虑三个维度元数据存储MySQL文件本体存储本地/S3事务一致性这是我的实现方案app.post(/files) async def upload_file( file: UploadFile File(...), db: Session Depends(get_db) ): # 生成唯一文件名 file_name f{uuid.uuid4()}{Path(file.filename).suffix} file_path fuploads/{file_name} try: # 开启事务 db.begin() # 存储文件 with open(file_path, wb) as buffer: shutil.copyfileobj(file.file, buffer) # 记录元数据 db_file models.File( namefile.filename, stored_namefile_name, sizeos.path.getsize(file_path), content_typefile.content_type ) db.add(db_file) db.commit() return {id: db_file.id} except Exception as e: db.rollback() # 清理已上传文件 if os.path.exists(file_path): os.remove(file_path) raise HTTPException(status_code500, detailstr(e))3.2 高级查询技巧分页查询有几种实现方式性能对比方法10万数据耗时优点缺点LIMIT/OFFSET320ms实现简单大数据量慢游标分页45ms性能稳定不支持随机跳页覆盖索引80ms折中方案需要特定索引推荐使用游标分页的写法app.get(/files) async def list_files( last_id: int Query(0), limit: int Query(50), db: Session Depends(get_db) ): query db.query(models.File) if last_id 0: query query.filter(models.File.id last_id) files query.order_by(models.File.id).limit(limit).all() return { data: [{id: f.id, name: f.name} for f in files], next_cursor: files[-1].id if files else None }4. 性能优化实战4.1 批量插入的三种姿势测试数据插入1万条标签记录单条插入耗时98秒executemany耗时3.2秒LOAD DATA INFILE耗时0.8秒重点说下最优方案app.post(/labels/bulk) async def bulk_create_labels( labels: List[schemas.LabelCreate], db: Session Depends(get_db) ): # 生成临时CSV文件 tmp_path /tmp/labels.csv with open(tmp_path, w) as f: writer csv.writer(f) for label in labels: writer.writerow([label.name, label.value]) # 使用MySQL原生加载命令 with db.connection().connection.cursor() as cursor: cursor.execute(f LOAD DATA LOCAL INFILE {tmp_path} INTO TABLE labels FIELDS TERMINATED BY , (name, value) ) db.commit() os.unlink(tmp_path) return {count: len(labels)}注意需要给MySQL用户添加FILE权限并在连接字符串加上local_infile1参数4.2 连接池调优经验分享几个实战中总结的参数配置连接数公式最大连接数 (核心数 * 2) 有效磁盘数比如4核CPUSSD服务器(4*2)19监控指标SHOW STATUS LIKE Threads_connected; SHOW STATUS LIKE Max_used_connections;超时设置engine create_engine( ..., pool_recycle3600, # 1小时回收连接 pool_timeout30, # 获取连接超时时间 )5. 错误处理与调试技巧5.1 自定义异常处理FastAPI的异常处理中间件非常实用from fastapi import FastAPI, HTTPException from fastapi.exceptions import RequestValidationError app FastAPI() app.exception_handler(HTTPException) async def custom_http_exception_handler(request, exc): return JSONResponse( status_codeexc.status_code, content{ code: exc.status_code, message: exc.detail, data: None } ) app.exception_handler(RequestValidationError) async def validation_exception_handler(request, exc): return JSONResponse( status_code422, content{ code: 422, message: 参数校验失败, errors: exc.errors() } )5.2 性能问题排查遇到慢查询时我常用的诊断步骤开启MySQL慢查询日志SET GLOBAL slow_query_log ON; SET GLOBAL long_query_time 1;使用EXPLAIN分析EXPLAIN SELECT * FROM files WHERE type image;查看索引使用情况SHOW INDEX FROM files;最近遇到一个真实案例分页查询突然变慢EXPLAIN发现没走索引。原因是使用了WHERE type LIKE %image%导致索引失效改成WHERE type image后查询时间从1200ms降到15ms。6. 安全防护措施6.1 SQL注入防护虽然ORM已经帮我们处理了大部分风险但直接写SQL时仍需注意错误示范query fSELECT * FROM files WHERE name {filename}正确做法# 使用参数化查询 query SELECT * FROM files WHERE name %s cursor.execute(query, (filename,))6.2 敏感数据保护文件管理系统尤其要注意存储路径不要使用原始文件名文件下载时检查权限app.get(/files/{file_id}/download) async def download_file( file_id: int, user: User Depends(get_current_user), db: Session Depends(get_db) ): file db.query(models.File).get(file_id) if not file: raise HTTPException(404) if not check_permission(user, file): raise HTTPException(403) return FileResponse( file.stored_path, filenamefile.original_name )7. 项目部署建议7.1 生产环境配置我的标准部署方案Web服务器Uvicorn Gunicorngunicorn -w 4 -k uvicorn.workers.UvicornWorker main:appMySQL配置[mysqld] max_connections 200 innodb_buffer_pool_size 1G7.2 监控指标必备的监控项API响应时间P99 500ms数据库连接数使用率80%慢查询数量5次/分钟推荐使用PrometheusGranfa搭建监控看板关键指标from prometheus_fastapi_instrumentator import Instrumentator Instrumentator().instrument(app).expose(app)这套技术栈在我们生产环境已经稳定运行两年支撑日均百万级API调用。最大的体会是前期把数据库设计和连接管理做好后期扩展功能会非常顺畅。最近我们正在用同样的架构开发标签管理系统代码复用率超过70%。