基于FastApi的介绍与应用-平芜编程栈

FastAPI简介

FastAPI是一个现代、快速（高性能）的Web框架，用于基于Python构建API。它基于标准Python类型提示，使用Starlette和Pydantic库，支持异步请求处理。主要特点包括：

高性能：与NodeJS和Go相当
快速开发：代码自动补全和类型检查
自动生成文档：支持OpenAPI和JSON Schema
易于学习：简洁直观的API设计

安装FastAPI

安装FastAPI和ASGI服务器（如Uvicorn）：

 pip install fastapi uvicorn

基本应用示例

创建一个简单的FastAPI应用：

from fastapi import FastAPI app = FastAPI() @app.get("/") async def read_root(): return {"message": "Hello World"} @app.get("/items/{item_id}") async def read_item(item_id: int, q: str = None): return {"item_id": item_id, "q": q}

运行应用：

uvicorn main:app --reload

请求与响应模型

使用Pydantic模型定义数据结构和验证：

from pydantic import BaseModel class Item(BaseModel): name: str description: str = None price: float tax: float = None @app.post("/items/") async def create_item(item: Item): return item

路径参数与查询参数

@app.get("/users/{user_id}") async def read_user(user_id: int, limit: int = 10, skip: int = 0): return {"user_id": user_id, "limit": limit, "skip": skip}

异步支持

FastAPI原生支持异步请求处理：

import httpx @app.get("/async-example/") async def async_request(): async with httpx.AsyncClient() as client: response = await client.get("https://httpbin.org/get") return response.json()

自动文档

启动应用后访问以下URL查看自动生成的文档：

http://127.0.0.1:8000/docs（交互式Swagger UI）
http://127.0.0.1:8000/redoc（ReDoc文档）

数据库集成示例

使用SQLAlchemy集成数据库：

from sqlalchemy import create_engine, Column, Integer, String from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db" engine = create_engine(SQLALCHEMY_DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) name = Column(String) email = Column(String, unique=True) Base.metadata.create_all(bind=engine) @app.post("/users/") async def create_user(name: str, email: str): db = SessionLocal() user = User(name=name, email=email) db.add(user) db.commit() db.refresh(user) return user

中间件示例

添加自定义中间件：

from fastapi import Request @app.middleware("http") async def add_process_time_header(request: Request, call_next): response = await call_next(request) response.headers["X-Custom-Header"] = "FastAPI" return response