Python中Web开发-FastAPI框架

Python中Web开发-FastAPI框架-LMLPHP

大家好，在当今Web开发领域，高性能、易用性和可扩展性是开发者们追求的目标。Python作为一种流行的编程语言，在Web开发领域也有着强大的影响力。而在众多的Python Web框架中，FastAPI凭借其快速、现代和易用的特性，成为了开发者们的首选之一。本文将深入探讨FastAPI框架的特性、用法和实践。

FastAPI官网：https://fastapi.tiangolo.com/

FastAPI开发文档：https://devdocs.io/fastapi/

一、FastAPI框架介绍

FastAPI是一个基于Python 3.7+的现代Web框架，其设计理念是快速、简单且高效。它结合了Python的强大特性和类型注解的优势，使得开发者能够编写出高性能、易维护和类型安全的Web应用程序。FastAPI还提供了自动生成交互式API文档、自动验证请求数据和异步处理等先进功能，使得开发者能够更轻松地构建现代化的Web服务。

优点：

快速开发：FastAPI 基于 Python 类型提示和异步编程，可以快速地开发出高性能的 Web 服务。它具有直观的 API 设计和自动文档生成功能，可以大大加速开发速度。
性能优秀：FastAPI 基于 Starlette 框架，使用了高性能的 ASGI 服务器，支持异步请求处理，可以处理大量并发请求而不降低性能。
自动文档生成：FastAPI 支持自动生成 API 文档，并提供交互式的 API 文档界面，让开发者可以方便地查看 API 接口的说明和使用示例。
数据验证：FastAPI 集成了 Pydantic 库，可以对请求数据进行自动验证和转换，确保数据的类型安全和有效性，减少了开发过程中的错误和异常情况。
异步支持：FastAPI 支持异步请求处理和异步任务处理，可以使用 async/await 关键字来定义异步函数，提高了 Web 服务的性能和吞吐量。
安全性：FastAPI 提供了丰富的安全性配置功能，包括身份验证、授权、HTTPS 支持、CORS 配置等，可以保护 Web 应用程序的安全性和可靠性。
生态系统丰富：FastAPI 是一个活跃的开源项目，拥有庞大的社区支持和丰富的第三方库，可以满足各种需求和场景下的开发需求。

缺点：

学习曲线较陡：FastAPI 使用 Python 类型提示和异步编程模型，对于初学者来说可能有一定的学习曲线，需要一定的时间和经验来熟悉其使用方法和最佳实践。
文档不够完善：虽然 FastAPI 提供了自动生成 API 文档的功能，但有时文档的描述和示例不够详细和清晰，可能需要查阅更多的资料和文档来理解其使用方法。
生态系统相对较小：相比于其他成熟的 Web 框架，如 Django 和 Flask，FastAPI 的生态系统相对较小，可能缺乏一些常用的扩展库和工具，需要开发者自行选择和集成。

应用场景：

API 服务开发：FastAPI 提供了快速、高性能的 API 开发框架，适用于构建各种类型的 API 服务，包括 RESTful API、GraphQL API 等。
微服务架构：FastAPI 的异步请求处理和高性能特性使其非常适合于构建微服务架构中的服务端组件，可以轻松处理大量并发请求。
实时数据处理：对于需要实时数据处理和响应的应用场景，如实时监控系统、即时通讯应用等，FastAPI 的异步请求处理能力可以满足高并发和低延迟的要求。
机器学习模型部署：FastAPI 可以与机器学习框架（如 TensorFlow、PyTorch 等）结合使用，快速部署机器学习模型为 Web 服务，提供实时的预测和推理功能。
自动化测试：FastAPI 提供了自动生成 API 文档的功能，可以方便地进行接口测试和文档测试，是自动化测试框架的理想选择。
数据可视化：FastAPI 可以与前端框架（如 React、Vue.js 等）结合使用，快速构建数据可视化的 Web 应用程序，展示数据分析和图表展示等功能。
异步任务处理：FastAPI 支持异步任务处理，可以处理各种异步任务，如后台任务、定时任务等，提高了系统的响应速度和吞吐量。

二、路由定义

1、基本路由定义

使用@app.get()、@app.post()、@app.put()、@app.delete()等装饰器来定义路由，分别对应HTTP的GET、POST、PUT、DELETE方法。

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "Hello, World"}

上面的代码定义了一个根路径的GET请求路由，当用户通过浏览器访问根路径时，将返回一个JSON对象{"message": "Hello, World"}。

2、路径参数

使用{parameter}的形式在URL路径中定义路径参数，然后在路由处理函数中使用相同的参数名来接收传入的值。

@app.get("/items/{item_id}")
def read_item(item_id: int):
    return {"item_id": item_id}

