fastapi框架知识点总结

1. 异步编程：FastAPI 是一个基于 Python 异步编程的 web 框架，使用了 Python 3.4 引入的 asyncio 标准库来实现异步处理。因此，需要了解协程、异步函数和异步上下文等相关概念。
 1.1 在异步编程模型中，程序可以在等待 IO 操作时切换到其他任务，从而避免了 CPU 空闲等待 IO 操作的浪费：

from fastapi import FastAPI
import asyncio

app = FastAPI()

async def process_item(item):
    # 模拟 IO 操作
    await asyncio.sleep(1)
    return {"id": item["id"], "name": item["name"]}

@app.post("/items/")
async def create_item(item):
    result = await process_item(item)
    return result

 在这个示例中，我们定义了一个 process_item 函数，用于处理单个项目。该函数使用了 async/await 关键字，表示它是一个异步函数，当它遇到 IO 操作时，可以让控制权交给事件循环。
 在 create_item 路由函数中，我们使用了 await process_item(item)，这表示我们等待 process_item 函数执行完成，然后才能返回结果给客户端。由于 process_item 函数中包含了模拟 IO 操作的代码，所以在等待期间可以让事件循环去处理其他任务。
 对于业务场景，假设我们有一个 Web 服务，用户可以通过该服务上传大文件，该文件需要经过一系列处理步骤，然后才能保存到服务器上。在传统的同步编程模型中，处理大文件可能需要很长时间，而在异步编程模型中，我们可以利用 IO 等待的时间来执行其他任务，从而提高整个系统的并发能力。
 在这个场景下，我们可以使用 FastAPI 和异步编程模型来实现文件上传和处理功能，如下所示：

@app.post("/upload/")
async def upload_file(file: UploadFile):
    # 将上传的文件保存到服务器上
    with open("files/" + file.filename, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)

    # 异步处理文件
    asyncio.create_task(process_file(file.filename))

    return {"status": "ok"}

async def process_file(filename):
    # 模拟文件处理操作
    await asyncio.sleep(10)

    # 将处理后的文件保存到服务器上
    with open("processed_files/" + filename, "wb") as buffer:
        buffer.write(b"processed file data")

 在这个示例中，我们首先定义了一个上传文件的路由函数 upload_file，当用户上传文件时，我们首先将文件保存到服务器上，然后创建一个异步任务 process_file，用于处理上传的文件。在 process_file 函数中，我们模拟了一个 IO 操作，然后将处理后的文件保存到服务器上。
 由于 process_file 函数是一个异步函数，当它遇到 IO 操作时，可以让控制权交给事件循环，所以return会立即返回，客户端感受不到耗时

2. Pydantic 模型：FastAPI 使用 Pydantic 库来处理数据验证和解析。Pydantic 是一个用于数据验证和解析的 Python 库，可以将数据解析成 Python 类，并对数据进行验证和类型转换。
 2.1 Pydantic 提供了一个强大的数据模型定义方式，可以让我们定义一个数据模型类，并且利用这个类来验证和解析请求和响应的数据。这个类可以将 JSON、Form data 和 Query parameters 等多种数据格式解析成 Python 类型，并且可以对数据进行验证和类型转换，从而保证数据的正确性和一致性。
 以下是一个简单的示例，展示了如何在 FastAPI 中使用 Pydantic 数据模型来验证和解析请求数据：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    id: int
    name: str
    age: int

@app.post("/users/")
def create_user(user: User):
    return {"user": user}

 在上面的例子中，我们定义了一个名为 User 的 Pydantic 模型，它有三个属性：id、name 和 age。这些属性都有对应的类型，id 是整数类型，name 是字符串类型，age 是整数类型。
 接着，在我们的应用程序中定义了一个 /users/ 路由，它接受一个 User 对象作为参数。当客户端发送一个 HTTP POST 请求到 /users/ 时，FastAPI 会将请求中的 JSON 数据解析成一个 User 对象，并将其传递给 create_user 函数。如果解析失败或数据验证不通过，FastAPI 将返回一个 422 Unprocessable Entity 错误响应。
 以下是一个使用示例:

$ curl -X POST "http://localhost:8000/users/" \
  -H "Content-Type: application/json" \
  -d '{"id": 123, "name": "Alice", "age": 25}'

{"user": {"id": 123, "name": "Alice", "age": 25}}

3. 路由和依赖注入：FastAPI 提供了易于定义路由的方式，支持类似 Flask 的装饰器。此外，FastAPI 还支持依赖注入，可以方便地在路由处理函数中注入需要的依赖。
 3.1 我们可以使用类似 Flask 的装饰器语法来定义路由。以下是一个使用装饰器定义路由的例子：

