导航菜单

  • 1.vector
  • 2.milvus
  • 3.pymilvus
  • 4.rag
  • 5.rag_measure
  • ragflow
  • heapq
  • HNSW
  • cosine_similarity
  • math
  • typing
  • etcd
  • minio
  • collections
  • jieba
  • random
  • beautifulsoup4
  • chromadb
  • sentence_transformers
  • numpy
  • lxml
  • openpyxl
  • PyMuPDF
  • python-docx
  • requests
  • python-pptx
  • text_splitter
  • all-MiniLM-L6-v2
  • openai
  • llm
  • BPETokenizer
  • Flask
  • RAGAS
  • BagofWords
  • langchain
  • Pydantic
  • abc
  • faiss
  • MMR
  • scikit-learn
  • Runnable
  • PromptEngineering
  • dataclasses
  • LaTeX
  • rank_bm25
  • TF-IDF
  • asyncio
  • sqlalchemy
  • fastapi
  • Starlette
  • uvicorn
  • argparse
  • Generic
  • ssl
  • urllib
  • python-dotenv
  • RRF
  • CrossEncoder
  • Lost-in-the-middle
  • Jinja2
  • logger
  • io
  • venv
  • concurrent
  • parameter
  • SSE
  • 1. Starlette 是什么?
    • 1.1 为什么需要 Starlette?
    • 1.2 Starlette 的核心特点
  • 2. 前置知识
    • 2.1 什么是 ASGI?
    • 2.2 异步编程基础
    • 2.3 HTTP 基础知识
    • 2.4 Python 基础知识
  • 3. 安装与环境准备
    • 3.1 安装 Starlette
    • 3.2 验证安装
  • 4. 第一个完整示例:Hello World
  • 5. 路径参数:从 URL 中获取数据
  • 6. 查询参数:从 URL 问号后获取数据
  • 7. 请求体:接收 POST/PUT 请求的数据
  • 8. 响应类型:不同类型的 HTTP 响应
  • 9. 完整的 CRUD 示例:增删改查
  • 10. 错误处理:HTTPException
  • 11. 请求对象:获取请求的详细信息
  • 12. 应用生命周期:启动和关闭事件
  • 13. 常见错误与最佳实践
    • 13.1 常见错误
      • 错误 1:忘记使用 await
      • 错误 2:路径参数类型错误
      • 错误 3:未处理不存在的资源
    • 13.2 最佳实践
  • 14. Starlette 与 FastAPI 的关系
  • 15. 总结
    • 15.1 核心概念回顾
    • 15.2 基本使用流程

1. Starlette 是什么? #

Starlette 是一个轻量级的 ASGI(异步服务器网关接口)框架,它是构建高性能异步 Web 应用的基础。FastAPI 就是基于 Starlette 构建的。

1.1 为什么需要 Starlette? #

如果你已经学过 FastAPI,可能会问:为什么还要学 Starlette?

简单理解:

  • FastAPI:高级框架,提供了很多便利功能(自动文档、数据验证等)
  • Starlette:底层框架,提供了更基础、更灵活的控制

使用 Starlette 的场景:

  1. 需要更精细的控制(不需要 FastAPI 的高级特性)
  2. 学习 ASGI 的工作原理
  3. 构建轻量级的异步服务
  4. 开发自定义框架或中间件

1.2 Starlette 的核心特点 #

  1. 极简设计:只提供核心功能,不包含太多高级抽象
  2. 高性能:专为异步编程设计,性能优秀
  3. ASGI 原生:完全兼容 ASGI 标准
  4. 模块化:可以独立使用各个组件

2. 前置知识 #

在学习 Starlette 之前,你需要了解以下基础知识。

2.1 什么是 ASGI? #

ASGI(Asynchronous Server Gateway Interface,异步服务器网关接口) 是 Python Web 应用和服务器之间的标准接口。