上面的代码定义了一个带有路径参数的GET请求路由，当用户访问/items/123时，将返回一个JSON对象{"item_id": 123}。

3、查询参数

@app.get("/items/")
def read_items(q: str = None):
    return {"q": q}

上面的代码定义了一个带有查询参数的GET请求路由，当用户访问/items/?q=test时，将返回一个JSON对象{"q": "test"}。

4、请求体参数

使用Pydantic模型类作为参数注解来接收请求体参数（在第四部分会做详细的介绍）。

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.post("/items/")
def create_item(item: Item):
    return item

上面的代码定义了一个接收JSON请求体参数的POST请求路由，客户端发送的请求体参数需要符合Item模型的定义。

5、多个路由

一个处理函数可以同时处理多个路由。

@app.get("/items/{item_id}")
@app.get("/items/{item_id}/details")
def read_item(item_id: int):
    return {"item_id": item_id}

上面的代码定义了两个路由，它们都将调用同一个处理函数read_item()来处理请求。

三、请求处理

1、请求体的解析

FastAPI自动解析请求体中的数据，并将其转换为适当的Python类型。对于POST、PUT、DELETE等请求，可以通过函数参数注解来声明请求体参数，并指定参数的类型，FastAPI会自动解析请求体中的数据并校验其类型。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.post("/items/")
def create_item(item: Item):
    return item

上面的代码定义了一个接收JSON请求体参数的POST请求路由，参数item的类型为Item模型类，FastAPI会自动解析请求体中的JSON数据，并将其转换为Item对象。

2、响应体的构建

在处理请求时，可以返回任意类型的数据作为响应，FastAPI会自动将其序列化为JSON格式的响应体。同时，可以通过函数返回注解来指定响应的状态码。

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(item_id: int):
    return {"item_id": item_id, "name": "Item Name"}

上面的代码定义了一个GET请求路由，当客户端发送请求时，将返回一个JSON对象作为响应体，状态码为200。

3、文件上传和下载

FastAPI支持文件上传和下载的功能，可以通过File类型参数来接收上传的文件，或者返回文件对象作为响应。

from fastapi import FastAPI, File, UploadFile
from starlette.responses import FileResponse

app = FastAPI()

@app.post("/uploadfile/")
async def upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    with open(file.filename, "wb") as f:
        f.write(contents)
    return {"filename": file.filename}

@app.get("/downloadfile/")
def download_file():
    return FileResponse("example.txt", filename="example.txt")

上面的代码定义了一个文件上传的POST请求路由和一个文件下载的GET请求路由，分别用于接收上传的文件和提供下载文件的功能。

4、异步请求和响应

FastAPI支持异步请求和响应处理，可以使用async和await关键字来定义异步函数，以提高应用程序的并发性能。

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def async_read():
    return {"message": "Hello, World"}

@app.post("/")
async def async_create():
    await some_async_function()
    return {"message": "Created"}

上面的代码定义了两个异步函数来处理GET和POST请求，通过async关键字来声明异步函数，在函数中使用await关键字来调用异步函数或执行异步操作。

5、定义返回的网络状态

在 FastAPI 中，可以使用函数返回注解来定义接口返回的网络状态码。通过在处理函数中使用特定的返回注解，可以指定要返回的状态码和对应的响应内容。

下面是一个示例：

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(item_id: int):
    if item_id == 1:
        return {"item_id": item_id, "name": "Item Name"}
    else:
        # 使用 HTTPException 来指定异常情况下的状态码和错误信息
        raise HTTPException(status_code=404, detail="Item not found")

在上面的示例中，read_item() 函数定义了一个 GET 请求路由 /items/{item_id}，根据 item_id 的值返回相应的数据。如果 item_id 等于 1，则返回状态码为 200，表示请求成功；否则使用 HTTPException 抛出一个状态码为 404 的异常，表示资源未找到，并附带错误信息 "Item not found"。FastAPI 会自动将这个异常转换为一个带有指定状态码和错误信息的 HTTP 响应。

四、数据验证

在 FastAPI 中，可以使用 Pydantic 库来实现请求数据的自动验证和转换。Pydantic 是一个数据验证和序列化库，可以帮助您定义数据模型，并根据模型定义自动验证和转换请求数据，确保数据的类型安全和有效性。

1、定义数据模型

首先，需要定义一个 Pydantic 模型来描述请求数据的结构和类型。可以使用 Pydantic 提供的字段类型来定义模型的各个字段。

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

在上面的示例中，定义了一个名为 Item 的 Pydantic 模型，它包含了四个字段：name、description、price 和 tax。其中 name 和 price 是必填字段，而 description 和 tax 是可选字段，默认值为 None。

2、使用数据模型验证请求数据

接下来，在 FastAPI 的路由处理函数中，可以通过参数注解的方式使用定义好的 Pydantic 模型来验证请求数据。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.post("/items/")
def create_item(item: Item):
    return item

