Move responses to another module
This commit is contained in:
@@ -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",
|
||||
]
|
||||
|
||||
54
slow/responses.py
Normal file
54
slow/responses.py
Normal file
@@ -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"
|
||||
)
|
||||
59
slow/slow.py
59
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):
|
||||
|
||||
Reference in New Issue
Block a user