简单理解:

  • WSGI(旧标准):同步接口,一次只能处理一个请求
  • ASGI(新标准):异步接口,可以同时处理多个请求

ASGI 的优势:

  • 支持异步编程(async/await)
  • 支持 WebSocket
  • 支持 HTTP/2
  • 性能更好

2.2 异步编程基础 #

Starlette 大量使用异步编程,你需要了解:

  • async def:定义异步函数
  • await:等待异步操作完成
  • asyncio:Python 的异步编程库

如果你不熟悉异步编程,建议先学习 Python 的 asyncio 模块。

2.3 HTTP 基础知识 #

  • 请求方法:GET、POST、PUT、DELETE 等
  • URL 路径:如 /users/123
  • 查询参数:如 ?skip=0&limit=10
  • 请求头:如 Content-Type: application/json
  • 请求体:POST/PUT 请求中发送的数据

2.4 Python 基础知识 #

你需要熟悉:

  • 函数定义:def 和 async def
  • 字典操作:访问字典的键值
  • 列表操作:列表的基本操作
  • 类型提示:基本的类型注解(可选,但推荐)

3. 安装与环境准备 #

3.1 安装 Starlette #

使用 pip 安装:

pip install starlette
pip install uvicorn[standard]

说明:

  • starlette:Starlette 框架本身
  • uvicorn:ASGI 服务器,用于运行 Starlette 应用

3.2 验证安装 #

创建一个简单的测试文件 test_install.py:

# 导入 Starlette 和 JSONResponse
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

# 定义处理函数:处理根路径的请求
async def homepage(request):
    # 返回 JSON 响应
    return JSONResponse({"message": "Starlette 安装成功!"})

# 创建 Starlette 应用
# routes 参数指定路由列表
app = Starlette(
    debug=True,  # 开启调试模式
    routes=[
        Route("/", homepage),  # 将根路径 "/" 映射到 homepage 函数
    ]
)

# 运行应用(使用 uvicorn)
if __name__ == "__main__":
    import uvicorn
    # 运行服务器,监听 8000 端口
    uvicorn.run(app, host="0.0.0.0", port=8000)

运行测试:

python test_install.py

然后在浏览器访问 http://localhost:8000,应该能看到 {"message": "Starlette 安装成功!"}。

4. 第一个完整示例:Hello World #

让我们创建一个最简单的 Starlette 应用,理解基本结构。

# 导入必要的模块
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

# 定义处理函数:处理 GET 请求到根路径 "/"
# request 参数包含请求的所有信息
async def read_root(request):
    # 返回 JSON 响应
    return JSONResponse({"message": "Hello World"})

# 定义另一个处理函数:处理 GET 请求到 "/hello"
async def say_hello(request):
    # 返回问候语
    return JSONResponse({"message": "你好,Starlette!"})

# 创建 Starlette 应用实例
# routes 参数是一个路由列表,每个路由将 URL 路径映射到处理函数
app = Starlette(
    debug=True,  # 开启调试模式(开发时使用)
    routes=[
        Route("/", read_root),  # 将 "/" 映射到 read_root 函数
        Route("/hello", say_hello),  # 将 "/hello" 映射到 say_hello 函数
    ]
)

# 运行应用
if __name__ == "__main__":
    import uvicorn
    # 启动服务器
    uvicorn.run(app, host="0.0.0.0", port=8000)

运行方法:

  1. 将代码保存为 hello.py
  2. 运行 python hello.py
  3. 在浏览器访问 http://localhost:8000 或 http://localhost:8000/hello

或者使用命令行运行:

uvicorn hello:app --reload

其中 hello 是文件名(不含 .py),app 是 Starlette 应用实例的变量名。

5. 路径参数:从 URL 中获取数据 #

路径参数是 URL 路径中的一部分,用于传递数据。例如,/users/123 中的 123 就是路径参数。

# 导入必要的模块
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

