手写Python Web框架：从零实现路由、请求与模板渲染

Caveman这个词，字面意思就是穴居人。我第一次看到有人拿它当项目名，是在一个技术社群里——作者说他的Web框架只依赖Python标准库，连路由都是用正则硬撸的，活得像个原始人，所以起了这个名字。我后来自己也照着这个思路写了一个同名框架，把Web服务的核心链路完整走了一遍：不依赖任何第三方框架，用最笨的办法实现路由、请求解析、响应构造、模板渲染，然后把每一行代码都摊开来看明白。这篇文章就是这次折腾的完整记录，包括设计思路、核心机制、完整实操和一路踩过的坑。

如果你现在看到Flask和Django这类框架心里总有个疑问——“它到底是怎么把URL映射到函数的？请求数据又是怎么到我手上的？”——那这篇文章非常适合你。它不需要你有多深的基础，只要会Python基础语法、能看懂正则表达式，就能照着代码把一个小型Web框架从零写出来。我还会用它搭一个真实的“书签管理API”，让你看到这套极简框架在真实场景里能干什么、不能干什么。全程代码我贴了完整版本，你可以直接复制运行。

1. 这个项目的定位：给自己写的“毛坯房”框架

1.1 为什么叫Caveman，它到底想解决什么问题

Frameworks这些年的趋势是越做越重。Django自带ORM、Admin后台、迁移工具、认证体系，Flask虽然轻，但加上各种扩展插件之后，项目结构也能膨胀到让人看不懂。框架本身不是坏东西，它解决的是工程效率问题。但问题在于，如果你只学过Flask、Django，你永远不知道“路由”这个东西的本质是什么。

Caveman这个项目想做的是反过来——用最原始的写法，把Web框架的核心能力手工实现一遍。它不追求生产级健壮性，不追求生态，就追求一件事：让你肉眼看清一个HTTP请求从进入进程到返回响应，中间到底发生了什么。

我给自己定了几条规则：

• 不依赖任何第三方库，只用Python标准库里的http.server、re、sqlite3。
• 路由用正则表达式实现，不用任何现成的路由库。
• 模板渲染用一个最简单的字符串替换引擎，不引入Jinja2。
• 整体代码控制在500行以内，每一行都要能讲清楚为什么存在。

这等于自己给自己造了一个“毛坯房”。没有精装修，水管电线全裸露在外面，但对学习的人来说，这正是价值所在。后来我做的一个书签管理小工具就是跑在它上面的，几百行代码加上一个SQLite数据库文件，部署在一台小机器上，稳得很。

1.2 它适合谁，不适合谁

先说不适合的。如果你要做的是一个商业项目、需要高并发、要求各种安全加固、需要团队协作——别用这种自己写的框架，一率上Django或者FastAPI。这不是技术情怀问题，是工程效率与安全边界问题。自己写的东西再熟，也扛不住团队人员流动带来的维护成本。

适合的是这几类人：

• 想弄懂Web框架原理的初中级开发者。我在带新人时发现一个规律：凡是能徒手说出“URL路由”背后是“正则匹配加函数调用”的人，后面用任何框架都顺手；凡是只会@app.route的人，遇到诡异问题就抓瞎。
• 需要极轻量HTTP服务的场景。比如内网小工具、个人脚本、原型验证，一个几百K的脚本能搞定的事，没必要拖一个几十MB的框架。
• 对代码有“掌控欲”的开发者。有人天生喜欢把系统拆穿看清楚，这类人写Caveman会非常愉快。

我自己属于第三类加第二类的混合体。在这个项目之前，我对HTTP协议和WSGI的理解是书本化的，这次亲手实现之后，很多以前死记硬背的概念——比如状态码的语义、Content-Length为什么重要、GET和POST在报文层面的差异——才真正落地了。

2. 整体设计思路：从零到一的核心方案选型

2.1 单一入口：用标准库的HTTP服务器做地基

Caveman的架构非常简单，本质上就是一个“路由器加分发器”。底层我直接用Python标准库的http.server.BaseHTTPRequestHandler作为HTTP协议的接收入口。它负责监听端口、解析HTTP报文、回调对应的处理方法。

