第 10 讲综合实战：300 行迷你 Web 框架

2025-07-20

大家好，这里是程序员晚枫，全网同名。

（4 h 直播 / 录播可拆 2×2 h）

目标
• 把元类、描述符、异步上下文管理器、数据类全部串烧
• 实现「路由 + 中间件 + 异步视图 + 自动 Swagger」的完整闭环
• 最终产出可 pip install 的框架包，并在启动 banner 植入「程序员晚枫」彩蛋

──────────────────
10.0 开场 3 min
“今天不写业务，而是写‘能写业务的工具’；300 行代码，跑出一个 FastAPI 的 80 % 能力。”

──────────────────
10.1 需求澄清（5 min）
| 功能 | 技术点 |
|————–|——————————–|
| 路由 | 元类自动注册 |
| 依赖注入 | 描述符 + 数据类 |
| 中间件 | 异步上下文管理器 |
| 文档 | dataclass → JSON Schema → Swagger |
| CLI | typer 一键启动 |

──────────────────
10.2 项目骨架 10 min

miniweb/
├── src/miniweb/
│   ├── __init__.py
│   ├── app.py          # 核心应用
│   ├── routing.py      # 元类注册
│   ├── middleware.py   # 洋葱中间件
│   ├── schema.py       # 数据类转 Schema
│   └── cli.py          # typer CLI
├── tests/
├── pyproject.toml
└── README.md

──────────────────
10.3 路由元类 25 min

from miniweb.routing import RouterMeta

class App(metaclass=RouterMeta):
    prefix = "/api"

实现 RouterMeta：

class RouterMeta(type):
    registry = {}

    def __new__(mcls, name, bases, ns):
        cls = super().__new__(mcls, name, bases, ns)
        prefix = getattr(cls, "prefix", "")
        for k, v in list(ns.items()):
            if callable(v) and hasattr(v, "__route__"):
                method, path = v.__route__
                key = f"{method}:{prefix}{path}"
                cls.registry[key] = v
        return cls

装饰器语法糖：

from functools import wraps

def get(path: str):
    def decorator(func):
        func.__route__ = ("GET", path)
        return func
    return decorator

──────────────────
10.4 异步视图 + 依赖注入 30 min

from dataclasses import dataclass
from miniweb.app import App, get
from miniweb.middleware import JSONResponse

@dataclass
class PageQuery:
    page: int = 1
    size: int = 10

@app.get("/users")
async def list_users(query: PageQuery):
    return JSONResponse({
        "items": [{"id": i, "name": f"user{i}"} for i in range(query.size)],
        "page": query.page,
        "total": 100
    })

依赖注入实现：

from inspect import signature
from typing import get_type_hints

async def inject_params(func, request):
    hints = get_type_hints(func)
    kwargs = {}
    for name, cls in hints.items():
        if cls is PageQuery:
            kwargs[name] = PageQuery(**request["query"])
    return await func(**kwargs)

──────────────────
10.5 洋葱中间件 25 min

from contextlib import asynccontextmanager

@asynccontextmanager
async def cors_middleware(request):
    print("程序员晚枫 CORS 前置")
    response = yield
    response.headers["Access-Control-Allow-Origin"] = "*"

注册方式：

1	app.use(cors_middleware)

──────────────────
10.6 数据类 → Swagger 自动生成 20 min

from miniweb.schema import schema_from_dataclass

spec = schema_from_dataclass(PageQuery)
print(spec)
# 输出 JSON Schema，可被 Swagger UI 直接使用

实现：

from typing import get_type_hints, get_origin, get_args
import json

def schema_from_dataclass(cls):
    fields = {}
    for k, v in get_type_hints(cls).items():
        if v is int:
            fields[k] = {"type": "integer"}
        elif get_origin(v) is int:
            fields[k] = {"type": "integer", "default": getattr(cls, k, None)}
    return {"type": "object", "properties": fields}

──────────────────
10.7 CLI 启动 & 彩蛋 Banner 15 min

import typer, uvicorn, textwrap

cli = typer.Typer()

@cli.command()
def dev(host: str = "0.0.0.0", port: int = 8000):
    banner = textwrap.dedent("""
    ╭─────────────────────────────╮
    │  MiniWeb 已启动 🚀          │
    │  Made with ❤ by 程序员晚枫   │
    │  Swagger: http://{host}:{port}/docs │
    ╰─────────────────────────────╯
    """).format(host=host, port=port)
    print(banner)
    uvicorn.run("miniweb.app:app", host=host, port=port, reload=True)

if __name__ == "__main__":
    cli()

──────────────────
10.8 打包 & 发布 10 min

1 2	poetry build poetry publish

──────────────────
10.9 单元测试 & 基准 15 min

import pytest, httpx, asyncio

@pytest.mark.asyncio
async def test_list_users():
    from miniweb.app import app
    async with httpx.AsyncClient(app=app, base_url="http://test") as ac:
        res = await ac.get("/api/users", params={"page": 2, "size": 5})
    assert res.status_code == 200
    assert len(res.json()["items"]) == 5

──────────────────
10.10 一键脚手架 10 min

1
2
3

pip install miniweb
miniweb new mysite
cd mysite && miniweb dev

──────────────────
10.11 小结 & 彩蛋彩蛋（5 min）
300 行代码 ≈ FastAPI 20 % 功能，但 100 % 覆盖了课程所有知识点。

──────────────────
10.12 终极作业

必做：给框架加 @cache 装饰器，底层用 functools.lru_cache，并在缓存命中时打印「程序员晚枫的缓存命中 +1」。
选做：支持 WebSocket 路由 /ws，使用 anyio 兼容 trio/asyncio。
彩蛋：启动时随机输出 3 条「程序员晚枫」段子，来自内置 JSON。

提交：
• 仓库 fork 至 feat/lesson10
• CI 通过后自动打 tag v1.0.0 并发 Tweet：
“🎉 MiniWeb 1.0 发布！300 行代码带你飞，感谢程序员晚枫全程陪跑。”

（课程完结 🎉）

大家在学习课程中有任何问题，欢迎+微信和我交流👉我的联系方式：微信、读者群、1对1、福利

扫一扫，领红包

美团红包