（十七）把请求处理组合成完整应用
1. 应用对象保存配置，入口按请求调用
现在各个职责已经有对应函数，把它们组成一个 BookApp：
# 项目文件：web_project/src/webbook/app.py
# WSGI入口组合请求、路由、事务和响应；数据库异常回滚，客户端收到可识别的失败状态。
import logging
import sqlite3
import time
from pathlib import Path
from .cors import CorsPolicy
from .database import connect
from .http import Problem, Request, Response, one_value, read_form, read_json
from .render import render_home, render_login
from .routing import route
from .security import (
authenticate,
load_session,
new_session,
require_csrf,
require_user,
rotate_session,
session_cookie,
)
from .service import as_book, create_book, finish_book, list_books, owned_book
from .validation import form_book, validate_book
STATIC = Path(__file__).resolve().parent / "static"
logger = logging.getLogger(__name__)
class BookApp:
def __init__(
self,
database,
origin="http://127.0.0.1:8767",
cors_origins=None,
secure_cookie=False,
):
self.database, self.origin = database, origin
self.cors = CorsPolicy(set(cors_origins or []))
self.secure_cookie = secure_cookie
self.now = time.time实例保存数据库位置、应用来源、跨源许可与 Cookie 配置，now 提供当前时间。实际请求调用 __call__，沿用第四章可调用对象的知识：服务器可以像调用函数一样调用这个应用实例。
# 续写同一文件：web_project/src/webbook/app.py
# 先处理预检和静态文件，再分发业务请求；异常也统一转换成可发送的响应。
def __call__(self, environ, start_response):
request = None
try:
request = Request.read(environ)
response = self.cors.preflight(request)
if response is None:
handler, parameters, _ = route(request.method, request.path)
if handler == "asset":
name = parameters["name"]
kind = "text/css" if name.endswith("css") else "text/javascript"
response = Response.text(
(STATIC / name).read_text(encoding="utf-8"),
kind=kind + "; charset=utf-8",
)
else:
response = self.handle(request, handler, parameters)
except Problem as error:
response = self.problem_response(
request, error, environ.get("PATH_INFO", "")
)
except sqlite3.Error:
logger.exception("书籍请求的数据库操作失败")
response = self.problem_response(
request, Problem(500, "本次操作未完成，请稍后重试")
)
if request is not None:
self.cors.decorate(request, response)
return response.send(start_response)先读取请求并处理预检，再匹配路由；静态资源读出文件，业务请求进入 handle。已知 Problem 转为对应响应；数据库异常记录完整服务器日志，再交回不包含内部路径的 500。
# 续写同一文件：web_project/src/webbook/app.py
# API 返回 JSON 错误对象，页面返回可读的 HTML 提示；两者保留相同状态码。
def problem_response(self, request, error, fallback_path=""):
path = request.path if request is not None else fallback_path
if path.startswith("/api/"):
return Response.json(
{"error": error.message, "fields": error.fields},
error.status,
error.headers,
)
from html import escape
return Response.text(
f'<h1>{error.status}</h1><p>{escape(error.message)}</p><a href="/">返回书架</a>',
error.status,
headers=error.headers,
)API 错误用 JSON 保留 error 与 fields，页面错误用 HTML 显示说明；解析请求时失败，也按原来的路径选择响应形式。
2. 会话、动作与事务在一条请求中结束
# 续写同一文件：web_project/src/webbook/app.py
def handle(self, request, handler, parameters):
now = int(self.now())
with connect(self.database) as connection:
session = load_session(connection, request.headers.get("cookie", ""), now)
fresh = session is None
session = session or new_session(connection, now)
if handler in {"home", "login_page", "session_info"}:
response = self.read_action(connection, session, handler)
elif handler == "login":
response, updated = self.login_action(connection, request, session, now)
fresh = fresh or updated.sid != session.sid
session = updated
elif handler == "logout":
require_user(session)
form = read_form(request)
require_csrf(
request,
session,
one_value(form, "csrf_token"),
self.cors.origins | {self.origin},
)
session = rotate_session(connection, session, now)
fresh, response = True, Response.redirect("/")
else:
require_user(session)
response = self.book_action(
connection, request, session, handler, parameters
)
# 事务已经提交，才把新会话编号交回浏览器。
if fresh:
response.headers.append(session_cookie(session, now, self.secure_cookie))
return responsehandle 在本次连接中取得会话，再按动作处理页面、登录、退出或书籍。响应对象先形成，事务退出完成提交，随后才把新会话 Cookie 放进响应；若数据库操作失败，异常先让事务回滚，再由外层入口处理。
# 续写同一文件：web_project/src/webbook/app.py
# 读取页面时只查当前账户的书籍；登录成功后替换会话，再跳转回书架。
def read_action(self, connection, session, handler):
if handler == "login_page":
return Response.text(render_login(session))
if handler == "session_info":
return Response.json(
{"username": session.username, "csrf_token": session.csrf}
)
books = (
list_books(connection, session.user_id)
if session.user_id is not None
else []
)
return Response.text(render_home(session, books))
def login_action(self, connection, request, session, now):
form = read_form(request)
require_csrf(
request,
session,
one_value(form, "csrf_token"),
self.cors.origins | {self.origin},
)
try:
user = authenticate(
connection, one_value(form, "username"), one_value(form, "password")
)
except Problem as error:
response = Response.text(
render_login(session, error.message, one_value(form, "username")),
error.status,
)
return response, session
return Response.redirect("/"), rotate_session(connection, session, now, user)read_action 只读取页面或会话信息。login_action 先核对表单令牌，再核对密码；失败时保留原会话并显示账号，成功时交回新会话与重定向。
3. 两种输入最终进入同一个新增动作
# 续写同一文件：web_project/src/webbook/app.py
# 先按操作识别表单或 JSON，再核对 CSRF、校验数据并调用书籍服务。
def book_action(self, connection, request, session, handler, parameters):
if handler == "book_list":
flag = one_value(request.query, "finished", "all")
if flag not in {"all", "0", "1"}:
raise Problem(422, "finished查询应为all、0或1")
books = list_books(
connection, session.user_id, None if flag == "all" else int(flag)
)
return Response.json({"items": books})
if handler == "book_detail":
return Response.json(
as_book(owned_book(connection, parameters["book_id"], session.user_id))
)
form = read_form(request) if handler == "create_form" else None
token = (
one_value(form, "csrf_token")
if form is not None
else request.headers.get("x-csrf-token", "")
)
require_csrf(request, session, token, self.cors.origins | {self.origin})
if handler == "finish":
return Response.json(
finish_book(connection, session.user_id, parameters["book_id"])
)
try:
draft = (
form_book(form)
if form is not None
else validate_book(read_json(request))
)
book = create_book(connection, session.user_id, draft)
except Problem as error:
if form is None:
raise
values = {
key: one_value(form, key) for key in ("title", "tags", "finished")
}
message = error.message + "；" + "；".join(error.fields.values())
return Response.text(
render_home(
session, list_books(connection, session.user_id), message, values
),
error.status,
)
if form is not None:
return Response.redirect("/")
return Response.json(book, 201, [("Location", f"/api/books/{book['id']}")])列表查询与详情读取各自处理自己的输入；写入先核对会话令牌。表单走 form_book，JSON 走 read_json 与 validate_book，随后都调用 create_book。
表单校验失败时把输入与字段说明放回页面，让读者修改；JSON 失败则继续交给统一错误响应。成功后，表单返回 303，API 返回 201 与 Location。两种响应形式共用同一条书籍记录。
4. 命令入口连接服务器与应用
# 项目文件：web_project/src/webbook/cli.py
# 初始化演示账号与采集CSV；开发服务器把HTTP请求转换成WSGI调用。
import argparse
import logging
from pathlib import Path
from wsgiref.simple_server import make_server
from .app import BookApp
from .database import initialize
def main(argv=None):
parser = argparse.ArgumentParser(description="Web共同基础：阅读书架")
parser.add_argument("action", choices=["init", "serve"], nargs="?", default="serve")
parser.add_argument("--db", type=Path, default=Path("data/library.sqlite3"))
parser.add_argument("--csv", type=Path, default=Path("data/books.csv"))
parser.add_argument("--port", type=int, default=8767)
parser.add_argument("--cors-origin", action="append", default=[])
args = parser.parse_args(argv)
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
if args.action == "init":
initialize(args.db, args.csv)
print("已初始化alice、bob和采集书目：", args.db)
return 0
origin = f"http://127.0.0.1:{args.port}"
application = BookApp(args.db, origin, args.cors_origin)
with make_server("127.0.0.1", args.port, application) as server:
print("打开 " + origin, flush=True)
try:
server.serve_forever()
except KeyboardInterrupt:
print("服务器已停止")
return 0init 建立账号和采集记录，serve 构造 BookApp，再把实例交给 make_server。origin 从本次启动配置取得，写入来源核对不会随任意 Host 请求头改变。
# 项目文件：web_project/src/webbook/__init__.py
# 导入只定义包，不启动服务器、不创建数据库。
"""Web应用共同基础练习。"""# 项目文件：web_project/src/webbook/__main__.py
# 模块入口和web-books调用同一个main。
from .cli import main
raise SystemExit(main())导入模块只取得定义，调用命令才运行初始化或启动服务器。数据库文件保存记录与会话，重开应用仍能从同一份文件读取。