如何使用FastAPI [详细Python指南]

FastAPI Python结合了现代Python特性与高性能Web开发能力。这个框架因其速度、易用性和内置的异步编程支持而脱颖而出。无论您是构建API、微服务还是全栈应用程序,FastAPI都提供了简化开发流程的工具。本指南将引导您使用FastAPI创建高效、可扩展且文档完善的Web应用程序,突出其与其他Python Web框架的区别。

什么是FastAPI?

FastAPI Python利用标准类型提示来创建高性能API。这个为Python 3.6及以上版本构建的框架通过其直观的设计和强大的功能加速开发。FastAPI在生产环境中表现卓越,能够有效处理高负载,同时保持代码可读性并减少错误。其自动API文档和验证功能使其成为既注重速度又注重质量的开发者的理想选择。

为什么选择FastAPI?

1. 速度快:FastAPI是最快的Python框架之一,可与NodeJS和Go媲美
2. 易于学习:如果您懂Python,您已经成功了一半
3. 开发快速:将功能开发速度提高约200%到300%
4. 减少错误:减少约40%的人为(开发者)引起的错误
5. 直观:出色的编辑器支持,处处自动完成,减少调试时间
6. 简单:设计易于使用和学习,减少阅读文档的时间
7. 代码简洁:最小化代码重复,每个参数声明支持多种功能
8. 稳健:获得生产就绪的代码,带有自动交互式文档
9. 基于标准:基于(并完全兼容)API的开放标准:OpenAPI和JSON Schema

FastAPI入门

安装FastAPI及其依赖项

要开始使用FastAPI,您首先需要安装它及其依赖项。以下是使用pip安装的方法:

pip install fastapi
pip install "uvicorn[standard]"

这将安装FastAPI以及Uvicorn,一个我们将用于运行FastAPI应用程序的ASGI服务器。

设置开发环境

为了获得最佳开发体验,建议使用虚拟环境。以下是设置方法:

python -m venv fastapi-env
source fastapi-env/bin/activate  # Windows上使用 `fastapi-env\Scripts\activate`

创建您的第一个FastAPI应用程序

让我们创建一个简单的"Hello World"示例来开始使用FastAPI:

1. 创建名为main.py的新文件:

from fastapi import FastAPI

app = FastAPI()

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

2. 运行服务器:

uvicorn main:app --reload

3. 打开浏览器并导航到http://127.0.0.1:8000。您应该看到JSON响应:{"message": "Hello World"}。

FastAPI基础

理解Python类型提示及FastAPI如何使用它们

FastAPI利用Python的类型提示提供自动数据验证、序列化和文档。以下是示例:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = None

@app.post("/items/")
async def create_item(item: Item):
    return {"item_name": item.name, "item_price": item.price}

FastAPI中的请求处理和路由

FastAPI使定义路由和处理不同HTTP方法变得简单:

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

这个路由处理对/items/{item_id}的GET请求,其中item_id是路径参数,q是可选的查询参数。

路径参数和查询参数

FastAPI允许您轻松使用路径和查询参数:

• 路径参数:@app.get("/users/{user_id}")
• 查询参数:@app.get("/items/")

示例:

@app.get("/users/{user_id}/items/")
async def read_user_item(
    user_id: int, item_id: str, q: str = None, short: bool = False
):
    item = {"item_id": item_id, "owner_id": user_id}
    if q:
        item.update({"q": q})
    if not short:
        item.update(
            {"description": "这是一个有长描述的超赞物品"}
        )
    return item

请求体和数据验证

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/")
async def create_item(item: Item):
    return item

响应模型和状态码

您可以在FastAPI中指定响应模型和状态码:

from fastapi import FastAPI, status
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.post("/items/", response_model=Item, status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    return item

FastAPI高级功能

FastAPI中的依赖注入

FastAPI的依赖注入系统允许您轻松管理多个路由处理程序之间的共享逻辑:

from fastapi import Depends, FastAPI

app = FastAPI()

async def common_parameters(q: str = 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 {"message": "读取物品", "params": commons}

@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return {"message": "读取用户", "params": commons}

安全性和身份验证

FastAPI为各种安全和身份验证方法提供内置支持。以下是使用OAuth2和JWT令牌的示例:

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="用户名或密码错误")
    access_token = create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

后台任务和异步操作

FastAPI开箱即用地支持后台任务和异步操作:

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"发送给 {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, message="某些通知")
    return {"message": "通知在后台发送"}

FastAPI中的WebSocket支持

FastAPI还提供WebSocket支持:

from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse

app = FastAPI()

html = """
<!DOCTYPE html>
<html>
    <head>
        <title>聊天</title>
    </head>
    <body>
        <h1>WebSocket聊天</h1>
        <form action="" onsubmit="sendMessage(event)">
            <input type="text" id="messageText" autocomplete="off"/>
            <button>发送</button>
        </form>
        <ul id='messages'>
        </ul>
        <script>
            var ws = new WebSocket("ws://localhost:8000/ws");
            ws.onmessage = function(event) {
                var messages = document.getElementById('messages')
                var message = document.createElement('li')
                var content = document.createTextNode(event.data)
                message.appendChild(content)
                messages.appendChild(message)
            };
            function sendMessage(event) {
                var input = document.getElementById("messageText")
                ws.send(input.value)
                input.value = ''
                event.preventDefault()
            }
        </script>
    </body>
</html>
"""

@app.get("/")
async def get():
    return HTMLResponse(html)

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"消息文本是: {data}")