# 定义处理函数:获取用户信息
# {user_id} 是路径参数,Starlette 会自动提取并传递给函数
async def get_user(request):
    # 从 request.path_params 字典中获取路径参数
    # path_params 是一个字典,包含所有路径参数
    user_id = request.path_params["user_id"]
    # 返回用户信息
    return JSONResponse({
        "user_id": user_id,
        "message": f"获取用户 {user_id} 的信息"
    })

# 定义处理函数:获取商品信息
# {item_name} 是路径参数,类型为字符串
async def get_item(request):
    # 获取路径参数
    item_name = request.path_params["item_name"]
    # 返回商品信息
    return JSONResponse({
        "item_name": item_name,
        "message": f"获取商品 {item_name}"
    })

# 创建应用
app = Starlette(
    debug=True,
    routes=[
        # 定义路由:{user_id} 是路径参数
        Route("/users/{user_id}", get_user),
        # 定义路由:{item_name} 是路径参数
        Route("/items/{item_name}", get_item),
    ]
)

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

测试方法:

  • 访问 http://localhost:8000/users/123,会返回 {"user_id": "123", "message": "获取用户 123 的信息"}
  • 访问 http://localhost:8000/items/apple,会返回 {"item_name": "apple", "message": "获取商品 apple"}

注意:路径参数默认是字符串类型。如果需要类型转换,需要在处理函数中手动转换。

6. 查询参数:从 URL 问号后获取数据 #

查询参数是 URL 中 ? 后面的参数,如 /items?skip=0&limit=10 中的 skip=0 和 limit=10。

# 导入必要的模块
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

# 定义处理函数:获取商品列表
async def read_items(request):
    # 从 request.query_params 中获取查询参数
    # query_params 是一个类似字典的对象,包含所有查询参数
    # 使用 get() 方法获取参数,如果不存在则使用默认值
    skip = int(request.query_params.get("skip", "0"))  # 默认值为 "0",转换为整数
    limit = int(request.query_params.get("limit", "10"))  # 默认值为 "10",转换为整数
    # 返回查询参数
    return JSONResponse({
        "skip": skip,
        "limit": limit,
        "message": f"跳过 {skip} 条,返回 {limit} 条"
    })

# 定义处理函数:搜索商品
async def search_items(request):
    # 获取查询参数 q(可选)
    q = request.query_params.get("q")
    # 如果提供了搜索关键词
    if q:
        return JSONResponse({
            "query": q,
            "message": f"搜索关键词:{q}"
        })
    else:
        # 如果没有提供搜索关键词
        return JSONResponse({
            "message": "未提供搜索关键词"
        })

# 创建应用
app = Starlette(
    debug=True,
    routes=[
        Route("/items/", read_items),
        Route("/search/", search_items),
    ]
)

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

测试方法:

  • 访问 http://localhost:8000/items/,会使用默认值 skip=0, limit=10
  • 访问 http://localhost:8000/items/?skip=20&limit=5,会使用提供的值
  • 访问 http://localhost:8000/search/?q=python,会返回搜索结果

7. 请求体:接收 POST/PUT 请求的数据 #

请求体是 POST/PUT 请求中发送的数据(通常是 JSON 格式)。Starlette 提供了简单的方法来读取请求体。

# 导入必要的模块
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

# 定义处理函数:创建商品
async def create_item(request):
    # 检查请求方法
    if request.method == "POST":
        # 读取 JSON 格式的请求体
        # await request.json() 返回解析后的字典
        data = await request.json()
        # 返回创建的商品信息
        return JSONResponse({
            "message": "商品创建成功",
            "item": data
        })
    else:
        # 如果不是 POST 请求,返回提示信息
        return JSONResponse({
            "message": "请使用 POST 方法发送请求"
        })

