Request/Response¶
Request - HTTP request. Response - HTTP response.
Introduction¶
In computer science, request-response or request-replica is one of the basic methods used by computers to communicate with each other on a network, in which the first computer sends a request for some data and the second one responds to the request. More specifically, it is a messaging pattern in which a requester sends a request message to a responder system, which receives and processes the request, ultimately returning a message in response. This is similar to a telephone call, in which the caller must wait for the recipient to pick up the phone before anything can be discussed.
Request¶
Request is a request that contains data for interaction between the client and the API: base URL, endpoint, method to use, headers, etc.
class Request:
"""
This class describes a request.
"""
def __init__(self, environ: dict, settings: Settings):
"""
Constructs a new instance.
:param environ: The environ
:type environ: dict
"""
self.environ: dict = environ
self.settings: Settings = settings
self.method: str = self.environ["REQUEST_METHOD"]
self.path: str = self.environ["PATH_INFO"]
self.GET: dict = self._build_get_params_dict(self.environ["QUERY_STRING"])
self.POST: dict = self._build_post_params_dict(self.environ["wsgi.input"].read())
self.user_agent: str = self.environ["HTTP_USER_AGENT"]
self.extra: dict = {}
logger.debug(f"New request created: {self.method} {self.path}")
def __getattr__(self, item: Any) -> Union[Any, None]:
"""
Magic method for get attrs (from extra)
:param item: The item
:type item: Any
:returns: Item from self.extra or None
:rtype: Union[Any, None]
"""
return self.extra.get(item, None)
def _build_get_params_dict(self, raw_params: str):
"""
Builds a get parameters dictionary.
:param raw_params: The raw parameters
:type raw_params: str
"""
return parse_qs(raw_params)
def _build_post_params_dict(self, raw_params: bytes):
"""
Builds a post parameters dictionary.
:param raw_params: The raw parameters
:type raw_params: bytes
"""
return parse_qs(raw_params.decode())
Request requires the following arguments to create:
environ (dictionary) - web environment (generated by gunicorn)
settings (pyechonext.config.Settings dataclass object)
Request has the following public attributes:
environ (dictionary) - web environment
settings (pyechonext.config.Settings dataclass object)
method (string) - http method
path (string) - path
GET (dictionary) - get request parameters
POST (dictionary) - post request parameters
user_agent (string) - User-Agent
extra (dictionary) - additional parameters (for example for middleware)
Request also has the following methods:
__getattr__- magic descriptor method for getting attributes (to get elements from the extra attribute)_build_get_params_dict- private method for parsing get request parameters_build_post_params_dict- private method for parsing post request parameters
Response¶
Response is a response that contains the data returned by the server, including content, status code, and headers.
import json
from typing import Dict, Iterable, Union, Any, List, Tuple, Optional
from socks import method
from loguru import logger
from pyechonext.request import Request
class Response:
"""
This dataclass describes a response.
"""
default_content_type: str = "text/html"
default_charset: str = "UTF-8"
unicode_errors: str = "strict"
default_conditional_response: bool = False
default_body_encoding: str = "UTF-8"
def __init__(
self,
request: Request,
use_i18n: bool = False,
status_code: Optional[int] = 200,
body: Optional[str] = None,
headers: Optional[Dict[str, str]] = {},
content_type: Optional[str] = None,
charset: Optional[str] = None,
**kwargs,
):
"""
Constructs a new instance.
:param request: The request
:type request: Request
:param use_i18n: The use i 18 n
:type use_i18n: bool
:param status_code: The status code
:type status_code: int
:param body: The body
:type body: str
:param headers: The headers
:type headers: Dict[str, str]
:param content_type: The content type
:type content_type: str
:param charset: The charset
:type charset: str
:param kwargs: The keywords arguments
:type kwargs: dictionary
"""
if status_code == 200:
self.status_code: str = "200 OK"
else:
self.status_code: str = str(status_code)
if content_type is None:
self.content_type: str = self.default_content_type
else:
self.content_type: str = content_type
if charset is None:
self.charset: str = self.default_charset
else:
self.charset: str = charset
if body is not None:
self.body: str = body
else:
self.body: str = ""
self._headerslist: list = headers
self._added_headers: list = []
self.request: Request = request
self.extra: dict = {}
self.use_i18n: bool = use_i18n
self.i18n_kwargs = kwargs
self._update_headers()
def __getattr__(self, item: Any) -> Union[Any, None]:
"""
Magic method for get attrs (from extra)
:param item: The item
:type item: Any
:returns: Item from self.extra or None
:rtype: Union[Any, None]
"""
return self.extra.get(item, None)
def _structuring_headers(self, environ):
headers = {
"Host": environ["HTTP_HOST"],
"Accept": environ["HTTP_ACCEPT"],
"User-Agent": environ["HTTP_USER_AGENT"],
}
for name, value in headers.items():
self._headerslist.append((name, value))
for header_tuple in self._added_headers:
self._headerslist.append(header_tuple)
def _update_headers(self) -> None:
"""
Sets the headers by environ.
:param environ: The environ
:type environ: dict
"""
self._headerslist = [
("Content-Type", f"{self.content_type}; charset={self.charset}"),
("Content-Length", str(len(self.body))),
]
def add_headers(self, headers: List[Tuple[str, str]]):
"""
Adds new headers.
:param headers: The headers
:type headers: List[Tuple[str, str]]
"""
for header in headers:
self._added_headers.append(header)
def _encode_body(self):
"""
Encodes a body.
"""
if self.content_type.split("/")[-1] == "json":
self.body = str(self.json)
try:
self.body = self.body.encode("UTF-8")
except AttributeError:
self.body = str(self.body).encode("UTF-8")
def __call__(self, environ: dict, start_response: method) -> Iterable:
"""
Makes the Response object callable.
:param environ: The environ
:type environ: dict
:param start_response: The start response
:type start_response: method
:returns: response body
:rtype: Iterable
"""
self._encode_body()
self._update_headers()
self._structuring_headers(environ)
logger.debug(
f"[{environ['REQUEST_METHOD']} {self.status_code}] Run response: {self.content_type}"
)
start_response(status=self.status_code, headers=self._headerslist)
return iter([self.body])
@property
def json(self) -> dict:
"""
Parse request body as JSON.
:returns: json body
:rtype: dict
"""
if self.body:
if self.content_type.split("/")[-1] == "json":
return json.dumps(self.body)
else:
return json.dumps(self.body.decode("UTF-8"))
return {}
def __repr__(self):
"""
Returns a unambiguous string representation of the object (for debug...).
:returns: String representation of the object.
:rtype: str
"""
return f"<{self.__class__.__name__} at 0x{abs(id(self)):x} {self.status_code}>"
Response has the following arguments:
request (request class object) - request
[optional] status_code (integer value) - response status code
[optional] body (string) - response body
[optional] headers (dictionary) - response headers
[optional] content_type (string) - response content type
[optional] charset (string) - response encoding
[optional] use_i18n (boolean value) - whether to use i18n (default False)
Response has the following attributes:
status_code (string) - status code (default “200 OK”)
content_type (string) - content type (defaults to default_content_type)
charset (string) - encoding (defaults to default_charset)
body (string) - body of the answer (defaults to the empty string)
_headerslist(list) - private list of response headers_added_headers(list) - private list of added response headersrequest (request class object) - request
extra (dictionary) - additional parameters
Response has the following methods:
__getattr__- magic descriptor method for getting attributes (to get elements from the extra attribute)_structuring_headers- private method for structuring headers from the web environment_update_headers- private method for updating (overwriting) header listsadd_headers- public method for adding headers_encode_body- response body encoding__call__- magic method, makes the Response object callablejson- class property for receiving the response body as json