1. Starlette 是什么? #
Starlette 是一个轻量级的 ASGI(异步服务器网关接口)框架,它是构建高性能异步 Web 应用的基础。FastAPI 就是基于 Starlette 构建的。
1.1 为什么需要 Starlette? #
如果你已经学过 FastAPI,可能会问:为什么还要学 Starlette?
简单理解:
- FastAPI:高级框架,提供了很多便利功能(自动文档、数据验证等)
- Starlette:底层框架,提供了更基础、更灵活的控制
使用 Starlette 的场景:
- 需要更精细的控制(不需要 FastAPI 的高级特性)
- 学习 ASGI 的工作原理
- 构建轻量级的异步服务
- 开发自定义框架或中间件
1.2 Starlette 的核心特点 #
- 极简设计:只提供核心功能,不包含太多高级抽象
- 高性能:专为异步编程设计,性能优秀
- ASGI 原生:完全兼容 ASGI 标准
- 模块化:可以独立使用各个组件
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)运行方法:
- 将代码保存为
hello.py - 运行
python hello.py - 在浏览器访问
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 最佳实践 #
- 使用 async/await:充分利用异步编程的优势
- 错误处理:使用
HTTPException返回标准的错误响应 - 类型转换:路径参数默认是字符串,需要时手动转换
- 资源检查:访问资源前先检查是否存在
- 代码组织:将路由、处理函数等分离到不同文件(项目较大时)
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 基本使用流程 #
- 导入必要的模块(
Starlette、JSONResponse、Route等) - 定义处理函数(使用
async def) - 创建路由列表(使用
Route) - 创建 Starlette 应用实例
- 运行应用(使用 uvicorn)