Python个人博客程序开发实例框架设计

前言写博客程序的重点不在能渲染出一个页面而在框架的分层请求怎么进来、路由怎么匹配、业务逻辑放哪、模板怎么渲染、数据怎么存取。把这几件事拆清楚代码就只有几百行却能轻松加功能拆不清楚两千行也是一团浆糊。这里有一个常见误解以为框架必须用第三方 Web 框架。其实标准库自带的http.server加上sqlite3、string.Template就足以搭出一个结构完整的小博客而且因为少了大框架的抽象层每个环节都看得见。本文用纯标准库实现一个分层清晰的博客骨架路由表、视图函数、模板、数据访问四个部分各司其职。需要事先声明边界http.server的定位是开发调试和教学演示它没有经过生产级的并发与安全加固。真要对外提供服务应该换成 WSGI 服务器配合成熟框架。本文的目的是让你理解一个 Web 框架到底在做什么而不是让你拿它去扛流量。另外安全上只说防御侧所有从页面输入进来的内容输出到 HTML 之前必须转义否则访问者提交的脚本会在别人浏览器里执行跨站脚本XSS所有 SQL 一律用参数化占位符绝不拼接。如果你手上的老教程还是 Python 2 的写法要注意BaseHTTPServer、SimpleHTTPServer这两个模块名在 Python 3 里已经不存在了统一成了http.serverprint xxx这种语句写法也必须改成print(xxx)。一、分层的职责层职责本文对应路由把「方法 路径」映射到处理函数Router类视图组装数据、返回响应view_index等函数模板把数据套进 HTMLstring.Template数据访问读写数据库connect()/init_db()服务器适配解析请求、写回响应BlogHandler关键约束视图函数不该知道self.wfile是什么。视图只返回一个状态码 内容类型 正文的响应对象由适配层负责真正写回 socket。这样视图可以被单元测试直接调用不需要起服务器。二、路由用正则匹配路径路由表就是一张列表每项是「HTTP 方法 正则 处理函数」。匹配时把路径和解码后的查询串分开路径交给正则做fullmatch查询串用urllib.parse.parse_qs解析成字典。用正则的命名分组可以把路径里的动态片段直接变成处理函数的参数比如/post/(?Ppost_id\d)匹配/post/42时处理函数就收到{post_id: 42}。三、模板为什么用string.Template拼接 HTML 有两种常见写法str.format和string.Template。推荐后者。原因是str.format会把用户内容里的花括号当成占位符来解析——用户只要在正文里写一个{渲染就会抛KeyError或IndexError。string.Template用的是$name语法且只在模板文本里解析占位符替换进去的值里出现$也不会被二次解释写法占位符用户内容含{或$时str.format{}/{0}可能抛KeyError%格式化%s需要转义%string.Template$name不受影响当然模板安全的前提是先转义再拼。html.escape()会把、、以及引号变成实体用户提交的脚本就只能当纯文本显示。四、完整实现以下代码存成blog.py即可运行需要 Python 3.8全部依赖标准库# 适用于 Python 3.8import htmlimport reimport sqlite3from dataclasses import dataclass, fieldfrom datetime import datetime, timezonefrom http.server import BaseHTTPRequestHandler, ThreadingHTTPServerfrom string import Templatefrom urllib.parse import parse_qs, urlparseDB_PATH blog.dbSCHEMA CREATE TABLE IF NOT EXISTS posts (id INTEGER PRIMARY KEY AUTOINCREMENT,title TEXT NOT NULL,body TEXT NOT NULL,created_at TEXT NOT NULL);PAGE Template(!doctype htmlhtml langzhheadmeta charsetutf-8title$title/title/headbodyh1$title/h1$content/body/html)NEW_FORM form methodpost action/newp标题: input nametitle size40/pp正文: textarea namebody rows8 cols60/textarea/ppbutton typesubmit发布/button/p/formdataclassclass Response:status: intcontent_type: strbody: str headers: dict field(default_factorydict)def connect():conn sqlite3.connect(DB_PATH, timeout5)conn.row_factory sqlite3.Rowreturn conndef init_db():with connect() as conn:conn.executescript(SCHEMA)def now_iso():return datetime.now(timezone.utc).isoformat(timespecseconds)# ---------------- 视图只返回 Response不碰 socket ----------------def view_index(params, query, form):with connect() as conn:rows conn.execute(SELECT id, title, created_at FROM posts ORDER BY id DESC).fetchall()if rows:items .join(lia href/post/{id}{title}/a - {ts}/li.format(idrow[id],titlehtml.escape(row[title]),tshtml.escape(row[created_at]))for row in rows)else:items li还没有文章/licontent ul{}/ulpa href/new写一篇/a/p.format(items)return Response(200, text/html; charsetutf-8,PAGE.substitute(title我的博客, contentcontent))def view_post(params, query, form):with connect() as conn:row conn.execute(SELECT title, body, created_at FROM posts WHERE id ?,(params[post_id],)).fetchone()if row is None:return Response(404, text/plain; charsetutf-8, 文章不存在)body html.escape(row[body]).replace(\n, br)content p{}/pp{}/ppa href/返回首页/a/p.format(html.escape(row[title]), body)# title 里也是用户内容同样必须转义return Response(200, text/html; charsetutf-8,PAGE.substitute(titlehtml.escape(row[title]),contentcontent))def view_new(params, query, form):if not form:return Response(200, text/html; charsetutf-8,PAGE.substitute(title写文章, contentNEW_FORM))title form.get(title, ).strip()body form.get(body, ).strip()if not title or not body:hint p标题和正文都不能为空。/preturn Response(400, text/html; charsetutf-8,PAGE.substitute(title写文章, contentNEW_FORM hint))with connect() as conn:conn.execute(INSERT INTO posts (title, body, created_at) VALUES (?, ?, ?),(title, body, now_iso()))# 303 让浏览器改用 GET 重新请求避免刷新时重复提交return Response(303, text/plain; charsetutf-8, 已创建,{Location: /})# ---------------- 路由 ----------------class Router:def __init__(self):self.routes []def add(self, method, pattern, func):self.routes.append((method, re.compile(pattern), func))def match(self, method, path):for allowed, regex, func in self.routes:if allowed ! method:continuefound regex.fullmatch(path)if found:return func, found.groupdict()return None, Nonerouter Router()router.add(GET, r/, view_index)router.add(GET, r/post/(?Ppost_id\d), view_post)router.add(GET, r/new, view_new)router.add(POST, r/new, view_new)# ---------------- 服务器适配 ----------------class BlogHandler(BaseHTTPRequestHandler):protocol_version HTTP/1.1def do_GET(self):self._dispatch(GET)def do_POST(self):self._dispatch(POST)def _read_form(self):length int(self.headers.get(Content-Length) or 0)if length 0:return {}raw self.rfile.read(length).decode(utf-8, replace)return {k: v[0] for k, v in parse_qs(raw, keep_blank_valuesTrue).items()}def _dispatch(self, method):parsed urlparse(self.path)func, params router.match(method, parsed.path)if func is None:self._send(Response(404, text/plain; charsetutf-8, 404 Not Found))returnquery parse_qs(parsed.query, keep_blank_valuesTrue)form self._read_form() if method POST else {}try:response func(params, query, form)except Exception as exc: # 别把堆栈吐给浏览器print(处理 {} 出错: {!r}.format(self.path, exc))response Response(500, text/plain; charsetutf-8, 服务器内部错误)self._send(response)def _send(self, response):payload response.body.encode(utf-8)self.send_response(response.status)self.send_header(Content-Type, response.content_type)self.send_header(Content-Length, str(len(payload)))for key, value in response.headers.items():self.send_header(key, value)self.end_headers()if self.command ! HEAD:self.wfile.write(payload)def log_message(self, fmt, *args):print([{}] {}.format(self.address_string(), fmt % args))def main():init_db()# 只绑定本机回环地址不要随手写 0.0.0.0server ThreadingHTTPServer((127.0.0.1, 8000), BlogHandler)print(已启动请访问 http://127.0.0.1:8000/)try:server.serve_forever()except KeyboardInterrupt:server.shutdown()if __name__ __main__:main()常见坑点1. 把用户输入直接拼进 HTML❌h2{}/h2.format(row[title])✅h2{}/h2.format(html.escape(row[title]))标题里写一段脚本就会在所有访客的浏览器里执行这就是 XSS。转义是输出侧的底线。2. 多线程共享一个sqlite3连接❌ 在模块顶层建一个全局conn所有请求复用 ✅ 每个请求在自己的视图里connect()用完即关with语句结束就提交sqlite3的连接不能跨线程使用ThreadingHTTPServer下会抛ProgrammingError。3. HTTP/1.1 不写Content-Length❌protocol_version HTTP/1.1但响应里没有Content-Length✅ 每次响应都先算好字节数再send_header(Content-Length, ...)HTTP/1.1 靠Content-Length判断响应结束缺失会让浏览器一直等页面卡住不渲染。4. 直接int(Content-Length)❌length int(self.headers[Content-Length])✅int(self.headers.get(Content-Length) or 0)GET 请求和小体积 POST 都可能没有这个头取不到时int(None)会抛TypeError。5. POST 成功后直接返回页面内容❌ 处理完创建逻辑就return一个 200 的 HTML ✅ 返回 303 并带上Location让浏览器重新用 GET 拉首页 否则用户按 F5 会重复提交表单同一篇文章被写进去好几次。6. 服务器绑定0.0.0.0❌ThreadingHTTPServer((0.0.0.0, 8000), Handler)✅ 开发时写127.0.0.1只监听本机0.0.0.0会把服务暴露给同网段所有人而这个服务器没有任何鉴权。7. 把异常堆栈写回浏览器❌except Exception as exc: self.wfile.write(str(exc).encode())✅ 前端只回一句通用错误详情打到服务端日志 堆栈里可能包含文件路径、数据库结构甚至配置信息。8. 用str.format渲染含用户内容的模板❌TEMPLATE.format(contentuser_text)用户输入里有{}就崩 ✅ 用string.Template的$name占位符或先转义str.format会把替换值之外的整段文本都当模板解析。总结问题错误做法正确做法输出用户内容直接拼 HTML先html.escape数据库连接全局共享每请求一个连接响应长度不写Content-Length按字节数显式声明表单提交后返回 200 页面返回 303 重定向监听地址0.0.0.0127.0.0.1错误处理堆栈回显日志记录前端通用提示博客程序的骨架就四样路由表、视图函数、模板、数据访问。把这四层分开之后你会发现成熟框架里的route、view、template、model无非是同一套分工的工业化版本。理解了这层对应关系再去学任何 Web 框架都会快得多。