在上面的示例中，我们定义了一个 POST 请求路由 /items/，并将 Item 模型作为参数注解来接收请求数据。当客户端发送 POST 请求时，FastAPI 会自动将请求体中的数据解析并转换为 Item 对象，并根据模型定义进行类型验证和数据转换。如果请求数据的类型或格式不符合模型定义，FastAPI 将返回相应的错误信息。

五、异常处理

在 FastAPI 中，异常处理是保证 Web 应用程序稳定性和用户体验的重要部分。FastAPI 提供了一种简单而有效的方式来处理各种可能出现的异常情况，并返回友好的错误信息给客户端。

1、默认异常处理

FastAPI 默认提供了对常见异常情况的处理，例如请求参数验证失败、资源未找到等情况。当发生这些异常时，FastAPI 会自动返回相应的 HTTP 响应，包含标准的错误信息和状态码。

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(item_id: int):
    if item_id == 1:
        return {"item_id": item_id, "name": "Item Name"}
    else:
        # 抛出 HTTPException 来表示资源未找到的异常情况
        raise HTTPException(status_code=404, detail="Item not found")

在上面的示例中，如果客户端访问了不存在的资源，比如 /items/2，FastAPI 将抛出一个状态码为 404 的 HTTPException 异常，并返回错误信息 "Item not found" 给客户端。

2、自定义异常处理器

除了使用默认的异常处理机制外，还可以自定义异常处理器来统一处理异常并返回自定义的错误信息。可以使用 RequestValidationError 异常来捕获请求参数验证失败的异常，或者使用 HTTPException 异常来捕获其他类型的异常。

from fastapi import FastAPI, Request, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

app = FastAPI()

# 自定义请求参数验证失败的异常处理器
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={"detail": "Validation Error", "errors": exc.errors()}
    )

# 自定义其他类型异常的处理器
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail}
    )

@app.get("/items/{item_id}")
def read_item(item_id: int):
    if item_id == 1:
        return {"item_id": item_id, "name": "Item Name"}
    else:
        # 抛出 HTTPException 来表示资源未找到的异常情况
        raise HTTPException(status_code=404, detail="Item not found")

在上面的示例中，我们定义了两个自定义异常处理器：一个用于处理请求参数验证失败的异常，另一个用于处理其他类型的异常。当发生异常时，FastAPI 将根据异常的类型调用相应的异常处理器来处理异常，并返回自定义的错误信息给客户端。

六、安全性配置

在 FastAPI 中，可以通过配置来增强 Web 应用程序的安全性，包括身份验证、授权、HTTPS 支持、CORS 配置等。

1、身份验证与授权

FastAPI 提供了多种身份验证和授权机制，包括 OAuth2、JWT、HTTP Basic 认证等。您可以选择适合您应用程序需求的身份验证方案，并根据需要进行配置。

from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # 实现用户身份验证逻辑
    return {"access_token": "fake_access_token", "token_type": "bearer"}

@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
    # 实现用户授权逻辑
    return {"token": token}

在上面的示例中，我们使用 OAuth2 密码授权模式来实现身份验证和授权。客户端需要在登录时通过 /token 路由获取访问令牌，然后在访问受保护资源时将令牌作为参数传递给 /users/me 路由进行身份验证。

2、HTTPS 支持

FastAPI 支持通过配置来启用 HTTPS 安全连接，以确保客户端与服务器之间的通信是加密的，提高数据传输的安全性。

from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import HTTPBasic, HTTPBasicCredentials

app = FastAPI()

security = HTTPBasic()

@app.get("/items/")
async def read_items(credentials: HTTPBasicCredentials = Depends(security)):
    if credentials.username != "user" or credentials.password != "password":
        raise HTTPException(
            status_code=401,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Basic"},
        )
    return {"items": [{"item_id": "foo"}, {"item_id": "bar"}]}

在上面的示例中，我们通过配置 HTTPBasic 身份验证机制来启用 HTTP Basic 认证，并在 /items/ 路由中使用 Depends 来定义身份验证依赖。当客户端发送请求时，FastAPI 将根据配置的身份验证机制来验证用户身份，并返回相应的错误信息或数据。

3、CORS 配置

FastAPI 支持配置 CORS（跨域资源共享）来限制跨域请求的来源和访问权限，防止恶意攻击和数据泄露。

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost", "http://localhost:8080"],
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["Authorization", "Content-Type"],
)

在上面的示例中，我们通过配置 CORSMiddleware 中间件来启用 CORS 支持，并设置允许跨域请求的来源、允许的请求方法和请求头。

七、结合常用库

在 FastAPI 中，结合 Pydantic、Starlette 和其他常用的 Python 库可以帮助您构建复杂的 Web 应用程序，包括数据库访问、异步任务处理、缓存管理等功能的实现。