# 定义处理函数:更新商品
async def update_item(request):
    # 检查请求方法
    if request.method == "PUT":
        # 获取路径参数
        item_id = request.path_params["item_id"]
        # 读取 JSON 格式的请求体
        data = await request.json()
        # 返回更新后的商品信息
        return JSONResponse({
            "message": f"商品 {item_id} 更新成功",
            "item_id": item_id,
            "item": data
        })
    else:
        # 如果不是 PUT 请求,返回提示信息
        return JSONResponse({
            "message": "请使用 PUT 方法发送请求"
        })

# 创建应用
app = Starlette(
    debug=True,
    routes=[
        # 定义路由,methods 参数指定允许的 HTTP 方法
        Route("/items/", create_item, methods=["GET", "POST"]),
        Route("/items/{item_id}", update_item, methods=["GET", "PUT"]),
    ]
)

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

测试方法(使用 curl 或 Postman):

# 创建商品
curl -X POST "http://localhost:8000/items/" \
  -H "Content-Type: application/json" \
  -d '{"name": "苹果", "price": 5.5}'

# 更新商品
curl -X PUT "http://localhost:8000/items/123" \
  -H "Content-Type: application/json" \
  -d '{"name": "红苹果", "price": 6.0}'

8. 响应类型:不同类型的 HTTP 响应 #

Starlette 提供了多种响应类型,用于返回不同格式的数据。

# 导入必要的模块
from starlette.applications import Starlette
from starlette.responses import (
    JSONResponse,      # JSON 响应
    HTMLResponse,      # HTML 响应
    PlainTextResponse, # 纯文本响应
    RedirectResponse,  # 重定向响应
    Response           # 通用响应
)
from starlette.routing import Route

# 定义处理函数:返回 JSON 响应
async def json_response(request):
    # JSONResponse 用于返回 JSON 格式的数据
    return JSONResponse({"status": "ok", "message": "这是 JSON 响应"})

# 定义处理函数:返回 HTML 响应
async def html_response(request):
    # HTMLResponse 用于返回 HTML 内容
    content = "<h1>Hello World</h1><p>这是 HTML 响应</p>"
    return HTMLResponse(content)

# 定义处理函数:返回纯文本响应
async def text_response(request):
    # PlainTextResponse 用于返回纯文本内容
    return PlainTextResponse("这是纯文本响应")

# 定义处理函数:重定向
async def redirect_example(request):
    # RedirectResponse 用于重定向到其他 URL
    # status_code=302 表示临时重定向
    return RedirectResponse(url="/json", status_code=302)

# 定义处理函数:自定义响应
async def custom_response(request):
    # Response 是通用响应类,可以自定义状态码、内容类型等
    content = b"Custom binary data"  # 二进制数据
    return Response(
        content=content,  # 响应内容
        status_code=201,  # HTTP 状态码
        media_type="application/octet-stream",  # 内容类型
        headers={"X-Custom": "Header"}  # 自定义响应头
    )

# 创建应用
app = Starlette(
    debug=True,
    routes=[
        Route("/json", json_response),
        Route("/html", html_response),
        Route("/text", text_response),
        Route("/redirect", redirect_example),
        Route("/custom", custom_response),
    ]
)

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

测试方法:

  • 访问 http://localhost:8000/json,会返回 JSON 数据
  • 访问 http://localhost:8000/html,会返回 HTML 页面
  • 访问 http://localhost:8000/text,会返回纯文本
  • 访问 http://localhost:8000/redirect,会重定向到 /json

9. 完整的 CRUD 示例:增删改查 #

让我们创建一个完整的 CRUD(Create、Read、Update、Delete)示例,展示如何使用 Starlette 构建简单的 API。

# 导入必要的模块
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
from starlette.exceptions import HTTPException

# 模拟数据库(实际应用中应该使用真实数据库)
items_db = {}
next_id = 1

# 定义处理函数:获取所有商品
async def get_all_items(request):
    # 返回所有商品
    return JSONResponse({"items": list(items_db.values())})

