Skip to main content

Starlette

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 基础知识

你需要熟悉:

  • 函数定义defasync 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:8000http://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=0limit=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 请求中发送的数据
  • 响应类型JSONResponseHTMLResponse
  • HTTPException :用于返回 HTTP 错误响应

15.2 基本使用流程

  1. 导入必要的模块(StarletteJSONResponseRoute 等)
  2. 定义处理函数(使用 async def
  3. 创建路由列表(使用 Route
  4. 创建 Starlette 应用实例
  5. 运行应用(使用 uvicorn)