构件
Pydantic
Pydantic是Python的数据确认库。当在IDE中写代码的时候,Pydantic提供基于标注的类型提示。FastAPI广泛的利用Pydantic建模数据确认、序列化和自动API文档。这些模型都使用标准的Python类型提示,它们提供了一种声明性方式,用来规定到来请求(比如HTTP主体)和外出响应的数据的结构和类型。
from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class Item(BaseModel): name: str price: float is_offer: bool | None = None @app.post("/items/") def create_item(item: Item): # The 'item' object is already validated and typed return {"message": "Item received", "item_name": item.name}
Starlette
Starlette是轻量级ASGI框架/工具箱,它支持Python中的async功能。
Uvicorn
Uvicorn是针对async框架的小型低层服务器/应用Web服务器,它遵从ASGI规定。在技术上,它实现了多进程模型,具有一个主进程,它负责管理工作进程池并将到来的HTTP请求发布给它们。工人进程的数目是预配置的,但也可以在运行时上下调整。
OpenAPI集成
FastAPI为API自动生成OpenAPI文档。这个文档包括Swagger UI和ReDoc二者,它们提供了交互式API文档,人们可以实时探索并测试他们的端点。这对于同其他开发者或用户开发、测试和共享API而言特别有用。Swagger UI可以默认访问于/docs路由而ReDoc在/redoc路由。
特征
异步操作
FastAPI的架构固有的支持异步编程。这种设计允许用单线程的事件循环来有效处理大量的并发请求,特别是在处理I/O密集操作如数据库查询或外部API调用的时候。
依赖注入
FastAPI结合了依赖注入系统,用来管理并向HTTP端点提供服务。这种机制允许开发者将构件比如数据库会话或验证逻辑声明为函数参数。FastAPI为每个请求自动解析这些依赖,注入所需的实例。
from fastapi import Depends, HTTPException, status from db import DbSession # --- Dependency for Database Session --- def get_db(): db = DbSession() try: yield db finally: db.close() @app.post("/items/", status_code=status.HTTP_201_CREATED) def create_item(name: str, description: str, db: DbSession = Depends(get_db)): new_item = Item(name=name, description=description) db.add(new_item) db.commit() db.refresh(new_item) return {"message": "Item created successfully!", "item": new_item} @app.get("/items/{item_id}") def read_item(item_id: int, db: DbSession = Depends(get_db)): item = db.query(Item).filter(Item.id == item_id).first() if item is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found") return item
WebSockets支持
WebSocket允许在客户和服务器之间的全双工通信。这种能力对于要求持续数据交换的应用而言是根本性的,比如即时通讯平台、实时数据仪表盘或多人在线游戏。FastAPI凭借底层Starlette实现,允许对连接和消息处理的有效管理。
# You must have 'websockets' package installed 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(f"Message text was: {data}")
后台任务
FastAPI确使在已经将HTTP响应发送给客户之后能执行后台任务。这允许API立即响应用户请求,却同时在后台处理非紧要或耗时的操作。典型应用包括发送邮件通知,更新缓存或进行数据后处理。
import time import shutil from fastapi import BackgroundTasks, UploadFile, File from utils import generate_thumbnail @app.post("/upload-image/") async def upload_image( image: UploadFile = File(...), background_tasks: BackgroundTasks): file_location = f"uploaded_images/{image.filename}" # Save uploaded image with open(image_path, "wb") as f: contents = await file.read() f.write(contents) # Add thumbnail generation as a background task background_tasks.add_task(generate_thumbnail, file_location, "200x200") return {"message": f"Image '{image.filename}' uploaded. Thumbnail generation started in background."}
例子
下述代码展示了一个简单的Web应用,它在被访问时显示"Hello, World!":
# Import FastAPI class from the fastapi package from fastapi import FastAPI # Create an instance of the FastAPI app app = FastAPI() # Define a GET route for the root URL ("/") @app.get("/") async def read_root() -> str: # Return a plain text response return "Hello, World!"