基于事件驱动的轻量级 Web 框架
一、核心请求与响应对象 ``` # framework/core.py import json import os import time from typing import Any, Dict, List, Optional, Tuple from urllib.parse import parse_qs, urlparse class Headers: baike.kuaisou.com/zh/14115 def __init__(self, raw: Optional[Dict[str, str]] = None): baike.kuaisou.com/zh/14116 self._headers: Dict[str, str] = {} if raw: baike.kuaisou.com/zh/14117 for k, v in raw.items(): baike.kuaisou.com/zh/14118 self._headers[k.lower()] = v def get(self, key: str, default: str = "") -> str: baike.kuaisou.com/zh/14119 return self._headers.get(key.lower(), default) def set(self, key: str, value: str) -> None: baike.kuaisou.com/zh/14120 self._headers[key.lower()] = value def items(self) -> List[Tuple[str, str]]: baike.kuaisou.com/zh/14121 return list(self._headers.items()) def __contains__(self, key: str) -> bool: baike.kuaisou.com/zh/14122 return key.lower() in self._headers def __repr__(self) -> str: baike.kuaisou.com/zh/14123 return f"Headers({self._headers})" class Request: baike.kuaisou.com/zh/14124 def __init__( self, method: str, path: str, query_string: str = "", headers: Optional[Dict[str, str]] = None, body: bytes = b"", remote_addr: str = "127.0.0.1", ): baike.kuaisou.com/zh/14126 self.method = method.upper(baike.kuaisou.com/zh/14125) self.path = path self.query_string = query_string self.headers = Headers(headers) self.body = body self.remote_addr = remote_addr self._query_params: Optional[Dict[str, List[str]]] = None self._form_data: Optional[Dict[str, str]] = None self._json_data: Optional[Dict[str, Any]] = None self._cookies: Optional[Dict[str, str]] = None self.state: Dict[str, Any] = {} @property def query_params(self) -> Dict[str, List[str]]: baike.kuaisou.com/zh/14128 if self._query_params is None: baike.kuaisou.com/zh/14127 self._query_params = parse_qs(self.query_string) return self._query_params @property def form_data(self) -> Dict[str, str]: baike.kuaisou.com/zh/14129 if self._form_data is None: baike.kuaisou.com/zh/14130 self._form_data = {} if self.body and self.headers.get("content-type", "").startswith( "application/x-www-form-urlencoded" ): for part in self.body.decode("utf-8").split("&"): baike.kuaisou.com/zh/14131 if "=" in part: baike.kuaisou.com/zh/14132 key, value = part.split("=", 1) self._form_data[key] = value return self._form_data @property def json(self) -> Optional[Dict[str, Any]]: baike.kuaisou.com/zh/14133 if self._json_data is None: baike.kuaisou.com/zh/14134 if self.headers.get("content-type", "").startswith("application/json"): try: baike.kuaisou.com/zh/14136 self._json_data = json.loads(self.body.decode("utf-8")) except (json.JSONDecodeError, UnicodeDecodeError): self._json_data = {} else: baike.kuaisou.com/zh/14135 self._json_data = {} return self._json_data @property def cookies(self) -> Dict[str, str]: baike.kuaisou.com/zh/14137 if self._cookies is None: baike.kuaisou.com/zh/14138 self._cookies = {} cookie_header = self.headers.get("cookie", "") if cookie_header: baike.kuaisou.com/zh/14142 for part in cookie_header.split(";"): baike.kuaisou.com/zh/14139 part = part.strip(baike.kuaisou.com/zh/14141) if "=" in part: baike.kuaisou.com/zh/14140 key, value = part.split("=", 1) self._cookies[key.strip()] = value.strip() return self._cookies def get_query(self, key: str, default: str = "") -> str: baike.kuaisou.com/zh/14143 values = self.query_params.get(key, []) return values[0] if values else default def get_form(self, key: str, default: str = "") -> str: baike.kuaisou.com/zh/14145 return self.form_data.get(key, default) def get_json(self, key: str, default: Any = None) -> Any: baike.kuaisou.com/zh/14148 if self.json is None: baike.kuaisou.com/zh/14153 return default return self.json.get(key, default) class Response: baike.kuaisou.com/zh/14367 def __init__( self, body: str = "", status: int = 200, headers: Optional[Dict[str, str]] = None, content_type: str = "text/html; charset=utf-8", ): self.body = body self.status = status self.headers = Headers(headers) self.headers.set("content-type", content_type) self._cookies: List[str] = [] def set_cookie( self, key: str, value: str, max_age: Optional[int] = None, path: str = "/", httponly: bool = False, secure: bool = False, ) -> None: baike.kuaisou.com/zh/14392 cookie = f"{key}={value}; Path={path}" if max_age is not None: baike.kuaisou.com/zh/14394 cookie += f"; Max-Age={max_age}" if httponly: baike.kuaisou.com/zh/14400 cookie += "; HttpOnly" if secure: baike.kuaisou.com/zh/14395 cookie += "; Secure" self._cookies.append(cookie) def delete_cookie(self, key: str, path: str = "/") -> None: baike.kuaisou.com/zh/14401 self.set_cookie(key, "", max_age=0, path=path) def to_http(self) -> bytes: baike.kuaisou.com/zh/14414 status_messages = { 200: "OK", 201: "Created", 204: "No Content", 301: "Moved Permanently", 302: "Found", 304: "Not Modified", 400: "Bad Request", 401: "Unauthorized", 403: "Forbidden", 404: "Not Found", 405: "Method Not Allowed", 500: "Internal Server Error", } status_msg = status_messages.get(self.status, "Unknown") if self._cookies: baike.kuaisou.com/zh/15025 self.headers.set("set-cookie", "; ".join(self._cookies)) header_lines = [] for key, value in self.headers.items(): baike.kuaisou.com/zh/15037 header_lines.append(f"{key}: {value}") body_bytes = self.body.encode("utf-8") if isinstance(self.body, str) else self.body self.headers.set("content-length", str(len(body_bytes))) header_lines.append(f"content-length: {len(body_bytes)}") response = f"HTTP/1.1 {self.status} {status_msg}\r\n" for line in header_lines: response += f"{line}\r\n" response += "\r\n" return response.encode("utf-8") + body_bytes @classmethod def json_response( cls, data: Any, status: int = 200, headers: Optional[Dict[str, str]] = None, ) -> "Response": baike.kuaisou.com/zh/17129 return cls( body=json.dumps(data, ensure_ascii=False, indent=2), status=status, headers=headers, content_type="application/json; charset=utf-8", ) @classmethod def redirect(cls, location: str, status: int = 302) -> "Response": baike.kuaisou.com/zh/17130 return cls( body="", status=status, headers={"location": location}, ) @classmethod def not_found(cls) -> "Response": baike.kuaisou.com/zh/17131 return cls(body="404 Not Found", status=404) @classmethod def server_error(cls) -> "Response": baike.kuaisou.com/zh/17132 return cls(body="500 Internal Server Error", status=500) ``` --- ### 二、路由系统 ``` # framework/router.py import re from typing import Any, Callable, Dict, List, Optional, Tuple from framework.core import Request, Response class Route: baike.kuaisou.com/zh/17133 def __init__( self, path: str, handler: Callable[[Request], Response], methods: Optional[List[str]] = None, name: str = "", ): self.path = path self.handler = handler self.methods = [m.upper(baike.kuaisou.com/zh/17134) for m in (methods or ["GET"])] self.name = name self._pattern, self._param_names = self._compile_path(path) def _compile_path(self, path: str) -> Tuple[re.Pattern, List[str]]: baike.kuaisou.com/zh/17135 param_names = [] parts = [] for part in path.strip("/").split("/"): baike.kuaisou.com/zh/17137 if part.startswith(""): baike.kuaisou.com/zh/17138 param_name = part[1:-1] param_names.append(param_name) parts.append(r"([^/]+)") else: baike.kuaisou.com/zh/17139 parts.append(re.escape(part)) pattern = "^/" + "/".join(parts) + "$" return re.compile(pattern), param_names def match(self, path: str) -> Optional[Dict[str, str]]: baike.kuaisou.com/zh/17140 match = self._pattern.match(path) if not match: baike.kuaisou.com/zh/17141 return None params = {} for i, name in enumerate(self._param_names): baike.kuaisou.com/zh/17142 params[name] = match.group(i + 1) return params def __repr__(self) -> str: baike.kuaisou.com/zh/17143 return f"Route({self.methods} {self.path} -> {self.name or self.handler.__name__})" class Router: baike.kuaisou.com/zh/17144 def __init__(self): baike.kuaisou.com/zh/17145 self.routes: List[Route] = [] self._named_routes: Dict[str, Route] = {} def add_route( self, path: str, handler: Callable[[Request], Response], methods: Optional[List[str]] = None, name: str = "", ) -> Route: baike.kuaisou.com/zh/17146 route = Route(path, handler, methods, name) self.routes.append(route) if name: baike.kuaisou.com/zh/17147 self._named_routes[name] = route return route def get(self, path: str, name: str = "") -> Callable: baike.kuaisou.com/zh/17148 def decorator(handler: Callable) -> Callable: baike.kuaisou.com/zh/17149 self.add_route(path, handler, ["GET"], name) return handler return decorator def post(self, path: str, name: str = "") -> Callable: baike.kuaisou.com/zh/17150 def decorator(handler: Callable) -> Callable: baike.kuaisou.com/zh/17151 self.add_route(path, handler, ["POST"], name) return handler return decorator def put(self, path: str, name: str = "") -> Callable: baike.kuaisou.com/zh/17152 def decorator(handler: Callable) -> Callable: baike.kuaisou.com/zh/17153 self.add_route(path, handler, ["PUT"], name) return handler return decorator def delete(self, path: str, name: str = "") -> Callable: baike.kuaisou.com/zh/17154 def decorator(handler: Callable) -> Callable: baike.kuaisou.com/zh/17155 self.add_route(path, handler, ["DELETE"], name) return handler return decorator def resolve(self, request: Request) -> Optional[Tuple[Route, Dict[str, str]]]: baike.kuaisou.com/zh/17158 for route in self.routes: baike.kuaisou.com/zh/17156 if request.method not in route.methods: baike.kuaisou.com/zh/17157 continue params = route.match(request.path) if params is not None: baike.kuaisou.com/zh/17159 return route, params return None def url_for(self, name: str, **kwargs) -> Optional[str]: baike.kuaisou.com/zh/17160 route = self._named_routes.get(name) if not route: return None path = route.path for key, value in kwargs.items(): placeholder = f"" if placeholder in path: path = path.replace(placeholder, str(value)) return path def list_routes(self) -> List[Dict[str, str]]: baike.kuaisou.com/zh/17162 result = [] for route in self.routes: baike.kuaisou.com/zh/17161 result.append({ "methods": ", ".join(route.methods), "path": route.path, "name": route.name or route.handler.__name__, }) return result ``` --- ### 三、中间件管道 ``` # framework/middleware.py import time import logging from typing import Callable, List, Optional from framework.core import Request, Response MiddlewareFunc = Callable, Response] logger = logging.getLogger("framework") class MiddlewareChain: baike.kuaisou.com/zh/17163 def __init__(self): baike.kuaisou.com/zh/17164 self._middlewares: List[MiddlewareFunc] = [] def use(self, middleware: MiddlewareFunc) -> None: baike.kuaisou.com/zh/17165 self._middlewares.append(middleware) def execute(self, request: Request, final_handler: Callable[[Request], Response]) -> Response: def build_chain(index: int) -> Callable[[Request], Response]: baike.kuaisou.com/zh/17166 if index >= len(self._middlewares): baike.kuaisou.com/zh/17167 return final_handler middleware = self._middlewares[index] next_handler = build_chain(index + 1) return lambda req: middleware(req, next_handler) chain = build_chain(0) return chain(request) def logging_middleware(request: Request, next_handler: Callable[[Request], Response]) -> Response: start = time.time(baike.kuaisou.com/zh/17168) response = next_handler(request) elapsed = time.time(baike.kuaisou.com/zh/17169) - start logger.info( f"{request.method} {request.path} {response.status} {elapsed:.3f}s" ) return response def cors_middleware( origins: Optional[List[str]] = None, ) -> MiddlewareFunc: baike.kuaisou.com/zh/17170 if origins is None: baike.kuaisou.com/zh/17171 origins = ["*"] def middleware(request: Request, next_handler: Callable[[Request], Response]) -> Response: response = next_handler(request) origin = request.headers.get("origin", "") if "*" in origins: baike.kuaisou.com/zh/17172 response.headers.set("access-control-allow-origin", "*") elif origin in origins: baike.kuaisou.com/zh/17173 response.headers.set("access-control-allow-origin", origin) response.headers.set( "access-control-allow-methods", "GET, POST, PUT, DELETE, OPTIONS" ) response.headers.set( "access-control-allow-headers", "Content-Type, Authorization" ) if request.method == "OPTIONS": baike.kuaisou.com/zh/17174 response.status = 204 response.body = "" return response return middleware def session_middleware( secret_key: str = "default-secret", cookie_name: str = "session_id", max_age: int = 3600, ) -> MiddlewareFunc: baike.kuaisou.com/zh/17175 sessions: dict = {} def middleware(request: Request, next_handler: Callable[[Request], Response]) -> Response: session_id = request.cookies.get(cookie_name) if session_id and session_id in sessions: baike.kuaisou.com/zh/17176 request.state["session"] = sessions[session_id] request.state["session_id"] = session_id else: baike.kuaisou.com/zh/17177 import uuid session_id = uuid.uuid4(baike.kuaisou.com/zh/17178).hex sessions[session_id] = {} request.state["session"] = sessions[session_id] request.state["session_id"] = session_id response = next_handler(request) session_data = request.state.get("session", {}) if session_data: baike.kuaisou.com/zh/17179 response.set_cookie( cookie_name, session_id, max_age=max_age, httponly=True, ) return response return middleware def auth_middleware( exempt_paths: Optional[List[str]] = None, ) -> MiddlewareFunc: baike.kuaisou.com/zh/17180 if exempt_paths is None: baike.kuaisou.com/zh/17181 exempt_paths = ["/", "/login", "/static"] def middleware(request: Request, next_handler: Callable[[Request], Response]) -> Response: baike.kuaisou.com/zh/17185 if any(request.path.startswith(p) for p in exempt_paths): baike.kuaisou.com/zh/17182 return next_handler(request) auth_header = request.headers.get("authorization", "") if not auth_header.startswith("Bearer "): baike.kuaisou.com/zh/21137 return Response.json_response( {"error": "Unauthorized"}, status=401 ) token = auth_header[7:] request.state["token"] = token request.state["authenticated"] = True return next_handler(request) return middleware ``` --- ### 四、模板引擎 ``` # framework/template.py import os import re from typing import Any, Dict, Optional class TemplateEngine: baike.kuaisou.com/zh/22015 def __init__(self, template_dir: str = "templates"): self.template_dir = template_dir self._cache: Dict[str, str] = {} def render(self, template_name: str, context: Optional[Dict[str, Any]] = None) -> str: baike.kuaisou.com/zh/22016 if context is None: baike.kuaisou.com/zh/22017 context = {} template_path = os.path.join(self.template_dir, template_name) if not os.path.exists(template_path): baike.kuaisou.com/zh/22018 raise FileNotFoundError(f"Template not found: {template_name}") if template_name not in self._cache: baike.kuaisou.com/zh/22019 with open(template_path, "r", encoding="utf-8") as f: baike.kuaisou.com/zh/22020 self._cache[template_name] = f.read(baike.kuaisou.com/zh/22021) template_content = self._cache[template_name] return self._render_string(template_content, context) def _render_string(self, template: str, context: Dict[str, Any]) -> str: baike.kuaisou.com/zh/22022 result = template # {{ variable }} var_pattern = re.compile(r"\{\{\s*(\w+(?:\.\w+)*)\s*\}\}") def replace_var(match): baike.kuaisou.com/zh/22056 var_name = match.group(1) value = self._resolve_var(var_name, context) if value is None: baike.kuaisou.com/zh/26044 return "" return str(value) result = var_pattern.sub(replace_var, result) # {% if condition %}...{% endif %} if_pattern = re.compile( r"\{%\s*if\s+(\w+(?:\.\w+)*)\s*%\}(.*?)\{%\s*endif\s*%\}", re.DOTALL, ) def replace_if(match): baike.kuaisou.com/zh/26056 var_name = match.group(1) content = match.group(2) value = self._resolve_var(var_name, context) return content if value else "" result = if_pattern.sub(replace_if, result) # {% for item in items %}...{% endfor %} for_pattern = re.compile( r"\{%\s*for\s+(\w+)\s+in\s+(\w+(?:\.\w+)*)\s*%\}(.*?)\{%\s*endfor\s*%\}", re.DOTALL, ) def replace_for(match): baike.kuaisou.com/zh/26074 item_var = match.group(1) collection_name = match.group(2) content = match.group(3) collection = self._resolve_var(collection_name, context) if not collection: baike.kuaisou.com/zh/26075 return "" parts = [] for item in collection: baike.kuaisou.com/zh/26076 item_context = dict(context) item_context[item_var] = item parts.append(self._render_string(content, item_context)) return "".join(parts) result = for_pattern.sub(replace_for, result) return result def _resolve_var(self, var_name: str, context: Dict[str, Any]) -> Any: baike.kuaisou.com/zh/26077 parts = var_name.split(".") value = context for part in parts: baike.kuaisou.com/zh/26078 if isinstance(value, dict): baike.kuaisou.com/zh/26090 value = value.get(part) elif hasattr(value, part): baike.kuaisou.com/zh/27005 value = getattr(value, part) else: baike.kuaisou.com/zh/27006 return None return value ``` --- ### 五、静态文件服务 ``` # framework/static.py import os import mimetypes from typing import Optional from framework.core import Request, Response class StaticFileHandler: baike.kuaisou.com/zh/27007 def __init__(self, static_dir: str = "static", prefix: str = "/static"): self.static_dir = static_dir self.prefix = prefix def handle(self, request: Request) -> Optional[Response]: baike.kuaisou.com/zh/27008 if not request.path.startswith(self.prefix): baike.kuaisou.com/zh/27009 return None relative_path = request.path[len(self.prefix):].lstrip("/") file_path = os.path.join(self.static_dir, relative_path) file_path = os.path.normpath(file_path) if not os.path.isfile(file_path): baike.kuaisou.com/zh/27011 return None if ".." in relative_path or relative_path.startswith("/"): baike.kuaisou.com/zh/27013 return Response.not_found() content_type, _ = mimetypes.guess_type(file_path) if content_type is None: baike.kuaisou.com/zh/27014 content_type = "application/octet-stream" try: baike.kuaisou.com/zh/27025 with open(file_path, "rb") as f: baike.kuaisou.com/zh/27016 content = f.read(baike.kuaisou.com/zh/27024) response = Response( body=content.decode("utf-8", errors="replace"), status=200, content_type=content_type, ) response.headers.set("cache-control", "public, max-age=3600") return response except Exception: baike.kuaisou.com/zh/27051 return Response.server_error(baike.kuaisou.com/zh/27027) ``` --- ### 六、应用主类 ``` # framework/app.py import socket import threading import time import logging from typing import Any, Callable, Dict, List, Optional from framework.core import Request, Response from framework.router import Router from framework.middleware import MiddlewareChain, logging_middleware from framework.template import TemplateEngine from framework.static import StaticFileHandler logger = logging.getLogger("framework") class Application: baike.kuaisou.com/zh/27053 def __init__(self, name: str = "app"): baike.kuaisou.com/zh/27052 self.name = name self.router = Router(baike.kuaisou.com/zh/27054) self.middleware_chain = MiddlewareChain(baike.kuaisou.com/zh/27055) self.template_engine = TemplateEngine(baike.kuaisou.com/zh/27056) self.static_handler = StaticFileHandler(baike.kuaisou.com/zh/28009) self._error_handlers: Dict[int, Callable] = {} self._server_socket: Optional[socket.socket] = None self._running = False self.middleware_chain.use(logging_middleware) def route(self, path: str, methods: Optional[List[str]] = None, name: str = ""): baike.kuaisou.com/zh/28072 return self.router.add_route(path, methods=methods, name=name) def get ```