选它有三个理由。第一，它是标准库自带的，不需要pip安装任何东西；第二，它的接口简单，只需要继承类并重写do_GET、do_POST这些方法；第三，它的报错信息和调试信息都很直白，方便初学阶段排查问题。

from http.server import HTTPServer, BaseHTTPRequestHandler import re class CavemanHandler(BaseHTTPRequestHandler): def do_GET(self): pass def do_POST(self): pass

你可能注意到，我用的是传统的HTTPServer加BaseHTTPRequestHandler，而不是更现代的ThreadingHTTPServer。前者是单线程的，后者是每个请求开一个线程。我在初版用了单线程，后来发现SQLite的写操作会阻塞整个服务，就换成了ThreadingHTTPServer。这个改动只是换了个类名，代价极小，收益明显。对学习而言，先用单线程理解流程，再切换线程模型，体验会很自然。

2.2 路由表的本质：一个存正则的列表

这是整个项目最核心的设计决策。Web框架的路由表，说白了就是一个“URL模式”到“函数”的映射关系。Flask这类框架会把@app.route('/user/<int:id>')这种语法解析成内部的路由规则，Caveman不搞这些花活，直接用正则表达式字符串表示URL模式。

路由表我用一个列表保存，每个元素是一个元组，包含三部分：HTTP方法、正则表达式、处理函数。

routes = [] def route(pattern, methods=['GET']): def decorator(func): routes.append((methods, re.compile(pattern), func)) return func return decorator

这个设计把问题降维了。/user/123在Caveman里就是r'/user/(\d+)'，匹配到的数字作为参数传给处理函数。正则表达式的捕获组直接变成视图函数的参数，这个映射关系非常直观。

这里有一个重要的选型点：为什么不用字典来做URL匹配？因为字典只能做精确匹配，URL里带可变部分（比如用户ID）就无法处理了。正则表达式则可以一套规则覆盖多个实际路径。代价是每次请求都要遍历路由表做正则匹配，性能上不如编译后的分发器，但对于学习和内网小工具来说，这点性能损耗完全可以忽略。

2.3 请求与响应：手工解析HTTP报文

Web框架对开发者屏蔽了两个关键动作：把HTTP报文解析成对象，以及把响应对象序列化成HTTP报文。Caveman必须自己干这两件事。

请求解析方面，BaseHTTPRequestHandler已经帮我们拿到了原始的请求行、请求头和请求体。我在此基础上做了一层薄封装，把查询参数解析成字典，把请求体按Content-Type分别处理成表单数据或JSON数据。

import json from urllib.parse import urlparse, parse_qs class Request: def __init__(self, handler): self.method = handler.command self.path = handler.path parsed = urlparse(handler.path) self.query = parse_qs(parsed.query) self.headers = handler.headers body = handler.rfile.read(int(handler.headers.get('Content-Length', 0))) self.body = body self.json = json.loads(body) if body else {} self.form = parse_qs(body.decode()) if body else {}

这里有个细节要注意：如果请求头没有Content-Length，说明没有请求体。读取body时必须用这个字段指定的字节数去读，否则会一直读阻塞。这个坑我在初版踩过一次，后面在“常见问题”部分会详细说。

响应构造方面，我封装了一个Response类，支持状态码、响应头、响应体三要素。核心就是构造出符合HTTP协议的文本。重点是三个东西：状态行、Content-Type、Content-Length。如果没有Content-Length，HTTP/1.1的连接无法判断响应到哪里结束，客户端就会傻等。

class Response: def __init__(self, body='', status=200, content_type='text/html; charset=utf-8'): self.body = body.encode() self.status = status self.content_type = content_type def render(self): lines = [] lines.append(f"HTTP/1.1 {self.status} {STATUS_TEXT.get(self.status, '')}") lines.append(f"Content-Type: {self.content_type}") lines.append(f"Content-Length: {len(self.body)}") lines.append("Connection: close") lines.append("") lines.append("") return "\r\n".join(lines).encode() + self.body

这个render方法把响应对象变成字节流。\r\n是HTTP协议规定的行分隔符，不能写成\n，这是一个非常容易忽略但很重要的细节。

3. 核心机制拆解：路由、模板、静态文件怎么实现

3.1 请求分发：从URL到函数的完整链路