from fastapi import FastAPI

app = FastAPI()

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

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

 在上面的例子中，我们定义了两个路由。第一个路由使用了 @app.get("/") 装饰器，表示这个路由对应 HTTP GET 请求，并处理应用程序的根路径。第二个路由使用了 @app.get("/items/{item_id}") 装饰器，表示这个路由对应 HTTP GET 请求，并处理 /items/ 路径下的子路径，其中 {item_id} 是一个参数，用于表示请求的具体资源。
 3.2 FastAPI 还支持依赖注入，可以方便地在路由处理函数中注入需要的依赖。以下是一个使用依赖注入的例子：

from fastapi import FastAPI, Depends

app = FastAPI()

def get_db():
    db = "fake_db"
    # 实际应用中会返回数据库连接等资源
    return db

@app.get("/")
async def root(db: str = Depends(get_db)):
    return {"db": db}

 在上面的例子中，我们定义了一个 get_db 函数，它返回一个虚假的数据库连接。我们在 @app.get("/")路由处理函数中使用了 Depends(get_db) 来声明一个依赖项，表示我们需要获取一个数据库连接。在实际处理请求之前，FastAPI 会调用 get_db 函数，并将其返回值传递给路由处理函数中的 db 参数。
 以上是一个简单的使用依赖注入的例子，实际应用中，我们可以使用依赖注入来注入数据库连接、配置信息、缓存等需要在多个路由中共享的资源。
以下是一个使用redis缓存的例子:

from fastapi import FastAPI, Depends
from pydantic import BaseSettings
from redis import Redis

app = FastAPI()

class Settings(BaseSettings):
    app_name: str = "My App"
    redis_host: str = "localhost"
    redis_port: int = 6379

settings = Settings()

@app.on_event("startup")
def startup_event():
    app.redis = Redis(host=settings.redis_host, port=settings.redis_port)

@app.get("/")
def read_root(redis: Redis = Depends(lambda: app.redis)):
    redis.incr("counter")
    return {"message": f"Hello {settings.app_name}!", "counter": redis.get("counter")}

 在上面的示例中，我们首先定义了一个 Settings 类，用来存储我们的配置信息。然后在 startup_event 中创建 Redis 实例，并将其赋值给 app.redis 属性。最后，在路由函数中使用 Depends 来注入 Redis 实例。这样，在每个请求中，我们就可以使用相同的 Redis 实例了。
 需要注意的是，这里的 Redis 实例是在应用程序启动时创建的，而不是在每个请求中创建的。这可以提高应用程序的性能，因为避免了在每个请求中创建 Redis 实例的开销。如果你需要在每个请求中动态创建 Redis 实例，可以将 Redis 的创建逻辑放到依赖项函数中，并在路由函数中使用 Depends 来注入依赖项函数。

4. 自动文档生成：FastAPI 提供了自动生成 API 文档的功能，可以根据代码注释和 Pydantic 模型自动生成文档，使得文档维护更加方便。
 4.1 FastAPI 使用 OpenAPI 和 JSON Schema 规范来定义和描述 API，它支持 Swagger UI 和 ReDoc 两种风格的文档生成器。通过使用 FastAPI，开发者可以通过代码注释和 Pydantic 模型来定义 API 的输入参数和输出结果，然后在运行应用程序时自动生成相应的文档。这使得文档的维护更加方便，避免了手动编写和更新文档的繁琐工作，同时也能够保证文档的准确性和一致性。
 以下是一个使用 FastAPI 自动生成 API 文档的示例：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    """
    商品模型，用于表示一个商品
    """
    name: str
    price: float

inventory = []

@app.post("/items/")
async def create_item(item: Item):
    """
    创建商品

    - **item**: `Item` 类型，表示要创建的商品

    Returns:
    - `Item` 类型，表示创建的商品
    """
    inventory.append(item)
    return item

@app.get("/items/")
async def read_items():
    """
    获取所有商品列表

    Returns:
    - `List[Item]` 类型，表示所有商品列表
    """
    return inventory

 在这个示例中，我们使用了 Pydantic 模型来定义商品模型，并在代码注释中添加了描述。在 create_item 和 read_items 方法中，我们也添加了代码注释来描述方法的输入、输出以及功能。FastAPI 会自动根据这些注释生成 API 文档，包括请求和响应的结构、字段类型和描述等信息。通过这种方式，我们可以方便地生成文档，同时也能提高代码的可读性和维护性。