# 定义处理函数:根据 ID 获取单个商品
async def get_item(request):
    # 获取路径参数
    item_id = int(request.path_params["item_id"])
    # 查找商品
    if item_id in items_db:
        return JSONResponse(items_db[item_id])
    else:
        # 如果商品不存在,抛出 404 错误
        raise HTTPException(status_code=404, detail="商品不存在")

# 定义处理函数:创建新商品
async def create_item(request):
    global next_id
    # 检查请求方法
    if request.method == "POST":
        # 读取请求体
        data = await request.json()
        # 创建新商品
        new_item = {
            "id": next_id,
            "name": data.get("name", ""),
            "price": data.get("price", 0.0)
        }
        # 保存到"数据库"
        items_db[next_id] = new_item
        next_id += 1
        # 返回创建的商品
        return JSONResponse(new_item, status_code=201)
    else:
        # 如果不是 POST 请求,返回所有商品
        return await get_all_items(request)

# 定义处理函数:更新商品
async def update_item(request):
    # 获取路径参数
    item_id = int(request.path_params["item_id"])
    # 检查请求方法
    if request.method == "PUT":
        # 检查商品是否存在
        if item_id not in items_db:
            raise HTTPException(status_code=404, detail="商品不存在")
        # 读取请求体
        data = await request.json()
        # 更新商品信息
        items_db[item_id].update(data)
        # 返回更新后的商品
        return JSONResponse(items_db[item_id])
    else:
        # 如果不是 PUT 请求,返回商品信息
        return await get_item(request)

# 定义处理函数:删除商品
async def delete_item(request):
    # 获取路径参数
    item_id = int(request.path_params["item_id"])
    # 检查商品是否存在
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="商品不存在")
    # 删除商品
    deleted_item = items_db.pop(item_id)
    # 返回删除的商品信息
    return JSONResponse({
        "message": "删除成功",
        "deleted_item": deleted_item
    })

# 创建应用
app = Starlette(
    debug=True,
    routes=[
        # GET /items/ - 获取所有商品
        # POST /items/ - 创建新商品
        Route("/items/", create_item, methods=["GET", "POST"]),
        # GET /items/{item_id} - 获取单个商品
        # PUT /items/{item_id} - 更新商品
        Route("/items/{item_id}", update_item, methods=["GET", "PUT"]),
        # DELETE /items/{item_id} - 删除商品
        Route("/items/{item_id}", delete_item, methods=["DELETE"]),
    ]
)

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

测试方法(使用 curl):

# 创建商品
curl -X POST "http://localhost:8000/items/" \
  -H "Content-Type: application/json" \
  -d '{"name": "苹果", "price": 5.5}'

# 获取所有商品
curl "http://localhost:8000/items/"

# 获取单个商品
curl "http://localhost:8000/items/1"

# 更新商品
curl -X PUT "http://localhost:8000/items/1" \
  -H "Content-Type: application/json" \
  -d '{"name": "红苹果", "price": 6.0}'

# 删除商品
curl -X DELETE "http://localhost:8000/items/1"

10. 错误处理:HTTPException #

当发生错误时,Starlette 提供了 HTTPException 来返回标准的 HTTP 错误响应。

# 导入必要的模块
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
from starlette.exceptions import HTTPException

# 模拟用户数据库
users_db = {
    1: {"id": 1, "username": "张三", "email": "zhangsan@example.com"},
    2: {"id": 2, "username": "李四", "email": "lisi@example.com"}
}

# 定义处理函数:获取用户信息
async def get_user(request):
    # 获取路径参数
    user_id = int(request.path_params["user_id"])
    # 检查用户是否存在
    if user_id not in users_db:
        # 抛出 404 错误(资源不存在)
        raise HTTPException(
            status_code=404,
            detail=f"用户 ID {user_id} 不存在"
        )
    # 返回用户信息
    return JSONResponse(users_db[user_id])