1、数据库访问

FastAPI 并不内置数据库访问功能，但可以结合第三方的异步数据库库来实现数据库访问功能。常用的异步数据库库包括 asyncpg、aiomysql、asyncio-sqlite 等，可以选择适合您项目需求的数据库库，并根据它们的文档进行配置和使用。

import asyncpg
from fastapi import FastAPI

app = FastAPI()

@app.get("/users/")
async def read_users():
    # 连接数据库
    conn = await asyncpg.connect(user="user", password="password", database="dbname", host="localhost")
    # 执行查询
    rows = await conn.fetch("SELECT * FROM users")
    # 关闭连接
    await conn.close()
    # 返回查询结果
    return rows

在上面的示例中，我们使用 asyncpg 库来连接 PostgreSQL 数据库，并执行查询操作，最后返回查询结果给客户端。

2、异步任务处理

FastAPI 支持异步请求处理和异步任务处理，可以使用 Python 的 asyncio 库来实现异步任务处理功能。常见的异步任务包括发送邮件、处理后台任务等。

from fastapi import FastAPI
import asyncio

app = FastAPI()

async def send_email(email: str, message: str):
    # 模拟发送邮件的异步操作
    await asyncio.sleep(5)
    print(f"Email sent to {email}: {message}")

@app.post("/send_email/")
async def send_email_endpoint(email: str, message: str):
    # 创建任务并等待任务完成
    asyncio.create_task(send_email(email, message))
    return {"message": "Email sent."}

在上面的示例中，我们定义了一个异步函数 send_email() 来模拟发送邮件的异步操作，并在路由处理函数中创建了一个异步任务来调用该函数。

3、缓存管理

FastAPI 并不提供缓存管理功能，但可以结合第三方的缓存库来实现缓存管理功能。常用的缓存库包括 aioredis、aiomcache 等，可以选择适合您项目需求的缓存库，并根据它们的文档进行配置和使用。

import aioredis
from fastapi import FastAPI

app = FastAPI()

@app.get("/cache/{key}")
async def read_cache(key: str):
    # 连接 Redis 缓存
    redis = await aioredis.create_redis_pool("redis://localhost")
    # 查询缓存
    value = await redis.get(key)
    # 关闭连接
    redis.close()
    await redis.wait_closed()
    # 返回查询结果
    return value

在上面的示例中，我们使用 aioredis 库来连接 Redis 缓存，并查询指定键的缓存值，然后返回给客户端。

八、完整示例

下面是一个完整的示例代码，尽可能的包含本文中介绍的所有内容。

from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from passlib.context import CryptContext
import jwt

# 定义常量
SECRET_KEY = "secret"
ALGORITHM = "HS256"
DATABASE_URL = "sqlite:///./test.db"

# 创建数据库连接
engine = create_engine(DATABASE_URL)
Base = declarative_base()
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 定义用户数据模型
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    password = Column(String)

# 创建数据库表
Base.metadata.create_all(bind=engine)

# 创建 FastAPI 实例
app = FastAPI()

# 定义请求数据模型
class UserIn(BaseModel):
    username: str
    password: str

# 定义响应数据模型
class UserOut(BaseModel):
    username: str

# 加密密码工具
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# 校验密码
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

# 密码哈希化
def get_password_hash(password):
    return pwd_context.hash(password)

# 获取用户
def get_user(db, username: str):
    return db.query(User).filter(User.username == username).first()

# 用户身份认证
def authenticate_user(db, username: str, password: str):
    user = get_user(db, username)
    if not user:
        return False
    if not verify_password(password, user.password):
        return False
    return user

# 生成访问令牌
def create_access_token(data: dict):
    to_encode = data.copy()
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

# 获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 登录获取访问令牌
@app.post("/token")
async def login_for_access_token(form_data: UserIn, db: Session = Depends(get_db)):
    user = authenticate_user(db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=401,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

# 创建用户
@app.post("/users/")
async def create_user(user: UserIn, db: Session = Depends(get_db)):
    db_user = User(username=user.username, password=get_password_hash(user.password))
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return {"username": db_user.username}

# 获取当前用户信息
@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
    return {"token": token}

# 后台任务
def background_task(name: str):
    print(f"Hello, {name}")

# 添加后台任务
@app.post("/background-task/")
async def background_task_endpoint(name: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(background_task, name)
    return {"message": "Background task added"}

# 运行 FastAPI 应用程序
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

这个示例实现了一个基本的用户身份验证和用户创建功能。用户的密码是通过 bcrypt 加密存储的，访问令牌是通过 JWT 进行签名生成的。同时，我们添加了一个异步后台任务的功能，用来模拟后台任务的处理。

寒秋丶