效果

5. 性能优化：FastAPI 采用了一系列性能优化措施，如异步请求处理、高效的数据解析和验证、自动化的 API 文档生成等，可以显著提高 web 应用的性能。这里主要讲下使用缓存，FastAPI中可以使用缓存来提高Web应用程序的性能。以下是一个使用缓存的示例demo，其中使用了FastAPI自带的基于内存的缓存库cachetools。
 5.1 首先，需要安装cachetools库：pip install cachetools
 5.2 然后，创建一个FastAPI应用程序，并使用cachetools库来缓存响应。
 示例1:

from fastapi import FastAPI
from cachetools.func import ttl_cache

app = FastAPI()

@ttl_cache(maxsize=128, ttl=10)
def get_data():
    # 执行耗时较长的操作
    data = fetch_data_from_database()
    return data

@app.get("/")
async def root():
    # 获取数据，如果缓存中有数据则从缓存中获取，否则从数据库中获取
    data = get_data()
    return {"data": data}

 上面的示例演示了如何将函数的结果缓存为10秒，在这个示例中，我们使用ttl_cache装饰器来缓存get_data函数的结果。maxsize参数指定了缓存中最多可以存储的项目数，ttl参数指定了缓存数据的有效时间（以秒为单位）。
 在root函数中，我们通过调用get_data函数来获取数据。如果缓存中有数据，则从缓存中获取，否则从数据库中获取数据，并将结果存储在缓存中。这可以显著提高应用程序的性能，因为数据库查询等操作通常比从缓存中获取数据要慢得多。
 以下为fetch_data_from_database方法的大概样子:

import sqlite3

def fetch_data_from_database():
    # 连接数据库
    conn = sqlite3.connect('example.db')

    # 执行查询
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM my_table")
    data = cursor.fetchall()

    # 关闭数据库连接
    conn.close()

    return data

 示例2:

from fastapi import FastAPI
from cachetools import cached, TTLCache

app = FastAPI()

cache = TTLCache(maxsize=100, ttl=60)  # 定义一个最大缓存大小为100，缓存时间为60秒的缓存对象


@cached(cache)  # 使用@cached装饰器缓存函数的返回结果
def fibonacci(n: int) -> int:
    if n < 2:
        return n
    else:
        return fibonacci(n - 1) + fibonacci(n - 2)


@app.get("/fibonacci/{n}")  # 定义一个GET路由
async def get_fibonacci(n: int):
    result = fibonacci(n)  # 调用fibonacci函数
    return {"result": result}

 在上述示例中，我们定义了一个TTLCache对象作为缓存，它的最大缓存大小为100，缓存时间为60秒。然后，我们使用@cached装饰器缓存了一个名为fibonacci的函数，该函数用于计算斐波那契数列的第n项。最后，我们定义了一个GET路由，调用fibonacci函数并返回结果。
 当我们调用GET路由时，FastAPI会检查缓存对象是否已经缓存了函数的返回结果。如果已经缓存，则直接返回缓存的结果，否则调用函数计算结果，并将结果缓存起来

 需要注意的是，ttl_cache缓存的数据会在一定时间后失效。如果需要在缓存中保留数据，可以使用cache装饰器，它会将数据永久缓存。以下是一个使用cache装饰器的示例：

from cachetools.func import cache

@cache(maxsize=128)
def get_data():
    # 执行耗时较长的操作
    data = fetch_data_from_database()
    return data

 在这个示例中，cache装饰器将get_data函数的结果永久缓存。当应用程序启动时，它会执行一次查询，并将结果存储在缓存中，之后每次调用get_data函数时都会从缓存中获取数据。
6. WebSocket 支持：FastAPI 支持 WebSocket，可以轻松地创建实时 web 应用。
 示例:

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

app = FastAPI()

# 存储所有连接到 WebSocket 的客户端
client_websockets = []

# WebSocket 路由处理函数，当客户端连接时调用
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    # 添加客户端到存储列表中
    client_websockets.append(websocket)

    # 发送欢迎信息给客户端
    await websocket.send_text("Welcome to the WebSocket server!")

    # 循环读取客户端发送的消息，并广播给所有客户端
    while True:
        data = await websocket.receive_text()
        for client in client_websockets:
            await client.send_text(f"User {id(websocket)} says: {data}")

