FastAPI：高性能Python Web框架详解
1. FastAPI框架概述FastAPI是一个用于构建API的现代、快速高性能的Python Web框架基于标准Python类型提示。它由Sebastián Ramírez又名tiangolo创建并于2018年首次发布。作为一个相对年轻的框架FastAPI迅速获得了Python社区的广泛认可成为构建API的首选工具之一。这个框架的设计哲学是快速开发高性能运行。它结合了Starlette用于Web部分和Pydantic用于数据部分两个强大的库提供了出色的性能表现。根据TechEmpower的基准测试FastAPI的性能在Python Web框架中名列前茅仅次于其底层依赖Starlette和Uvicorn。提示FastAPI的名称反映了它的两个核心特点——Fast代表高性能API则表明它专注于API开发。不过实际上它也可以用于构建完整的Web应用。2. FastAPI的核心特性2.1 极致的性能表现FastAPI的性能是其最引人注目的特点之一。它基于异步编程模型ASGI使用现代的Python异步特性async/await能够处理大量并发请求而不会阻塞。在实际测试中FastAPI的性能与Node.js和Go等以高性能著称的语言/框架相当。性能优势主要来自三个方面基于Starlette的异步处理能力Pydantic的高效数据验证和序列化自动化的请求/响应处理减少了不必要的开销2.2 开发效率提升FastAPI通过多种方式显著提高了开发效率自动生成API文档支持Swagger UI和ReDoc基于Python类型提示的自动数据验证编辑器自动补全和类型检查支持简洁直观的API设计根据官方数据使用FastAPI可以提升200%-300%的开发速度同时减少约40%的人为错误。2.3 强大的类型系统FastAPI充分利用了Python的类型提示系统这使得它能够在开发时提供更好的编辑器支持如VS Code、PyCharm等自动验证请求和响应数据生成精确的API文档减少运行时类型相关的错误类型系统不仅支持基本类型int, str, bool等还支持复杂的数据结构和嵌套模型。3. FastAPI与其他框架的对比3.1 与Django的比较Django是一个全功能的Web框架而FastAPI更专注于API开发。主要区别包括Django包含ORM、模板引擎等完整功能FastAPI则更轻量Django使用同步模型FastAPI基于异步Django有更丰富的内置功能FastAPI更灵活可扩展Django的学习曲线更陡峭FastAPI更易上手3.2 与Flask的比较Flask是一个微框架与FastAPI有更多相似之处但关键区别在于Flask是同步框架FastAPI是异步的Flask没有内置的数据验证和文档生成Flask的扩展系统更成熟但FastAPI的集成更紧密Flask的社区更大但FastAPI的增长速度更快3.3 与Starlette的关系Starlette是FastAPI的底层Web框架提供了核心的HTTP功能。FastAPI在Starlette基础上添加了数据验证和序列化通过Pydantic自动API文档生成更高级的路由和依赖注入系统简化的开发体验4. FastAPI的核心组件4.1 Pydantic模型Pydantic是FastAPI处理数据的核心。它允许你定义数据模型这些模型将自动用于请求体验证响应数据序列化文档生成一个典型的Pydantic模型定义如下from pydantic import BaseModel class Item(BaseModel): name: str description: str | None None price: float tax: float | None None4.2 路由系统FastAPI的路由系统简洁而强大。你可以使用装饰器定义路由和处理函数from fastapi import FastAPI app FastAPI() app.get(/items/{item_id}) async def read_item(item_id: int): return {item_id: item_id}路由支持路径参数、查询参数、请求体等多种参数类型并且会自动处理类型转换和验证。4.3 依赖注入系统FastAPI的依赖注入系统是其最强大的功能之一。它允许你声明和管理组件之间的依赖关系使代码更模块化和可测试from fastapi import Depends, FastAPI app FastAPI() def common_parameters(q: str | None None, skip: int 0, limit: int 100): return {q: q, skip: skip, limit: limit} app.get(/items/) async def read_items(commons: dict Depends(common_parameters)): return commons5. FastAPI的安装与基本使用5.1 安装FastAPI安装FastAPI及其标准依赖包括Uvicorn服务器pip install fastapi[standard]5.2 创建第一个应用创建一个简单的FastAPI应用只需要几行代码from fastapi import FastAPI app FastAPI() app.get(/) async def root(): return {message: Hello World}5.3 运行应用使用Uvicorn运行应用uvicorn main:app --reload其中main是Python文件名不含.pyapp是FastAPI实例的名称--reload启用开发时的自动重载功能5.4 访问API文档FastAPI会自动生成交互式API文档Swagger UI: http://127.0.0.1:8000/docsReDoc: http://127.0.0.1:8000/redoc6. FastAPI的高级特性6.1 异步支持FastAPI全面支持Python的async/await语法可以轻松编写异步处理函数app.get(/items/{item_id}) async def read_item(item_id: int): # 这里可以调用其他异步函数 item await get_item_from_db(item_id) return item6.2 安全性FastAPI内置了多种安全功能包括OAuth2带密码和Bearer令牌JWT令牌HTTP基本认证CORS跨域资源共享6.3 后台任务对于不需要立即返回结果的操作可以使用后台任务from fastapi import BackgroundTasks def write_notification(email: str, message): with open(log.txt, modew) as email_file: content fnotification for {email}: {message} email_file.write(content) app.post(/send-notification/{email}) async def send_notification(email: str, background_tasks: BackgroundTasks): background_tasks.add_task(write_notification, email, messagesome notification) return {message: Notification sent in the background}6.4 WebSocket支持FastAPI支持WebSocket协议可以轻松构建实时应用from fastapi import WebSocket app.websocket(/ws) async def websocket_endpoint(websocket: WebSocket): await websocket.accept() while True: data await websocket.receive_text() await websocket.send_text(fMessage text was: {data})7. FastAPI的项目结构7.1 基本项目结构一个典型的FastAPI项目可能包含以下目录结构my_project/ ├── app/ │ ├── __init__.py │ ├── main.py # 主应用文件 │ ├── models.py # Pydantic模型 │ ├── schemas.py # 数据模式 │ ├── crud.py # 数据库操作 │ ├── database.py # 数据库连接 │ ├── dependencies.py # 依赖项 │ └── routers/ # 路由模块 │ ├── __init__.py │ ├── items.py │ └── users.py ├── tests/ # 测试代码 ├── requirements.txt # 依赖列表 └── README.md7.2 大型应用组织对于大型应用推荐使用APIRouter将路由分组# routers/items.py from fastapi import APIRouter router APIRouter() router.get(/items/) async def read_items(): return [{name: Item 1}, {name: Item 2}] # main.py from fastapi import FastAPI from .routers import items app FastAPI() app.include_router(items.router, prefix/api)8. FastAPI的部署8.1 使用Uvicorn生产部署生产环境建议使用Uvicorn配合Gunicorngunicorn -w 4 -k uvicorn.workers.UvicornWorker app.main:app8.2 容器化部署使用Docker部署FastAPI应用的示例DockerfileFROM python:3.9 WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [uvicorn, app.main:app, --host, 0.0.0.0, --port, 80]8.3 云平台部署FastAPI可以部署到各种云平台包括AWS (使用ECS或Lambda)Google Cloud RunAzure App ServiceHerokuFastAPI Cloud官方托管服务9. FastAPI的测试9.1 使用TestClientFastAPI提供了TestClient用于测试API端点from fastapi.testclient import TestClient from .main import app client TestClient(app) def test_read_item(): response client.get(/items/42) assert response.status_code 200 assert response.json() {item_id: 42}9.2 异步测试对于异步代码可以使用pytest-asyncioimport pytest from httpx import AsyncClient from .main import app pytest.mark.asyncio async def test_read_item(): async with AsyncClient(appapp, base_urlhttp://test) as ac: response await ac.get(/items/42) assert response.status_code 200 assert response.json() {item_id: 42}10. FastAPI的生态系统10.1 相关工具FastAPI生态系统包含多个相关工具TyperFastAPI作者开发的CLI应用框架SQLModel结合SQLAlchemy和Pydantic的ORMFastAPI Users用户认证系统FastAPI Cache缓存扩展10.2 社区资源FastAPI拥有活跃的社区和丰富的学习资源官方文档多语言支持GitHub仓库活跃的issue讨论Discord和Gitter社区各种教程和视频课程10.3 企业采用许多知名公司已经采用FastAPI包括Uber用于Ludwig机器学习工具Netflix用于Dispatch危机管理框架Microsoft内部工具和服务CiscoAPI优先开发策略11. FastAPI的最佳实践11.1 项目组织按功能而非类型组织代码使用APIRouter模块化路由分离业务逻辑和路由处理使用依赖注入管理共享资源11.2 性能优化合理使用异步I/O操作实现缓存策略优化数据库查询使用ORJSONResponse提高JSON序列化速度11.3 安全实践始终验证输入数据使用HTTPS实施适当的认证和授权限制请求速率保护敏感信息12. FastAPI的局限性尽管FastAPI有许多优点但也存在一些限制对同步代码的支持不如异步代码完善某些高级数据库功能可能需要额外工作模板和前端支持不如全栈框架强大生态系统相比Django/Flask仍较小13. 学习FastAPI的路径13.1 初学者路径学习Python基础特别是类型提示了解HTTP和REST基础完成官方教程构建简单的CRUD API添加认证和数据库支持13.2 进阶学习深入理解Pydantic模型掌握依赖注入系统学习高级安全特性优化应用性能探索WebSocket和SSE13.3 生产准备实现完善的错误处理配置日志和监控设置CI/CD流程实施自动化测试规划扩展策略14. FastAPI的未来发展FastAPI仍在快速发展中未来可能的方向包括更强大的数据验证功能改进的开发者工具更好的前端集成增强的云部署体验更丰富的生态系统15. 常见问题解答15.1 FastAPI适合大型项目吗是的FastAPI完全适合大型项目。它的模块化设计和依赖注入系统使得大型应用可以保持良好的组织结构。许多知名公司已经在生产环境中使用FastAPI构建复杂系统。15.2 FastAPI的学习曲线如何如果你已经熟悉Python和Web开发基础FastAPI的学习曲线相对平缓。它的API设计直观文档完善特别是对于已经使用过类似框架如Flask的开发者来说更容易上手。15.3 FastAPI如何处理数据库FastAPI本身不包含ORM但可以与任何Python数据库工具配合使用如SQLAlchemy、Tortoise-ORM、Databases等。SQLModel是专门为FastAPI设计的ORM替代方案结合了SQLAlchemy和Pydantic的优点。15.4 FastAPI的性能真的那么好吗是的独立基准测试表明FastAPI的性能确实非常出色。这主要得益于其基于Starlette的异步设计和Pydantic的高效数据处理。对于I/O密集型应用性能优势尤为明显。15.5 如何调试FastAPI应用FastAPI应用可以使用标准的Python调试工具使用uvicorn的--reload标志进行开发时自动重载集成pdb或ipdb进行断点调试使用logging模块记录日志利用FastAPI的异常处理返回详细错误信息16. 实际案例构建一个待办事项API16.1 项目设置首先创建项目结构mkdir todo_api cd todo_api python -m venv venv source venv/bin/activate # Linux/Mac pip install fastapi[standard]16.2 定义模型创建models.pyfrom pydantic import BaseModel from typing import Optional class Todo(BaseModel): title: str description: Optional[str] None completed: bool False16.3 创建路由编写main.pyfrom fastapi import FastAPI, HTTPException from models import Todo app FastAPI() todos [] app.post(/todos/) async def create_todo(todo: Todo): todos.append(todo) return todo app.get(/todos/) async def read_todos(): return todos app.get(/todos/{todo_id}) async def read_todo(todo_id: int): if todo_id len(todos): raise HTTPException(status_code404, detailTodo not found) return todos[todo_id]16.4 运行和测试启动服务器uvicorn main:app --reload测试APIcurl -X POST http://localhost:8000/todos/ -H Content-Type: application/json -d {title:Learn FastAPI,description:Study the documentation}17. FastAPI中的错误处理17.1 基本错误处理FastAPI提供了HTTPException来处理错误情况from fastapi import FastAPI, HTTPException app FastAPI() items {foo: The Foo Wrestlers} app.get(/items/{item_id}) async def read_item(item_id: str): if item_id not in items: raise HTTPException(status_code404, detailItem not found) return {item: items[item_id]}17.2 自定义异常处理器你可以注册自定义的异常处理器from fastapi import FastAPI, Request from fastapi.responses import JSONResponse app FastAPI() class UnicornException(Exception): def __init__(self, name: str): self.name name app.exception_handler(UnicornException) async def unicorn_exception_handler(request: Request, exc: UnicornException): return JSONResponse( status_code418, content{message: fOops! {exc.name} did something. There goes a rainbow...}, ) app.get(/unicorns/{name}) async def read_unicorn(name: str): if name yolo: raise UnicornException(namename) return {unicorn_name: name}17.3 验证错误FastAPI会自动处理数据验证错误返回详细的错误信息。例如当请求体不符合定义的模型时会返回422状态码和错误详情。18. FastAPI与前端集成18.1 提供静态文件FastAPI可以方便地提供静态文件from fastapi.staticfiles import StaticFiles app.mount(/static, StaticFiles(directorystatic), namestatic)18.2 使用模板集成Jinja2模板引擎from fastapi import Request from fastapi.templating import Jinja2Templates templates Jinja2Templates(directorytemplates) app.get(/items/{id}) async def read_item(request: Request, id: str): return templates.TemplateResponse(item.html, {request: request, id: id})18.3 CORS设置处理跨域请求from fastapi.middleware.cors import CORSMiddleware app.add_middleware( CORSMiddleware, allow_origins[*], allow_credentialsTrue, allow_methods[*], allow_headers[*], )19. FastAPI的扩展性19.1 自定义中间件创建自定义中间件import time from fastapi import Request app.middleware(http) async def add_process_time_header(request: Request, call_next): start_time time.time() response await call_next(request) process_time time.time() - start_time response.headers[X-Process-Time] str(process_time) return response19.2 自定义响应创建自定义响应类from fastapi import FastAPI from fastapi.responses import JSONResponse app FastAPI() class Message(BaseModel): message: str app.get(/message) async def read_message(): return JSONResponse(content{message: Hello World}, status_code200)19.3 高级依赖注入复杂的依赖注入示例from fastapi import Depends, FastAPI, Header, HTTPException app FastAPI() async def verify_token(x_token: str Header(...)): if x_token ! fake-super-secret-token: raise HTTPException(status_code400, detailX-Token header invalid) async def verify_key(x_key: str Header(...)): if x_key ! fake-super-secret-key: raise HTTPException(status_code400, detailX-Key header invalid) return x_key app.get(/items/, dependencies[Depends(verify_token), Depends(verify_key)]) async def read_items(): return [{item: Foo}, {item: Bar}]20. FastAPI在生产环境中的实践20.1 配置管理使用环境变量进行配置from pydantic import BaseSettings class Settings(BaseSettings): app_name: str Awesome API admin_email: str items_per_user: int 50 class Config: env_file .env settings Settings()20.2 日志记录配置日志记录import logging from fastapi import FastAPI logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s ) logger logging.getLogger(__name__) app FastAPI() app.get(/) async def root(): logger.info(Root endpoint accessed) return {message: Hello World}20.3 监控和指标集成Prometheus监控from fastapi import FastAPI from starlette_exporter import PrometheusMiddleware, handle_metrics app FastAPI() app.add_middleware(PrometheusMiddleware) app.add_route(/metrics, handle_metrics) app.get(/) async def root(): return {message: Hello World}21. FastAPI与数据库集成21.1 使用SQLAlchemy集成SQLAlchemy ORMfrom sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker SQLALCHEMY_DATABASE_URL sqlite:///./sql_app.db engine create_engine( SQLALCHEMY_DATABASE_URL, connect_args{check_same_thread: False} ) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) Base declarative_base() # 依赖项 def get_db(): db SessionLocal() try: yield db finally: db.close()21.2 使用Tortoise-ORM异步ORM集成from tortoise import fields, models from tortoise.contrib.fastapi import register_tortoise class User(models.Model): id fields.IntField(pkTrue) username fields.CharField(max_length255) class Meta: table users register_tortoise( app, db_urlsqlite://db.sqlite3, modules{models: [app.models]}, generate_schemasTrue, add_exception_handlersTrue, )21.3 使用SQLModelSQLAlchemy和Pydantic的结合from sqlmodel import Field, Session, SQLModel, create_engine class Hero(SQLModel, tableTrue): id: int | None Field(defaultNone, primary_keyTrue) name: str secret_name: str age: int | None None engine create_engine(sqlite:///database.db) SQLModel.metadata.create_all(engine) def get_session(): with Session(engine) as session: yield session22. FastAPI与微服务架构22.1 服务间通信使用httpx进行服务间调用import httpx from fastapi import FastAPI app FastAPI() app.get(/call-service-b) async def call_service_b(): async with httpx.AsyncClient() as client: response await client.get(http://service-b:8000/data) return response.json()22.2 事件驱动架构集成消息队列如Redisimport redis from fastapi import FastAPI app FastAPI() r redis.Redis(hostredis, port6379) app.post(/publish/{channel}) async def publish(channel: str, message: str): r.publish(channel, message) return {status: Message published}22.3 服务发现集成Consul服务发现from consul import Consul from fastapi import FastAPI app FastAPI() consul Consul() def register_service(): consul.agent.service.register( my-service, service_idmy-service-1, address127.0.0.1, port8000, check{ HTTP: http://127.0.0.1:8000/health, interval: 10s } )23. FastAPI的性能优化技巧23.1 数据库查询优化使用异步数据库驱动实现查询缓存优化N1查询问题使用selectinload进行关系加载23.2 响应优化使用ORJSONResponse加速JSON序列化实现响应压缩启用HTTP/2使用流式响应处理大文件23.3 并发处理合理设置Uvicorn工作进程数使用背景任务处理非关键操作实现请求限流使用连接池管理数据库连接24. FastAPI的安全最佳实践24.1 认证实现使用OAuth2和JWTfrom fastapi import Depends, FastAPI, HTTPException, status from fastapi.security import OAuth2PasswordBearer app FastAPI() oauth2_scheme OAuth2PasswordBearer(tokenUrltoken) app.get(/items/) async def read_items(token: str Depends(oauth2_scheme)): return {token: token}24.2 输入验证始终验证所有输入参数使用Pydantic的正则表达式验证实现自定义验证器限制输入数据大小24.3 安全头部添加安全相关的HTTP头部from fastapi import FastAPI from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware from fastapi.middleware.trustedhost import TrustedHostMiddleware app FastAPI() app.add_middleware(HTTPSRedirectMiddleware) app.add_middleware( TrustedHostMiddleware, allowed_hosts[example.com, *.example.com] )25. FastAPI的调试技巧25.1 交互式调试使用pdb进行调试app.get(/debug) async def debug_endpoint(): import pdb; pdb.set_trace() # 设置断点 return {message: Debugging}25.2 请求检查检查请求详细信息from fastapi import Request app.get(/request-info) async def request_info(request: Request): return { method: request.method, headers: dict(request.headers), client: request.client }25.3 性能分析使用cProfile进行性能分析import cProfile from fastapi import FastAPI, Request app FastAPI() app.middleware(http) async def profile_request(request: Request, call_next): profiler cProfile.Profile() profiler.enable() response await call_next(request) profiler.disable() profiler.dump_stats(profile.prof) return response26. FastAPI的测试策略26.1 单元测试测试单个路由函数from fastapi.testclient import TestClient from .main import app, read_item client TestClient(app) def test_read_item(): response client.get(/items/42) assert response.status_code 200 assert response.json() {item_id: 42}26.2 集成测试测试整个应用流程async def test_create_and_read_item(): # 创建项目 create_response client.post(/items/, json{name: Test Item}) assert create_response.status_code 200 item_id create_response.json()[id] # 读取项目 read_response client.get(f/items/{item_id}) assert read_response.status_code 200 assert read_response.json()[name] Test Item26.3 模拟依赖使用依赖覆盖进行测试from fastapi import FastAPI from fastapi.testclient import TestClient from .dependencies import get_db app FastAPI() def override_get_db(): try: db TestingSessionLocal() yield db finally: db.close() app.dependency_overrides[get_db] override_get_db client TestClient(app)27. FastAPI的文档扩展27.1 自定义OpenAPI模式修改OpenAPI文档from fastapi import FastAPI app FastAPI() app.openapi_schema { openapi: 3.0.2, info: { title: My API, version: 1.0.0, description: Custom API description }, paths: app.openapi_schema[paths] }27.2 添加API标签组织API端点app.post( /items/, response_modelItem, tags[items], summaryCreate an item, descriptionCreate an item with all the information ) async def create_item(item: Item): return item27.3 响应示例定义响应示例from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class Item(BaseModel): name: str description: str | None None class Config: schema_extra { example: { name: Foo, description: A very nice Item } }28. FastAPI的部署架构28.1 单服务器部署基本架构Uvicorn作为ASGI服务器Gunicorn作为进程管理器Nginx作为反向代理28.2 高可用部署生产级架构多应用服务器负载均衡独立的数据库服务器Redis缓存层监控和日志收集系统28.3 无服务器部署使用Serverless平台AWS LambdaGoogle Cloud FunctionsAzure FunctionsVercel29. FastAPI的社区资源29.1 官方资源官方文档https://fastapi.tiangolo.comGitHub仓库https://github.com/tiangolo/fastapi官方Discord频道29.2 学习资源FastAPI官方教程测试驱动开发FastAPI应用书籍Real Python的FastAPI教程YouTube上的FastAPI视频教程29.3 第三方扩展FastAPI Users用户认证系统FastAPI Cache缓存支持FastAPI Limiter请求限流FastAPI Mail电子邮件支持30. 从其他框架迁移到FastAPI30.1 从Flask迁移主要变化同步到异步的转变路由定义方式类似但更类型安全需要学习Pydantic模型依赖注入系统替代Flask的上下文30.2 从Django迁移注意事项FastAPI没有内置的ORM和admin界面认证系统需要重新实现模板支持较弱更适合API而非全栈应用30.3 从Node.js迁移对比优势Python类型系统的优势相似的性能表现更简洁的代码结构更好的开发体验编辑器支持31. FastAPI的常见陷阱与解决方案31.1 异步混淆常见错误在异步上下文中调用阻塞代码解决方案使用asyncio.to_thread包装阻塞调用或者使用专门的异步库替代阻塞库31.2 依赖管理常见错误过度复杂的依赖关系解决方案保持依赖简单明了使用依赖覆盖进行测试避免深层嵌套的依赖31.3 数据库会话常见错误会话管理不当导致连接泄漏解决方案始终使用依赖项中的yield模式或者在finally块中明确关闭会话32. FastAPI的未来展望32.1 即将到来的特性根据开发路线图未来可能包括更强大的数据验证功能改进的开发者工具更好的前端集成增强的云部署体验32.2 生态系统发展FastAPI生态系统正在快速增长预计会有更多高质量的扩展更好的ORM集成更丰富的模板和示例企业级功能支持32.3 社区成长随着采用率提高可以期待更多的会议和聚会更完善的文档和翻译更多的就业机会更活跃的贡献者社区33. FastAPI在企业中的应用案例33.1 Uber的LudwigUber使用FastAPI构建了Ludwig的预测服务这是一个声明式的深度学习框架。FastAPI被选为其REST服务器的基础因为它能够快速构建高性能API同时保持代码简洁。33.2 Netflix的DispatchNetflix开源了其危机管理编排框架Dispatch该框架使用FastAPI作为Web层。Netflix团队特别赞赏FastAPI的性能和易用性以及它如何简化了复杂API的开发。33.3 Microsoft的内部工具Microsoft的一些团队正在将FastAPI用于机器学习服务和内部工具。开发者特别提到FastAPI的编辑器支持和类型安全是选择它的关键因素。34. FastAPI与机器学习34.1 模型服务化FastAPI是部署机器学习模型的理想选择from fastapi import FastAPI from pydantic import BaseModel import joblib app FastAPI() model joblib.load(model.pkl) class PredictionInput(BaseModel): feature1: float feature2: float app.post(/predict) async def predict(input: PredictionInput): prediction model.predict([[input.feature1, input.feature2]]) return {prediction: prediction[0]}34.2 批处理支持处理批量预测请求from typing import List class BatchPredictionInput(BaseModel): items: List[PredictionInput] app.post(/batch-predict) async def batch_predict(inputs: BatchPredictionInput): features [[item.feature1, item.feature2] for item in inputs.items] predictions model.predict(features) return {predictions: predictions.tolist()}34.3 模型监控集成模型性能监控from prometheus_client import Counter PREDICTION_COUNTER Counter( model_predictions_total, Total number of predictions, [model_name, status] ) app.post(/predict) async def predict(input: PredictionInput): try: prediction model.predict([[input.feature1, input.feature2]]) PREDICTION_COUNTER.labels(model_namemy_model, statussuccess).inc() return {prediction: prediction[0]} except Exception: PREDICTION_COUNTER.labels(model_namemy_model, statusfailure).inc() raise35. FastAPI与GraphQL35.1 Strawberry集成使用Strawberry添加GraphQL支持from fastapi import FastAPI from strawberry.fastapi import GraphQLRouter import strawberry strawberry.type class Query: strawberry.field def hello(self) - str: return Hello World schema strawberry.Schema(Query) graphql_app GraphQLRouter(schema)