使用OpenTelemetry进行监控和跟踪

FastAPI可以轻松与OpenTelemetry集成以进行全面的监控和跟踪。这可以显著增强您理解和优化应用程序性能的能力。

要在FastAPI应用程序中开始使用OpenTelemetry:

1. 安装必要的包:

pip install opentelemetry-api opentelemetry-sdk opentelemetry-instrumentation-fastapi

2. 插桩您的FastAPI应用程序:

from fastapi import FastAPI
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

app = FastAPI()

FastAPIInstrumentor.instrument_app(app)

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

FastAPI与数据库集成

在FastAPI中使用SQLAlchemy

FastAPI与SQLAlchemy配合良好进行数据库操作。以下是基本示例:

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

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)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

Base.metadata.create_all(bind=engine)

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="邮箱已注册")
    return crud.create_user(db=db, user=user)

测试FastAPI应用程序

使用pytest进行单元测试

FastAPI与pytest无缝配合进行测试。以下是测试用例示例:

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

def test_create_item():
    response = client.post(
        "/items/",
        json={"name": "Foo", "price": 42.0},
    )
    assert response.status_code == 200
    assert response.json() == {
        "name": "Foo",
        "price": 42.0,
        "is_offer": None,
    }

要运行这些测试,只需在终端中使用pytest命令。

FastAPI性能优化

使用async和await提高并发性

FastAPI构建在Starlette之上,允许异步请求处理:

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    result = await some_async_operation(item_id)
    return result

这使您的应用程序能够有效处理许多并发请求。

记录FastAPI应用程序

使用Swagger UI自动生成API文档

FastAPI Python通过其内置的Swagger UI革新了API文档。可在'/docs'端点访问,这个交互式界面自动为您的API生成全面的文档。开发人员可以探索端点、使用不同参数测试请求,并实时查看响应模型。此功能不仅加速开发和测试,而且作为活文档,始终与您的代码同步。FastAPI中的Swagger UI简化了内部团队和外部开发人员的API使用,增强了协作并减少了集成时间。

部署FastAPI应用程序

使用Docker容器化FastAPI应用

以下是FastAPI应用程序的简单Dockerfile:

FROM python:3.9
WORKDIR /code

COPY ./requirements.txt /code/requirements.txt

RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt

COPY ./app /code/app

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "80"]

构建和运行Docker容器:

docker build -t myapp .
docker run -d --name mycontainer -p 80:80 myapp

FastAPI最佳实践和设计模式

可扩展FastAPI应用程序的项目结构

FastAPI应用程序的推荐项目结构:

├── app
│   ├── __init__.py
│   ├── main.py
│   ├── dependencies.py
│   ├── routers
│   │   ├── __init__.py
│   │   ├── items.py
│   │   └── users.py
│   ├── internal
│   │   ├── __init__.py
│   │   └── admin.py
│   ├── schemas
│   │   ├── __init__.py
│   │   ├── item.py
│   │   └── user.py
│   └── crud
│       ├── __init__.py
│       ├── base.py
│       ├── item.py
│       └── user.py
├── tests
│   ├── __init__.py
│   ├── test_main.py
│   ├── test_items.py
│   └── test_users.py

错误处理和自定义异常类

FastAPI允许您创建自定义异常处理程序,以更精细地控制错误响应:

from fastapi import FastAPI, Request, status
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_code=418,
        content={"message": f"哎呀! {exc.name} 做了些什么。彩虹来了..."},
    )

@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

FastAPI中的中间件实现

中间件允许您向FastAPI应用程序添加自定义功能。以下是为每个响应添加自定义标头的简单中间件示例:

from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response

app = FastAPI()

class AddCustomHeaderMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        response.headers["X-Custom-Header"] = "某个值"
        return response

app.add_middleware(AddCustomHeaderMiddleware)

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

FastAPI生态系统和扩展

流行的FastAPI扩展和库

FastAPI拥有不断增长的扩展和库生态系统。以下是一些流行的:

1. FastAPI Users:为FastAPI提供即用且可自定义的用户管理
2. FastAPI SQL Database:使用SQLAlchemy简化数据库操作
3. FastAPI Cache:为FastAPI端点添加缓存支持
4. FastAPI CORS:处理FastAPI应用程序的跨域资源共享(CORS)
5. FastAPI Limiter:为FastAPI路由添加速率限制

