From a7f0001659f46ce802109c6e034fbadf204ba9d2 Mon Sep 17 00:00:00 2001 From: 0880 <98263509+0880880@users.noreply.github.com> Date: Tue, 20 Jan 2026 21:06:59 +0330 Subject: [PATCH] Move responses to another module --- slow/__init__.py | 6 ++--- slow/responses.py | 54 +++++++++++++++++++++++++++++++++++++++++++ slow/slow.py | 59 ++++++----------------------------------------- 3 files changed, 63 insertions(+), 56 deletions(-) create mode 100644 slow/responses.py diff --git a/slow/__init__.py b/slow/__init__.py index 94f3593..661fb42 100644 --- a/slow/__init__.py +++ b/slow/__init__.py @@ -1,9 +1,8 @@ +from . import responses from .slow import ( JSONAPI, App, Headers, - HTTPResponse, - JSONResponse, Request, redirect, render, @@ -12,10 +11,9 @@ from .slow import ( __all__ = [ "JSONAPI", "App", - "HTTPResponse", - "JSONResponse", "render", "Request", "Headers", "redirect", + "responses", ] diff --git a/slow/responses.py b/slow/responses.py new file mode 100644 index 0000000..d809e02 --- /dev/null +++ b/slow/responses.py @@ -0,0 +1,54 @@ +import http.client +import json +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from .slow import App + + +class Response: + status: int + headers: list[str] + body: bytes + + def __init__(self, status: int, headers: list[str], body: bytes): + self.status = status + self.headers = headers + self.body = body + self.no_header = False + + def header(self, app: App) -> bytes: + header = f"HTTP/1.1 {self.status} {http.client.responses.get(self.status, 'Unkown Status Code')}\r\n" + header += f"Content-Length: {len(self.body)}\r\n" + for h in self.headers: + header += h + "\r\n" + + header += "\r\n" + + return header.encode(encoding="utf-8") + + def render(self, app: "App") -> bytes: + if self.no_header: + return self.header(app) + return self.header(app) + self.body + + +def HTTPResponse( + content: str | bytes, + status=200, + content_type="text/plain; charset=utf-8", + headers=[], +) -> Response: + if isinstance(content, str): + content_bytes = content.encode(encoding="utf-8") + else: + content_bytes = content + return Response( + status, [f"Content-Type: {content_type}", *headers], body=content_bytes + ) + + +def JSONResponse(d: dict, status=200) -> Response: + return HTTPResponse( + json.dumps(d), status=status, content_type="text/json; charset=utf-8" + ) diff --git a/slow/slow.py b/slow/slow.py index afec896..b7c8f42 100644 --- a/slow/slow.py +++ b/slow/slow.py @@ -1,5 +1,4 @@ import asyncio -import http.client import json import re import urllib.parse @@ -7,6 +6,8 @@ from json.decoder import JSONDecodeError from pathlib import Path from typing import Any, Awaitable, Callable, Optional +from .responses import HTTPResponse, JSONResponse, Response + PR = re.compile(r"\<([a-zA-Z_][a-zA-Z0-9_]*)\>") @@ -62,33 +63,6 @@ class Request: return None -class Response: - status: int - headers: list[str] - body: bytes - - def __init__(self, status: int, headers: list[str], body: bytes): - self.status = status - self.headers = headers - self.body = body - self.no_header = False - - def header(self, app: "App") -> bytes: - header = f"HTTP/1.1 {self.status} {http.client.responses.get(self.status, 'Unkown Status Code')}\r\n" - header += f"Content-Length: {len(self.body)}\r\n" - for h in self.headers: - header += h + "\r\n" - - header += "\r\n" - - return header.encode(encoding="utf-8") - - def render(self, app: "App") -> bytes: - if self.no_header: - return self.header(app) - return self.header(app) + self.body - - async def _default_404_route(request: Request): return "HTTP/1.1 404 Not Found\r\nContent-Type: text/html\r\n\r\n404 Not Found contact admin".encode( encoding="utf-8" @@ -319,25 +293,12 @@ class App: await server.serve_forever() -def HTTPResponse( - content: str | bytes, - status=200, - content_type="text/plain; charset=utf-8", - headers=[], -) -> Response: - if isinstance(content, str): - content_bytes = content.encode(encoding="utf-8") - else: - content_bytes = content - return Response( - status, [f"Content-Type: {content_type}", *headers], body=content_bytes - ) - - _value_pattern = re.compile(r"\{\{\s*([a-zA-Z_][a-zA-Z_0-9]*)\s*\}\}") -def render(file: str | Path, variables: dict[str, Any] = {}) -> Response: +def render( + file: str | Path, variables: dict[str, Any] = {} +) -> Response: # TODO Move to another module if isinstance(file, str): file = Path(file) content: str = file.read_text(encoding="utf-8") @@ -347,17 +308,11 @@ def render(file: str | Path, variables: dict[str, Any] = {}) -> Response: return HTTPResponse(content, content_type="text/html; charset=utf-8") -def redirect(location: str): +def redirect(location: str): # TODO Move to another module return Response(307, ["Location: {location}"], b"") -def JSONResponse(d: dict, status=200) -> Response: - return HTTPResponse( - json.dumps(d), status=status, content_type="text/json; charset=utf-8" - ) - - -def JSONAPI(func): +def JSONAPI(func): # TODO Move to another module async def wrapper(*args, **kwargs): result = await func(*args, **kwargs) if not isinstance(result, dict):