有了路由表，分发器的工作就很简单了。每个请求进来，根据HTTP方法过滤出候选路由，然后依次用正则对路径做匹配。第一个命中的路由就把捕获组参数传给函数，调用它，拿到响应对象，再返回给客户端。

def handle(self, request): for methods, pattern, func in routes: if request.method not in methods: continue m = pattern.match(request.path) if m: args = m.groups() response = func(request, *args) return response return Response('Not Found', status=404)

这里我用了pattern.match而不是pattern.search，这是一个刻意的选择。match要求从路径的开头匹配，天然防止了类似/api/../admin这种绕过前缀的情况。如果用search，任何位置包含匹配内容都会被命中，安全边界就没了。

分发时的参数传递我选择了最原始的位置参数方式。路由正则里有多少个捕获组，函数就接收多少个位置参数。这样写最直观，也好调试。缺点是如果正则写复杂了，捕获组一多，参数顺序容易乱。我后来在代码里加了一条注释：捕获组顺序必须与函数参数顺序一致，这是唯一契约。

3.2 模板渲染：一个200行的极简引擎

项目初期我用字符串拼接生成HTML，比如f"<li>{item}</li>"一把梭。写了几次之后发现，页面结构一复杂，字符串拼接就是个灾难：引号嵌套、换行转义、条件分支，全混在一起，可读性极差。

于是我实现了一个简陋的模板引擎，核心思路是“占位符替换加循环块展开”。模板文件里用{{ var }}做变量占位，用{% for x in list %}和{% endfor %}做循环。渲染过程分两步：先用正则把循环块拆出来处理，再替换变量。

def render_template(template, **context): html = template # 处理循环块 pattern = r'{% for (\w+) in (\w+) %}(.*?){% endfor %}' def loop_repl(m): iterable = context.get(m.group(2), []) body = m.group(3) parts = [] for item in iterable: item_html = body.replace('{{ ' + m.group(1) + ' }}', str(item)) parts.append(item_html) return ''.join(parts) html = re.sub(pattern, loop_repl, html, flags=re.S) # 处理变量 html = re.sub(r'{{ (\w+) }}', lambda m: str(context.get(m.group(1), '')), html) return html

这套引擎只支持变量和循环，不支持继承、过滤器、条件判断。够用吗？对于书签管理工具这种页面，够用了。模板总共四五个，每个十几行，没有复杂嵌套，维护起来毫无压力。

我特意加了个flags=re.S参数，这又是一个实战教训。re.S让点号能匹配换行符，否则循环块内容跨行时正则根本匹配不上。初学者写这段代码最容易漏掉这个参数。

3.3 静态文件：让框架像个体面的“成年程序”

动态页面搞定后，我发现一个很尴尬的问题：CSS和JavaScript文件没法加载。因为所有请求都进了路由分发，浏览器请求/static/style.css时，路由表里没有这个模式，直接返回404。

解决方案很直接：写一个静态文件中间件。判断路径是否以/static/开头，如果是，就从指定目录读取文件，根据扩展名设置Content-Type，返回文件内容。

STATIC_DIR = 'static' STATIC_TYPES = { '.css': 'text/css', '.js': 'application/javascript', '.png': 'image/png', '.jpg': 'image/jpeg', } def serve_static(request): path = request.path if not path.startswith('/static/'): return None filename = path[len('/static/'):] full_path = os.path.join(STATIC_DIR, filename) # 防目录穿越 if not os.path.abspath(full_path).startswith(os.path.abspath(STATIC_DIR)): return Response('Forbidden', status=403) if not os.path.exists(full_path): return Response('Not Found', status=404) ext = os.path.splitext(full_path)[1] content_type = STATIC_TYPES.get(ext, 'application/octet-stream') with open(full_path, 'rb') as f: return Response(f.read(), content_type=content_type)

目录穿越是这类代码必须防的问题。用户如果请求/static/../../etc/passwd，os.path.join会拼出路径，如果没有abspath检查和前缀校验，就能读到系统文件。这段代码用的是白名单式的防护：先把路径转成绝对路径，再检查它是否在静态目录的绝对路径范围内。这个方法简单可靠。

4. 实操全记录：用Caveman搭一个书签管理服务

4.1 环境准备与项目结构

在动手写代码前，先把环境说清楚。我用的是Python 3.8+，不需要任何虚拟环境以外的依赖。项目结构非常清爽：