将FastAPI与其他Python框架集成

FastAPI可以与其他Python框架和库集成。例如,您可以将FastAPI与Celery用于任务队列:

from fastapi import FastAPI
from celery import Celery

app = FastAPI()
celery = Celery("tasks", broker="redis://localhost:6379")

@celery.task
def add_numbers(x, y):
    return x + y

@app.get("/add/{x}/{y}")
async def add(x: int, y: int):
    result = add_numbers.delay(x, y)
    return {"task_id": result.id}

真实世界的FastAPI用例

使用FastAPI构建RESTful API

FastAPI非常适合构建RESTful API。以下是待办事项列表的简单CRUD(创建、读取、更新、删除)API示例:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI()

class Todo(BaseModel):
    id: Optional[int] = None
    item: str
    completed: bool = False

todos = []

@app.post("/todos/", response_model=Todo)
async def create_todo(todo: Todo):
    todo.id = len(todos) + 1
    todos.append(todo)
    return todo

@app.get("/todos/", response_model=List[Todo])
async def read_todos():
    return todos

@app.get("/todos/{todo_id}", response_model=Todo)
async def read_todo(todo_id: int):
    try:
        return next(todo for todo in todos if todo.id == todo_id)
    except StopIteration:
        raise HTTPException(status_code=404, detail="未找到待办事项")

@app.put("/todos/{todo_id}", response_model=Todo)
async def update_todo(todo_id: int, todo: Todo):
    for i, t in enumerate(todos):
        if t.id == todo_id:
            todo.id = todo_id
            todos[i] = todo
            return todo
    raise HTTPException(status_code=404, detail="未找到待办事项")

@app.delete("/todos/{todo_id}")
async def delete_todo(todo_id: int):
    for i, todo in enumerate(todos):
        if todo.id == todo_id:
            del todos[i]
            return {"message": "待办事项删除成功"}
    raise HTTPException(status_code=404, detail="未找到待办事项")

比较FastAPI与其他Web框架

FastAPI vs Flask

FastAPI vs Django

何时为项目选择FastAPI

在以下情况选择FastAPI:

1. 您正在构建高性能API
2. 您需要自动API文档
3. 您需要强类型检查
4. 您正在开发微服务架构
5. 您想在Python中利用异步编程
6. 您需要WebSocket支持

结论

关于如何使用FastAPI的要点回顾

1. FastAPI为构建API提供了现代、快速且易于使用的框架
2. 它利用Python类型提示进行自动验证和文档生成
3. FastAPI同时支持同步和异步编程
4. 它为OAuth2、JWT令牌和其他安全功能提供内置支持
5. FastAPI与各种数据库和ORM良好集成
6. 它使用Swagger UI提供自动交互式API文档
7. FastAPI高度可扩展,拥有不断增长的库和工具生态系统

FastAPI和Web开发的未来

FastAPI在Web开发的未来中占据重要地位,特别是在高性能、类型安全的API开发领域。随着Python继续发展,更好地支持异步编程和类型提示,FastAPI将充分利用这些进步。

掌握FastAPI的后续步骤

1. 深入研究FastAPI的高级功能,如依赖注入和后台任务
2. 探索FastAPI的生态系统并尝试流行的扩展
3. 练习构建真实世界的应用程序,从简单的CRUD API到复杂的微服务
4. 通过创建扩展或向主项目提交拉取请求来为FastAPI社区做出贡献
5. 关注最新的FastAPI版本和最佳实践

常见问题

1. FastAPI适合初学者吗? 虽然与某些框架相比,FastAPI的学习曲线较陡,但对于掌握Python的初学者来说仍然易于上手。广泛的文档和直观的设计使其成为希望构建现代API的人的绝佳选择。

2. FastAPI如何处理数据库操作? FastAPI不包含内置ORM,但它与SQLAlchemy和其他数据库库配合良好。您可以轻松集成这些工具来处理FastAPI应用程序中的数据库操作。

3. FastAPI可以用于全栈Web开发吗? 虽然FastAPI擅长构建API,但与React或Vue.js等前端框架结合使用时,它可以用于全栈开发。但是,对于复杂的全栈应用程序,您可能想考虑像Django这样的框架。

4. FastAPI在生产环境中表现如何? FastAPI设计为开箱即用的生产就绪框架。其高性能,结合自动API文档和强大的错误处理等功能,使其成为生产环境的绝佳选择。

5. 部署FastAPI应用程序容易吗? 是的,FastAPI应用程序相对容易部署。它们可以使用Docker容器化并部署到各种云平台或本地服务器。该框架与Uvicorn等ASGI服务器的兼容性使部署变得简单。

参考资料

• FastAPI Documentation
• FastAPI GitHub
• FastAPI Twitter
• FastAPI Reddit
• FastAPI Stack Overflow
• FastAPI GitHub Discussions
• FastAPI GitHub Issues
• FastAPI GitHub Pull Requests