# 静态页面，用于测试 WebSocket
html = """
<!DOCTYPE html>
<html>
    <head>
        <title>WebSocket Test</title>
        <script>
            var ws = new WebSocket("ws://" + window.location
        ws.onopen = function(event) {
            console.log("WebSocket opened.");
        };

        ws.onmessage = function(event) {
            console.log(event.data);
        };

        function sendMessage() {
            var input = document.getElementById("message");
            var message = input.value;
            ws.send(message);
            input.value = "";
        }
    </script>
</head>
<body>
    <h1>WebSocket Test</h1>
    <div>
        <input type="text" id="message">
        <button onclick="sendMessage()">Send</button>
    </div>
  </body>
</html>
"""

# 返回静态页面
@app.get("/")
async def get():
    return HTMLResponse(html)

7. 安全性：FastAPI 通过采用安全的默认设置和内置的安全性措施来保护 web 应用程序。例如，FastAPI 在默认情况下启用了 CSRF 保护和 CORS。
 7.1 启用CORS:

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

app = FastAPI()

# 配置跨域资源共享
origins = [
    "http://localhost",
    "http://localhost:8080",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

 在这个示例中，我们允许来自http://localhost和http://localhost:8080的请求，设置allow_credentials为True，并允许任何方法和任何标头。
 7.1 启用CSRF:

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates

app = FastAPI()

# 启用 CSRF 保护
app.add_middleware(CSRFMiddleware, secret_key="secret")

# 定义模板
templates = Jinja2Templates(directory="templates")

# 路由
@app.get("/", response_class=HTMLResponse)
async def read_root(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

 在上面的示例代码中，我们首先导入了 FastAPI 应用程序所需的依赖项和模块。然后，我们使用 app.add_middleware 方法启用 CSRF 保护，它需要一个秘密密钥。在此示例中，我们使用字符串 "secret" 作为密钥。
 然后，我们定义了一个名为 templates 的模板对象，并指定了模板文件所在的目录。
 最后，我们定义了一个名为 read_root 的路由函数来处理根路由 ("/")。这个函数使用 Jinja2 模板引擎渲染 HTML 模板，并将请求对象传递给模板。
这个示例 Web 应用程序演示了如何使用 FastAPI 中的默认安全设置和措施来保护应用程序。

8. 扩展性：FastAPI 可以通过中间件和插件进行扩展，支持集成其他 Python 库和第三方服务。中间件是FastAPI应用程序中的可重用代码块，可以在请求和响应处理管道中对请求和响应进行修改和处理。通过使用中间件，可以添加一些全局性的行为，如请求认证、日志记录、错误处理等。
例如，下面的示例代码展示了如何在FastAPI应用程序中使用JWT认证中间件：

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from jwt import decode, InvalidTokenError

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")

def verify_token(token: str):
    try:
        payload = decode(token, "secret_key", algorithms=["HS256"])
        username = payload.get("sub")
        if username is None:
            raise InvalidTokenError("Invalid token")
        return username
    except InvalidTokenError as e:
        raise HTTPException(status_code=401, detail="Invalid token")

@app.get("/")
async def read_root(token: str = Depends(oauth2_scheme)):
    username = verify_token(token)
    return {"Hello": username}

 使用中间件记录日志信息：

from fastapi import FastAPI, Request
import logging

app = FastAPI()

# 创建一个logger对象，并设置日志级别为INFO
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# 创建一个中间件，用于记录请求和响应的信息
@app.middleware("http")
async def log_request(request: Request, call_next):
    # 获取请求路径、请求方法和请求参数
    path = request.url.path
    method = request.method
    params = await request.json()

    # 调用下一个中间件或路由处理函数，并获取响应结果
    response = await call_next(request)

    # 记录请求和响应的信息
    logger.info(f"{method} {path} {params} {response.json()}")

    return response

# 定义一个路由处理函数
@app.get("/")
async def hello_world():
    return {"message": "Hello World"}

 当我们访问 / 路径时，请求和响应的信息将会被记录到日志中，例如：

INFO:__main__:GET / {"message": "Hello World"} 200

 使用中间件进行错误处理:

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import logging

app = FastAPI()

# 创建一个logger对象，并设置日志级别为INFO
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# 创建一个中间件，用于处理应用程序中的异常
@app.exception_handler(Exception)
async def error_handler(request: Request, exc: Exception):
    # 记录异常的信息
    logger.error(f"{type(exc).__name__}: {exc}")

    # 返回适当的错误响应
    return JSONResponse(
        status_code=500,
        content={"message": "Internal Server Error"}
    )

# 定义一个路由处理函数，用于抛出异常
@app.get("/")
async def divide_by_zero():
    1/0

 当我们访问 / 路径时，抛出ZeroDivisionError异常并被中间件函数处理。