# 定义处理函数:创建用户(带验证)
async def create_user(request):
    if request.method == "POST":
        # 读取请求体
        data = await request.json()
        # 简单验证:检查用户名是否已存在
        username = data.get("username", "")
        for existing_user in users_db.values():
            if existing_user["username"] == username:
                # 抛出 400 错误(请求错误)
                raise HTTPException(
                    status_code=400,
                    detail="用户名已存在"
                )
        # 创建新用户
        new_id = max(users_db.keys()) + 1 if users_db else 1
        new_user = {
            "id": new_id,
            "username": username,
            "email": data.get("email", "")
        }
        users_db[new_id] = new_user
        # 返回创建的用户信息
        return JSONResponse(new_user, status_code=201)
    else:
        # 如果不是 POST 请求,返回所有用户
        return JSONResponse({"users": list(users_db.values())})

# 创建应用
app = Starlette(
    debug=True,
    routes=[
        Route("/users/{user_id}", get_user),
        Route("/users/", create_user, methods=["GET", "POST"]),
    ]
)

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

常见 HTTP 状态码:

  • 200:成功
  • 201:创建成功
  • 400:请求错误(如参数验证失败)
  • 404:资源不存在
  • 500:服务器内部错误

11. 请求对象:获取请求的详细信息 #

request 对象包含了请求的所有信息,让我们学习如何访问这些信息。

# 导入必要的模块
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

# 定义处理函数:获取请求的详细信息
async def request_info(request):
    # 获取请求方法(GET、POST、PUT、DELETE 等)
    method = request.method
    # 获取完整的 URL
    url = str(request.url)
    # 获取请求头(转换为字典)
    headers = dict(request.headers)
    # 获取客户端 IP 地址
    client_host = request.client.host if request.client else "unknown"
    # 获取查询参数(转换为字典)
    query_params = dict(request.query_params)
    # 获取路径参数(字典)
    path_params = dict(request.path_params)
    # 返回所有信息
    return JSONResponse({
        "method": method,
        "url": url,
        "client_host": client_host,
        "headers": headers,
        "query_params": query_params,
        "path_params": path_params
    })

# 定义处理函数:读取不同类型的请求体
async def read_body(request):
    # 获取 Content-Type 请求头
    content_type = request.headers.get("content-type", "")
    result = {"content_type": content_type}
    # 根据 Content-Type 读取不同的数据
    if "application/json" in content_type:
        # 读取 JSON 格式的请求体
        result["data"] = await request.json()
    elif "application/x-www-form-urlencoded" in content_type:
        # 读取表单格式的请求体
        form_data = await request.form()
        result["data"] = dict(form_data)
    else:
        # 读取原始请求体(字节)
        body = await request.body()
        result["data"] = body.decode("utf-8")
    return JSONResponse(result)

# 创建应用
app = Starlette(
    debug=True,
    routes=[
        Route("/info", request_info),
        Route("/body", read_body, methods=["POST"]),
    ]
)

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

测试方法:

# 获取请求信息
curl "http://localhost:8000/info?name=test&age=25"

# 发送 JSON 请求体
curl -X POST "http://localhost:8000/body" \
  -H "Content-Type: application/json" \
  -d '{"key": "value"}'

12. 应用生命周期:启动和关闭事件 #

Starlette 支持在应用启动和关闭时执行特定的操作,这对于初始化数据库连接、清理资源等非常有用。

# 导入必要的模块
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

# 定义启动事件:应用启动时执行
async def startup_event():
    # 这里可以执行初始化操作,如连接数据库、加载配置等
    print("应用正在启动...")
    print("初始化数据库连接...")
    print("加载配置文件...")
    print("应用启动完成!")

# 定义关闭事件:应用关闭时执行
async def shutdown_event():
    # 这里可以执行清理操作,如关闭数据库连接、保存数据等
    print("应用正在关闭...")
    print("关闭数据库连接...")
    print("保存数据...")
    print("应用已关闭!")

# 定义处理函数
async def homepage(request):
    return JSONResponse({"message": "Hello Starlette!"})