caveman/ ├── caveman.py # 框架核心 ├── server.py # 应用入口，定义路由 ├── templates/ │ ├── index.html # 书签列表页 │ └── add.html # 添加书签表单页 ├── static/ │ └── style.css # 极简样式 └── data.db # SQLite数据库文件

整个项目加起来，不含数据库内容，代码量在600行左右。这个体量让我可以随时把整个项目的结构在脑子里装下，改任何一处都不用翻文件。

4.2 框架核心代码：把前面讲的思路落成文件

我把前面分散的设计串起来，整合成一个完整的caveman.py。这里贴的是简化版，保留了全部核心功能，去掉了一些日志和异常分支，方便你阅读。

import json import os import re import sqlite3 from http.server import HTTPServer, BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import urlparse, parse_qs STATUS_TEXT = { 200: 'OK', 201: 'Created', 404: 'Not Found', 403: 'Forbidden', 405: 'Method Not Allowed', 500: 'Internal Server Error', } class Request: def __init__(self, handler): self.method = handler.command self.path = handler.path parsed = urlparse(handler.path) self.path_only = parsed.path self.query = parse_qs(parsed.query) self.headers = handler.headers length = int(handler.headers.get('Content-Length', 0) or 0) body = handler.rfile.read(length) if length else b'' self.body = body self.json = json.loads(body) if body else None self.form = parse_qs(body.decode()) if body else {} class Response: def __init__(self, body='', status=200, content_type='text/html; charset=utf-8'): self.body = body.encode() if isinstance(body, str) else body self.status = status self.content_type = content_type def render(self): head = f"HTTP/1.1 {self.status} {STATUS_TEXT.get(self.status, '')}\r\n" head += f"Content-Type: {self.content_type}\r\n" head += f"Content-Length: {len(self.body)}\r\n" head += "Connection: close\r\n\r\n" return head.encode() + self.body routes = [] def route(pattern, methods=['GET']): def decorator(func): routes.append((methods, re.compile(pattern), func)) return func return decorator def render_template(template, **context): html = template loop_pat = re.compile(r'{% for (\w+) in (\w+) %}(.*?){% endfor %}', re.S) def loop_repl(m): iterable = context.get(m.group(2), []) body = m.group(3) return ''.join( body.replace('{{ ' + m.group(1) + ' }}', str(item)) for item in iterable ) html = loop_pat.sub(loop_repl, html) return re.sub(r'{{ (\w+) }}', lambda m: str(context.get(m.group(1), '')), html) class Handler(BaseHTTPRequestHandler): def do_GET(self): self._dispatch() def do_POST(self): self._dispatch() def _dispatch(self): try: req = Request(self) if req.path_only.startswith('/static/'): response = self._serve_static(req) else: response = self._route(req) self.wfile.write(response.render()) except Exception as e: self.wfile.write(Response(f'Server Error: {e}', status=500).render()) def _route(self, request): for methods, pattern, func in routes: if request.method not in methods: continue m = pattern.match(request.path_only) if m: return func(request, *m.groups()) return Response('Not Found', status=404) def _serve_static(self, request): filename = request.path_only[len('/static/'):] full_path = os.path.abspath(os.path.join('static', filename)) static_abs = os.path.abspath('static') if not full_path.startswith(static_abs): return Response('Forbidden', status=403) if not os.path.isfile(full_path): return Response('Not Found', status=404) ext = os.path.splitext(full_path)[1] ctype = {'css': 'text/css', 'js': 'application/javascript'}.get(ext, 'application/octet-stream') with open(full_path, 'rb') as f: return Response(f.read(), content_type=ctype) def log_message(self, format, *args): pass # 关闭默认日志，避免刷屏 def run(host='0.0.0.0', port=8000): server = ThreadingHTTPServer((host, port), Handler) print(f'Caveman running at http://{host}:{port}') server.serve_forever()

这段代码是整个框架的骨干。我特意把异常捕获放在最外层，这样任何视图函数报错都不会让整个服务崩溃，而是返回一个500响应。这在开发期可能掩盖错误——所以我保留了错误信息的输出，直接拼在响应体里方便查看。生产环境再用一个开关关掉。

4.3 业务代码：书签管理API和页面

