77 lines
2.1 KiB
Python
77 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
import typing
|
|
import urllib.parse
|
|
from dataclasses import dataclass
|
|
from typing import Any, Mapping
|
|
|
|
from case_insensitive_dict import CaseInsensitiveDict
|
|
|
|
from turbosloth.internal_types import MethodType, Scope, ASGIMessage
|
|
|
|
|
|
@dataclass
|
|
class BasicRequest:
|
|
method: MethodType
|
|
path: str
|
|
headers: CaseInsensitiveDict[str, str]
|
|
query: dict[str, list[Any] | Any]
|
|
body: bytes
|
|
|
|
def __init__(self,
|
|
method: MethodType,
|
|
path: str,
|
|
headers: Mapping[str, str],
|
|
query: dict[str, list[Any] | Any],
|
|
body: bytes):
|
|
self.method = method
|
|
self.path = path
|
|
self.headers = CaseInsensitiveDict(headers)
|
|
self.query = query
|
|
self.body = body
|
|
|
|
@classmethod
|
|
def from_scope(cls, scope: Scope) -> BasicRequest:
|
|
path = scope['path']
|
|
method = typing.cast(MethodType, scope['method'])
|
|
headers = {}
|
|
for key, value in scope.get('headers', []):
|
|
headers[key.decode('latin1')] = value.decode('latin1')
|
|
|
|
qs = scope['query_string'].decode('latin1')
|
|
print(qs)
|
|
query_raw = urllib.parse.parse_qs(qs)
|
|
query = {}
|
|
for k, v in query_raw.items():
|
|
if len(v) == 1:
|
|
v = v[0]
|
|
query[k] = v
|
|
query = typing.cast(dict[str, list[Any] | Any], query)
|
|
|
|
body = scope['body']
|
|
|
|
return BasicRequest(method, path, headers, query, body)
|
|
|
|
|
|
@dataclass
|
|
class BasicResponse:
|
|
code: int
|
|
headers: CaseInsensitiveDict[str, str]
|
|
body: bytes
|
|
|
|
def __init__(self, code: int, headers: Mapping[str, str], body: bytes):
|
|
self.code = code
|
|
self.headers = CaseInsensitiveDict(headers)
|
|
self.body = body
|
|
|
|
def into_start_message(self) -> ASGIMessage:
|
|
enc_headers = []
|
|
for k, v in self.headers.items():
|
|
enc_headers.append((k.encode('latin1'), v.encode('latin1')))
|
|
|
|
return {
|
|
'type': 'http.response.start',
|
|
'status': self.code,
|
|
'headers': enc_headers
|
|
}
|