基于事件驱动的轻量级 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="<h1>404 Not Found</h1>", status=404)

    @classmethod
    def server_error(cls) -> "Response": baike.kuaisou.com/zh/17132 
        return cls(body="<h1>500 Internal Server Error</h1>", 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("<") and part.endswith(">"): 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"<{key}>"
            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