框架写好后，业务代码就是纯粹的使用了。我用它做了一个书签管理服务，支持两个功能：查看书签列表、新增书签。数据库用的是SQLite，因为它自带、零配置、单文件，和Caveman的理念完全契合。

from caveman import route, run, Response, Request, render_template import sqlite3, json def db(): conn = sqlite3.connect('data.db') conn.execute('CREATE TABLE IF NOT EXISTS bookmarks (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, url TEXT)') return conn @route(r'/', methods=['GET']) def index(request): conn = db() rows = conn.execute('SELECT id, title, url FROM bookmarks ORDER BY id DESC').fetchall() conn.close() html = render_template(open('templates/index.html').read(), bookmarks=rows) return Response(html) @route(r'/api/bookmarks', methods=['GET']) def api_list(request): conn = db() rows = conn.execute('SELECT id, title, url FROM bookmarks').fetchall() conn.close() data = [{'id': r[0], 'title': r[1], 'url': r[2]} for r in rows] return Response(json.dumps(data), content_type='application/json') @route(r'/api/bookmarks', methods=['POST']) def api_add(request): if not request.json: return Response('Bad Request', status=400) title = request.json.get('title', '') url = request.json.get('url', '') if not title or not url: return Response('Missing Fields', status=400) conn = db() conn.execute('INSERT INTO bookmarks (title, url) VALUES (?, ?)', (title, url)) conn.commit() conn.close() return Response('Created', status=201)

看这段代码，路由的用法和主流框架很像，但每个环节都透明：正则模式自己写，捕获组自己对应参数，JSON解析自己调，SQL自己写。没有任何“魔法”。这也是我整个项目的核心理念——把所有的隐式行为变成显式代码。

GET /返回的是HTML页面，GET /api/bookmarks返回JSON数据，同一份数据支撑两种输出。这种设计在实际项目中很常见——页面给浏览器，API给脚本或前端框架。

4.4 运行测试：从命令行到实际请求

运行服务就一条命令：

python server.py

启动后，我分别用浏览器和curl验证各个接口。先看首页，再添加两条书签：

curl -X POST http://localhost:8000/api/bookmarks \ -H "Content-Type: application/json" \ -d '{"title": "Python docs", "url": "https://docs.python.org"}' curl -X POST http://localhost:8000/api/bookmarks \ -H "Content-Type: application/json" \ -d '{"title": "SQLite home", "url": "https://sqlite.org"}' curl http://localhost:8000/api/bookmarks

返回结果符合预期：

[{"id": 1, "title": "Python docs", "url": "https://docs.python.org"}, {"id": 2, "title": "SQLite home", "url": "https://sqlite.org"}]

这说明框架的POST请求体解析、JSON序列化、正则路由、SQLite写入这几个核心环节全部打通了。整个服务从启动到数据落地，没有任何框架层的神秘阻碍，每一条数据流我都知道它经过了哪些代码。

我还在生产部署时加了一个小功能：给add.html写了一个表单页，提交后跳回首页。这样手机浏览器也能添加书签，不需要每次都敲curl。这个页面用了最朴素的HTML表单，POST数据以application/x-www-form-urlencoded格式传输，正好验证了request.form的解析逻辑。

5. 实战中遇到的坑与排查经验

5.1 正文读取卡死：Content-Length是关键

第一次实现POST接口时，我一直遇到同一问题：请求发过去后服务端很久不返回，最后超时。排查后发现是self.rfile.read()没有指定长度。

HTTP协议里，请求体的边界由Content-Length或Transfer-Encoding: chunked标识。对于普通表单和JSON请求，Content-Length一定存在。如果读取时没有传长度，Python的read()会一直读到连接关闭才返回，而客户端发送完请求体后不会立刻断开连接，于是服务端就永久挂起。

解决方式就是严格按int(handler.headers.get('Content-Length', 0))读取。这里有个隐藏细节：如果Content-Length: 0，说明没有请求体，直接给空字节即可。我当时还加了一个保护，把headers.get的结果转成int时用or 0，防止拿到None时报错。

5.2 路由顺序的坑：匹配命中的是“第一个”

Caveman的路由匹配是按路由表顺序遍历的，这带来一个实际的坑：如果两个正则都匹配同一个路径，永远返回先注册的那个。