# 创建应用
# on_startup 参数指定启动时执行的函数列表
# on_shutdown 参数指定关闭时执行的函数列表
app = Starlette(
    debug=True,
    routes=[
        Route("/", homepage),
    ],
    on_startup=[startup_event],  # 启动事件
    on_shutdown=[shutdown_event]  # 关闭事件
)

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

说明:

  • 启动事件在应用启动时执行一次
  • 关闭事件在应用关闭时执行一次(按 Ctrl+C 停止服务器时会触发)
  • 可以注册多个启动和关闭事件

13. 常见错误与最佳实践 #

13.1 常见错误 #

错误 1:忘记使用 await #

# 错误示例:忘记使用 await
async def read_body(request):
    data = request.json()  # 错误:应该使用 await request.json()
    return JSONResponse(data)

# 正确示例
async def read_body(request):
    data = await request.json()  # 正确:使用 await
    return JSONResponse(data)

错误 2:路径参数类型错误 #

# 错误示例:路径参数默认是字符串,需要手动转换
async def get_user(request):
    user_id = request.path_params["user_id"]
    # 如果直接用于数值计算会出错
    next_id = user_id + 1  # 错误:字符串不能直接相加

# 正确示例
async def get_user(request):
    user_id = int(request.path_params["user_id"])  # 转换为整数
    next_id = user_id + 1  # 正确
    return JSONResponse({"user_id": user_id, "next_id": next_id})

错误 3:未处理不存在的资源 #

# 错误示例:没有检查资源是否存在
async def get_item(request):
    item_id = int(request.path_params["item_id"])
    item = items_db[item_id]  # 如果 item_id 不存在会抛出 KeyError
    return JSONResponse(item)

# 正确示例
async def get_item(request):
    item_id = int(request.path_params["item_id"])
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="商品不存在")
    item = items_db[item_id]
    return JSONResponse(item)

13.2 最佳实践 #

  1. 使用 async/await:充分利用异步编程的优势
  2. 错误处理:使用 HTTPException 返回标准的错误响应
  3. 类型转换:路径参数默认是字符串,需要时手动转换
  4. 资源检查:访问资源前先检查是否存在
  5. 代码组织:将路由、处理函数等分离到不同文件(项目较大时)

14. Starlette 与 FastAPI 的关系 #

FastAPI 是基于 Starlette 构建的,它们的关系如下:

# FastAPI = Starlette + Pydantic + 自动文档生成

# Starlette 提供:
# - 基础的 ASGI 应用框架
# - 路由系统
# - 请求/响应处理
# - 中间件支持

# FastAPI 在此基础上添加:
# - 基于 Pydantic 的数据验证
# - 自动 API 文档(Swagger/ReDoc)
# - 依赖注入系统
# - 更多高级特性

选择建议:

  • 使用 FastAPI:如果你需要数据验证、自动文档等高级特性
  • 使用 Starlette:如果你需要更精细的控制,或者项目很简单

15. 总结 #

15.1 核心概念回顾 #

  • Starlette:轻量级的 ASGI 框架
  • ASGI:异步服务器网关接口
  • 路由:使用 Route 将 URL 路径映射到处理函数
  • 路径参数:URL 路径中的变量(如 /users/{user_id})
  • 查询参数:URL 中 ? 后面的参数
  • 请求体:POST/PUT 请求中发送的数据
  • 响应类型:JSONResponse、HTMLResponse 等
  • HTTPException:用于返回 HTTP 错误响应

15.2 基本使用流程 #

  1. 导入必要的模块(Starlette、JSONResponse、Route 等)
  2. 定义处理函数(使用 async def)
  3. 创建路由列表(使用 Route)
  4. 创建 Starlette 应用实例
  5. 运行应用(使用 uvicorn)
← 上一节 ssl 下一节 text_splitter →

访问验证

请输入访问令牌

Token不正确,请重新输入