构建文章管理应用
前 12 个专题分别解决了请求处理、页面呈现和应用运行中的问题。现在把这些能力放到同一个项目里:用户查看文章页面,脚本通过 API 创建文章,应用统一处理错误并在响应后写日志;WebSocket 页面提供消息回显,测试文件覆盖关键路径。
完整项目代码与本专题中的步骤对应。文章数据保存在内存中,服务重启后新增文章会消失;这个约束让示例集中展示 Starlette 的各部分如何协作。
明确验收结果
| 请求 | 预期行为 |
|---|---|
GET /、GET /articles、GET /articles/{article_id:int} | 返回首页、列表和详情页 |
GET /articles/new | 返回创建提示页 |
GET /api/articles、POST /api/articles | 读取与创建文章;空标题返回 400 |
GET /app-info | 查看 Lifespan 准备的配置和文章数量 |
GET /ws-page、WS /ws | 打开消息页面并验证回显 |
GET /static/style.css | 加载页面样式 |
文章不存在时返回 404。中间件为请求记录日志、统计耗时并补充响应头;创建成功后,后台任务向日志文件写入记录。TestClient 会验证这些行为。
项目使用 article_management.py 作为教学入口,旁边放置 templates/、static/、tests/ 和 requirements.txt。实现顺序与下面的步骤一致:先准备运行环境,再接入状态、页面和 API,最后补全错误处理、公共处理、后台任务、WebSocket 与测试。
第一步:准备结构与依赖
先准备目录:
starlette-blog/
├── article_management.py
├── requirements.txt
├── logs/
├── tests/
├── templates/
└── static/安装依赖:
python -m pip install -r requirements.txtrequirements.txt:
starlette==1.6.0
uvicorn[standard]>=0.30,<1
jinja2>=3.1,<4
pytest>=8,<10
httpx2>=2.13,<3这些依赖分别用于:
starlette:框架本身
uvicorn:运行 ASGI 应用
jinja2:模板渲染
pytest:运行测试
httpx2:TestClient 相关依赖第二步:初始化应用状态
这一部分负责:
初始化配置
初始化文章数据
确保日志目录存在示例:
from contextlib import asynccontextmanager
from pathlib import Path
BASE_DIR = Path(__file__).resolve().parent
LOG_DIR = BASE_DIR / "logs"
@asynccontextmanager
async def lifespan(app):
LOG_DIR.mkdir(exist_ok=True)
app.state.config = {
"app_name": "Starlette 文章管理应用",
"version": "1.0.0",
}
app.state.articles = [
{
"id": 1,
"title": "认识 Starlette 应用对象",
"summary": "学习如何创建最小应用。",
"content": "Starlette 应用对象是整个 Web 程序的入口。",
},
{
"id": 2,
"title": "学习路由和页面渲染",
"summary": "学习页面路由和模板。",
"content": "路由负责分发请求,模板负责渲染页面。",
},
]
print("应用启动:配置、文章数据和日志目录已初始化")
yield
print("应用关闭:清理完成")这一段的重点不是语法,而是职责边界:
启动时准备资源
关闭时做简单清理第三步:编写页面与模板
这一部分负责 4 个页面:
首页
文章列表页
文章详情页
文章创建提示页路由示例:
async def homepage(request):
config = request.app.state.config
return templates.TemplateResponse(
request,
"index.html",
{
"app_name": config["app_name"],
"version": config["version"],
},
)
async def article_list(request):
return templates.TemplateResponse(
request,
"article_list.html",
{
"articles": request.app.state.articles,
},
)
async def article_detail(request):
article_id = request.path_params["article_id"]
article = find_article(request.app.state.articles, article_id)
if article is None:
raise HTTPException(status_code=404, detail="文章不存在")
return templates.TemplateResponse(
request,
"article_detail.html",
{
"article": article,
},
)
async def article_new_page(request):
return templates.TemplateResponse(
request,
"article_new.html",
{},
)这一部分的重点:
页面路由返回模板
详情页需要按 ID 查找文章
文章不存在时抛出 404第四步:编写文章 API
本章 API 至少包含两个动作:
获取文章 JSON 列表
创建文章可以继续用函数视图,也可以用 HTTPEndpoint。 如果想顺便复习类视图,可以这样写:
from starlette.endpoints import HTTPEndpoint
from starlette.responses import JSONResponse
class ArticleAPI(HTTPEndpoint):
async def get(self, request):
return JSONResponse({"items": request.app.state.articles})
async def post(self, request):
data = await request.json()
title = data.get("title", "").strip()
summary = data.get("summary", "这是一篇新文章。").strip()
content = data.get("content", "暂无正文内容。").strip()
if not title:
raise HTTPException(status_code=400, detail="文章标题不能为空")
articles = request.app.state.articles
article = {
"id": len(articles) + 1,
"title": title,
"summary": summary,
"content": content,
}
articles.append(article)
return JSONResponse(
{
"message": "文章创建成功",
"article": article,
},
status_code=201,
)这一部分的重点:
GET 返回 JSON 列表
POST 接收 JSON 并创建文章
标题为空时抛出 400第五步:统一错误响应
如果不统一处理异常,错误响应的风格会比较散。
本章建议至少统一处理:
400
404
500示例:
from starlette.responses import JSONResponse
async def http_exception_handler(request, exc):
return JSONResponse(
{
"code": exc.status_code,
"message": exc.detail,
},
status_code=exc.status_code,
)
async def server_error_handler(request, exc):
return JSONResponse(
{
"code": 500,
"message": "服务器内部错误",
},
status_code=500,
)重点:
正常情况返回正常内容
异常情况返回统一 JSON 结构第六步:添加中间件
本章中间件只做教学常用的三件事:
记录请求路径
统计耗时
补充统一响应头示例:
import time
async def log_middleware(request, call_next):
print(f"请求开始:{request.method} {request.url.path}")
response = await call_next(request)
print(f"请求结束:{request.method} {request.url.path}")
return response
async def timing_middleware(request, call_next):
start = time.perf_counter()
response = await call_next(request)
duration = time.perf_counter() - start
response.headers["X-Process-Time"] = f"{duration:.6f}"
return response
async def header_middleware(request, call_next):
response = await call_next(request)
response.headers["X-App-Name"] = "Starlette Blog"
return response重点:
中间件围绕每个请求执行
适合做公共逻辑
不适合写具体业务第七步:添加后台任务
创建文章成功后,我们希望顺手记录一条日志。
这类动作适合后台任务。
示例:
from starlette.background import BackgroundTask
def write_article_log(message):
log_file = LOG_DIR / "article.log"
with log_file.open("a", encoding="utf-8") as file:
file.write(message + "\n")
class ArticleAPI(HTTPEndpoint):
async def post(self, request):
...
task = BackgroundTask(write_article_log, f"创建文章:{title}")
return JSONResponse(
{
"message": "文章创建成功",
"article": article,
},
status_code=201,
background=task,
)重点:
核心业务先完成
日志这种轻量附加动作放到响应之后第八步:添加 WebSocket 回显
本章不做复杂聊天室,只做一个最小回显页。
示例:
from starlette.websockets import WebSocket, WebSocketDisconnect
async def websocket_page(request):
return templates.TemplateResponse(
request,
"ws.html",
{"title": "WebSocket 消息演示"},
)
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
message = await websocket.receive_text()
await websocket.send_text(f"服务端回显:{message}")
except WebSocketDisconnect:
print("客户端断开 WebSocket 连接")重点:
HTTP 页面负责提供界面
WebSocket 路由负责持续收发消息第九步:整合应用入口
到这里,完整 article_management.py 至少需要把这些内容放在一起:
依赖导入
模板对象
Lifespan
工具函数
页面路由
API 路由
WebSocket 路由
异常处理器
中间件
routes 列表
exception_handlers
app = Starlette(...)推荐保留一个相对完整但仍然易读的版本。示例结构如下:
import time
from contextlib import asynccontextmanager
from pathlib import Path
from starlette.applications import Starlette
from starlette.background import BackgroundTask
from starlette.endpoints import HTTPEndpoint
from starlette.exceptions import HTTPException
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
from starlette.routing import Mount, Route, WebSocketRoute
from starlette.staticfiles import StaticFiles
from starlette.templating import Jinja2Templates
from starlette.websockets import WebSocket, WebSocketDisconnect
BASE_DIR = Path(__file__).resolve().parent
LOG_DIR = BASE_DIR / "logs"
LOG_FILE = LOG_DIR / "article.log"
templates = Jinja2Templates(directory="templates")
@asynccontextmanager
async def lifespan(app):
LOG_DIR.mkdir(exist_ok=True)
app.state.config = {
"app_name": "Starlette 文章管理应用",
"version": "1.0.0",
}
app.state.articles = [
{
"id": 1,
"title": "认识 Starlette 应用对象",
"summary": "学习如何创建最小应用。",
"content": "Starlette 应用对象是整个 Web 程序的入口。",
},
{
"id": 2,
"title": "学习路由和页面渲染",
"summary": "学习页面路由和模板。",
"content": "路由负责分发请求,模板负责渲染页面。",
},
]
print("应用启动:配置、文章数据和日志目录已初始化")
yield
print("应用关闭:清理完成")
def find_article(articles, article_id):
for article in articles:
if article["id"] == article_id:
return article
return None
def write_article_log(message):
LOG_DIR.mkdir(exist_ok=True)
with LOG_FILE.open("a", encoding="utf-8") as file:
file.write(message + "\n")
class LogMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
print(f"请求开始:{request.method} {request.url.path}")
response = await call_next(request)
print(f"请求结束:{request.method} {request.url.path}")
return response
class TimingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
start = time.perf_counter()
response = await call_next(request)
duration = time.perf_counter() - start
response.headers["X-Process-Time"] = f"{duration:.6f}"
return response
class HeaderMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
response = await call_next(request)
response.headers["X-App-Name"] = "Starlette Blog"
return response
async def homepage(request):
config = request.app.state.config
return templates.TemplateResponse(
request,
"index.html",
{
"app_name": config["app_name"],
"version": config["version"],
},
)
async def article_list(request):
return templates.TemplateResponse(
request,
"article_list.html",
{
"articles": request.app.state.articles,
},
)
async def article_detail(request):
article_id = request.path_params["article_id"]
article = find_article(request.app.state.articles, article_id)
if article is None:
raise HTTPException(status_code=404, detail="文章不存在")
return templates.TemplateResponse(
request,
"article_detail.html",
{
"article": article,
},
)
async def article_new_page(request):
return templates.TemplateResponse(request, "article_new.html", {})
class ArticleAPI(HTTPEndpoint):
async def get(self, request):
return JSONResponse({"items": request.app.state.articles})
async def post(self, request):
data = await request.json()
title = data.get("title", "").strip()
summary = data.get("summary", "这是一篇新文章。").strip()
content = data.get("content", "暂无正文内容。").strip()
if not title:
raise HTTPException(status_code=400, detail="文章标题不能为空")
articles = request.app.state.articles
article = {
"id": len(articles) + 1,
"title": title,
"summary": summary,
"content": content,
}
articles.append(article)
task = BackgroundTask(write_article_log, f"创建文章:{title}")
return JSONResponse(
{
"message": "文章创建成功",
"article": article,
},
status_code=201,
background=task,
)
async def app_info(request):
return JSONResponse(
{
"config": request.app.state.config,
"article_count": len(request.app.state.articles),
"log_file": str(LOG_FILE),
}
)
async def websocket_page(request):
return templates.TemplateResponse(
request,
"ws.html",
{"title": "WebSocket 消息演示"},
)
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
message = await websocket.receive_text()
await websocket.send_text(f"服务端回显:{message}")
except WebSocketDisconnect:
print("客户端断开 WebSocket 连接")
async def http_exception_handler(request, exc):
return JSONResponse(
{
"code": exc.status_code,
"message": exc.detail,
},
status_code=exc.status_code,
)
async def server_error_handler(request, exc):
return JSONResponse(
{
"code": 500,
"message": "服务器内部错误",
},
status_code=500,
)
routes = [
Route("/", homepage),
Route("/articles", article_list),
Route("/articles/new", article_new_page),
Route("/articles/{article_id:int}", article_detail),
Route("/api/articles", ArticleAPI),
Route("/app-info", app_info),
Route("/ws-page", websocket_page),
WebSocketRoute("/ws", websocket_endpoint),
Mount("/static", app=StaticFiles(directory="static"), name="static"),
]
middleware = [
Middleware(LogMiddleware),
Middleware(TimingMiddleware),
Middleware(HeaderMiddleware),
]
exception_handlers = {
HTTPException: http_exception_handler,
500: server_error_handler,
}
app = Starlette(
routes=routes,
middleware=middleware,
exception_handlers=exception_handlers,
lifespan=lifespan,
)
if __name__ == "__main__":
import uvicorn
uvicorn.run("article_management:app", host="127.0.0.1", port=8000, reload=True)请求处理流程
可以把这份应用理解成:
启动时初始化配置、数据和日志目录
↓
请求进入中间件
↓
路由分发到页面、API 或 WebSocket
↓
正常返回 HTML、JSON 或 WebSocket 消息
↓
异常时进入统一错误处理器
↓
创建文章后在后台写日志第十步:补齐模板与样式
这一部分补齐页面展示层。
templates/index.html
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>{{ app_name }}</title>
<link rel="stylesheet" href="/static/style.css">
</head>
<body>
<div class="container">
<h1>{{ app_name }}</h1>
<p>当前版本:{{ version }}</p>
<p><a href="/articles">查看文章列表</a></p>
<p><a href="/articles/new">查看创建文章说明</a></p>
<p><a href="/ws-page">进入 WebSocket 演示页面</a></p>
<p><a href="/app-info">查看应用信息 JSON</a></p>
</div>
</body>
</html>templates/article_list.html
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>文章列表</title>
<link rel="stylesheet" href="/static/style.css">
</head>
<body>
<div class="container">
<h1>文章列表</h1>
{% if articles %}
<ul class="article-list">
{% for article in articles %}
<li class="article-card">
<h2><a href="/articles/{{ article.id }}">{{ article.title }}</a></h2>
<p class="article-summary">{{ article.summary }}</p>
</li>
{% endfor %}
</ul>
{% else %}
<p>暂无文章。</p>
{% endif %}
<p><a href="/">返回首页</a></p>
</div>
</body>
</html>templates/article_detail.html
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>{{ article.title }}</title>
<link rel="stylesheet" href="/static/style.css">
</head>
<body>
<div class="container">
<article class="article-card">
<h1>{{ article.title }}</h1>
<p class="article-summary">{{ article.summary }}</p>
<div class="article-content">{{ article.content }}</div>
</article>
<p>
<a href="/articles">返回文章列表</a>
|
<a href="/">返回首页</a>
</p>
</div>
</body>
</html>templates/article_new.html
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>创建文章说明</title>
<link rel="stylesheet" href="/static/style.css">
</head>
<body>
<div class="container">
<h1>创建文章说明</h1>
<p>本章通过 JSON 接口创建文章。</p>
<p>请向 <code>POST /api/articles</code> 发送 JSON 数据。</p>
<p><a href="/">返回首页</a></p>
</div>
</body>
</html>templates/ws.html
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>{{ title }}</title>
<link rel="stylesheet" href="/static/style.css">
</head>
<body>
<div class="container">
<h1>{{ title }}</h1>
<input id="message-input" type="text" placeholder="请输入消息">
<button onclick="sendMessage()">发送</button>
<ul id="message-list" class="message-list"></ul>
<p><a href="/">返回首页</a></p>
</div>
<script>
const socket = new WebSocket("ws://127.0.0.1:8000/ws");
socket.onopen = function () {
addMessage("连接已建立");
};
socket.onmessage = function (event) {
addMessage(event.data);
};
function sendMessage() {
const input = document.getElementById("message-input");
socket.send(input.value);
input.value = "";
}
function addMessage(message) {
const list = document.getElementById("message-list");
const item = document.createElement("li");
item.textContent = message;
list.appendChild(item);
}
</script>
</body>
</html>static/style.css
body {
margin: 0;
font-family: Arial, "Microsoft YaHei", sans-serif;
background: #f5f5f5;
color: #333;
}
.container {
width: 900px;
margin: 40px auto;
background: #fff;
padding: 24px;
border-radius: 8px;
}
h1 {
margin-top: 0;
}
a {
color: #2563eb;
text-decoration: none;
}
a:hover {
text-decoration: underline;
}
.article-list {
list-style: none;
padding: 0;
}
.article-card {
padding: 16px;
border: 1px solid #e5e7eb;
border-radius: 6px;
margin-bottom: 12px;
background: #fafafa;
}
.article-summary {
color: #666;
}
.article-content {
line-height: 1.8;
margin-top: 16px;
}
.message-list {
margin-top: 16px;
padding: 12px 24px;
background: #fafafa;
border: 1px solid #e5e7eb;
border-radius: 6px;
}页面与静态资源的协作
这一层主要负责:
把后端数据渲染成页面
提供基本导航
提供基础样式
提供一个最小 WebSocket 演示界面第十一步:验证核心功能
最后补一份基础测试,验证核心功能没有跑偏。
tests/test_article_management.py:
from starlette.testclient import TestClient
from article_management import app
def test_homepage():
with TestClient(app) as client:
response = client.get("/")
assert response.status_code == 200
assert "Starlette" in response.text
def test_article_list():
with TestClient(app) as client:
response = client.get("/articles")
assert response.status_code == 200
assert "文章列表" in response.text
def test_article_detail():
with TestClient(app) as client:
response = client.get("/articles/1")
assert response.status_code == 200
assert "认识 Starlette 应用对象" in response.text
def test_article_new_page():
with TestClient(app) as client:
response = client.get("/articles/new")
assert response.status_code == 200
assert "创建文章说明" in response.text
def test_api_articles():
with TestClient(app) as client:
response = client.get("/api/articles")
assert response.status_code == 200
data = response.json()
assert "items" in data
def test_create_article():
with TestClient(app) as client:
response = client.post(
"/api/articles",
json={
"title": "测试文章",
"summary": "测试摘要",
"content": "测试正文",
},
)
assert response.status_code == 201
data = response.json()
assert data["article"]["title"] == "测试文章"
def test_create_article_without_title():
with TestClient(app) as client:
response = client.post(
"/api/articles",
json={"title": "", "summary": "摘要", "content": "正文"},
)
assert response.status_code == 400
assert response.json()["message"] == "文章标题不能为空"
def test_app_info():
with TestClient(app) as client:
response = client.get("/app-info")
assert response.status_code == 200
data = response.json()
assert "config" in data
assert "article_count" in data
def test_static_file():
with TestClient(app) as client:
response = client.get("/static/style.css")
assert response.status_code == 200
assert "body" in response.text
def test_custom_headers():
with TestClient(app) as client:
response = client.get("/")
assert response.status_code == 200
assert "X-App-Name" in response.headers
assert "X-Process-Time" in response.headers
def test_websocket_echo():
with TestClient(app) as client:
with client.websocket_connect("/ws") as websocket:
websocket.send_text("你好")
data = websocket.receive_text()
assert data == "服务端回显:你好"测试覆盖范围
这份测试没有追求复杂,只验证最核心的行为:
页面能打开
接口能返回正确结构
错误响应能生效
Lifespan 能初始化
WebSocket 能回显运行项目
启动服务:
uvicorn article_management:app --reload建议依次查看:
http://127.0.0.1:8000/
http://127.0.0.1:8000/articles
http://127.0.0.1:8000/articles/new
http://127.0.0.1:8000/app-info
http://127.0.0.1:8000/ws-page也可发送请求查看结果:
GET /api/articles
POST /api/articles重点观察:
页面是否正常
JSON 是否正常
中间件响应头是否存在
日志文件是否写入
WebSocket 是否能回显运行测试
在项目目录中执行:
pytest如果只跑某个文件:
pytest tests/test_article_management.py如果只跑某个函数:
pytest tests/test_article_management.py::test_homepage测试通过时,说明这份综合案例的核心路径基本可用。
项目结果与扩展
到这里,页面、API、错误处理、中间件、后台任务和 WebSocket 已在同一个应用中配合工作;测试覆盖主要访问路径。文章仍保存在内存里,日志写在本地文件中。
继续扩展时,可以先把持久化数据和配置从 article_management.py 中分离,再加入登录、权限与更完整的数据校验。每增加一项能力,都应先确定它解决的具体问题,并为关键行为补上验证。