例如，我注册了r'/api/bookmarks/(\d+)'用于获取单条数据，又注册了r'/api/bookmarks/new'用于展示新书签表单。请求/api/bookmarks/new时，第一个模式要求数字结尾，匹配不上，所以没问题。但如果我把两个模式顺序反过来，/api/bookmarks/new会先匹配到r'/api/bookmarks/(\d+)'，捕获组得到的是new，后面视图函数里做类型转换就直接抛异常了。

规则很简单：具体的模式注册在前面，宽泛的模式注册在后面。这和路由器的最长前缀匹配、防火墙规则顺序是同一个道理。我在代码注释里写了这个约定，避免未来维护时踩坑。

5.3 中文乱码问题：编码要一竿子插到底

写书签服务时，中文标题在浏览器里显示成了乱码。排查路径涉及三个环节：响应头、HTML文件编码、数据库存储。

响应头我已经设置了Content-Type: text/html; charset=utf-8，HTML文件本身是UTF-8编码，SQLite默认也存UTF-8。问题出在Response类的初始化——body.encode()默认用UTF-8没问题，但如果传入的body是bytes类型，再调用encode()就会报错或双编码。

我最后的处理是：在Response.__init__里判断类型，只有字符串才编码，bytes直接透传。这个改动很小，但排查过程让我养成了一个习惯：凡是涉及文本的地方，先确认编码链条是否连通——请求进来解码用什么、处理过程用什么、响应出去编码用什么，三段必须一致。

5.4 并发写数据库：SQLite锁冲突

上线后接连收到几次“database is locked”的报错。场景是手机端提交书签时，服务端正好同时在处理另一个请求的写操作。

排查结果是ThreadingHTTPServer带来的并发写冲突。SQLite对同一时刻的写操作是全局唯一的。解决方式有几个方案，我选了最朴素的一种：所有数据库写操作都包在一个全局锁里。因为这是一个个人工具，写频率极低，全局锁完全够用，根本没有必要上PostgreSQL或者设计队列。

import threading db_lock = threading.Lock() @route(r'/api/bookmarks', methods=['POST']) def api_add(request): with db_lock: conn = db() conn.execute('INSERT INTO bookmarks (title, url) VALUES (?, ?)', (title, url)) conn.commit() conn.close() return Response('Created', status=201)

顺带一提，读操作我没加锁。SQLite的读并发是允许的，只要不是同时写。这种区分对待的方式，在并发压力小的场景下能少写很多无谓的锁。

5.5 常见问题速查表

6. 从这次折腾里学到的东西

最后一个部分，聊聊项目之外的收获。

写Caveman之前，我对Web框架的使用停留在“背API”阶段。@app.route怎么写、request.args怎么拿、响应怎么返回，这些我都熟，但从来没想过框架底层在做什么。写完这套东西之后，再看Flask的源码，很多以前似懂非懂的设计突然就通了——Flask的endpoint和view_func的映射、before_request钩子的执行时机、Response对象的make_response流程，本质都是在处理HTTP协议层面的套路。

如果你也想走一遍这条路，我的建议是分三步走：第一步，先用标准库写一个能返回“Hello World”的HTTP服务，理解请求-响应循环。第二步，实现正则路由和参数传递，把装饰器语法用起来。第三步，加模板引擎和数据库，做成一个能实际用的工具。每一步的代码量都不大，但每一步都能让你对Web开发的认知深一层。

这个书签服务我到现在还在用，部署在家里的小服务器上，随手记录零散的链接。其实真要换成Flask，代码量也差不了多少。但用自己写的框架跑着真实业务，感觉完全不同——每次打开页面，我都知道从浏览器输入URL到屏幕渲染出列表，这中间经过的每一行代码都是我亲手写的。这种掌控感，是框架文档里永远找不到的。

最后分享一个我自己用着顺手的小技巧：把Caveman的框架代码文件放在自己项目的site-packages路径之外，就放项目根目录。这样你改框架代码后，业务代码马上就能看到效果，不需要重新安装。我在调试模板引擎时，就是开着两个终端，一边改caveman.py，一边刷浏览器页面，实时验证改动的效果。这种“所见即所得”的开发体验，反而是大型框架里少有的快乐。