PK!e| | molten/__init__.pyfrom .app import App, BaseApp from .dependency_injection import Component, DependencyInjector, DependencyResolver from .errors import ( DIError, HeaderMissing, HTTPError, MoltenError, ParamMissing, RequestParserNotAvailable, RouteNotFound, RouteParamMissing ) from .http import Cookies, Headers, QueryParams, Request, Response from .http.status_codes import * from .middleware import ResponseRendererMiddleware from .parsers import JSONParser, RequestParser, URLEncodingParser from .renderers import JSONRenderer, ResponseRenderer from .router import Include, Route, Router from .testing import TestClient, to_environ from .typing import ( Header, Host, Method, Middleware, Port, QueryParam, QueryString, RequestBody, RequestData, RequestInput, Scheme ) __version__ = "0.0.0" __all__ = [ "BaseApp", "App", "Middleware", # Router "Router", "Route", "Include", # HTTP "Method", "Scheme", "Host", "Port", "QueryString", "QueryParams", "QueryParam", "Headers", "Header", "RequestInput", "RequestBody", "RequestData", "Cookies", "Request", "Response", # Dependency-injection "DependencyInjector", "DependencyResolver", "Component", # Parsers "RequestParser", "JSONParser", "URLEncodingParser", # Renderers "ResponseRenderer", "JSONRenderer", # Middleware "ResponseRendererMiddleware", # Errors "MoltenError", "DIError", "HTTPError", "HeaderMissing", "ParamMissing", "RequestParserNotAvailable", "RouteNotFound", "RouteParamMissing", # Testing "TestClient", "to_environ", # Status codes # 1xx "HTTP_100", "HTTP_101", "HTTP_102", # 2xx "HTTP_200", "HTTP_201", "HTTP_202", "HTTP_203", "HTTP_204", "HTTP_205", "HTTP_206", "HTTP_207", "HTTP_208", # 3xx "HTTP_300", "HTTP_301", "HTTP_302", "HTTP_303", "HTTP_304", "HTTP_305", "HTTP_307", "HTTP_308", # 4xx "HTTP_400", "HTTP_401", "HTTP_402", "HTTP_403", "HTTP_404", "HTTP_405", "HTTP_406", "HTTP_407", "HTTP_408", "HTTP_409", "HTTP_410", "HTTP_411", "HTTP_412", "HTTP_413", "HTTP_414", "HTTP_415", "HTTP_416", "HTTP_417", "HTTP_418", "HTTP_421", "HTTP_422", "HTTP_423", "HTTP_424", "HTTP_426", "HTTP_428", "HTTP_429", "HTTP_431", "HTTP_444", "HTTP_451", "HTTP_499", # 5xx "HTTP_500", "HTTP_501", "HTTP_502", "HTTP_503", "HTTP_504", "HTTP_505", "HTTP_506", "HTTP_507", "HTTP_508", "HTTP_510", "HTTP_511", "HTTP_599", ] PK!|gPP molten/app.pyimport logging import sys from functools import partial from typing import Any, Callable, Iterable, List, Optional from wsgiref.util import FileWrapper # type: ignore from .components import ( CookiesComponent, HeaderComponent, QueryParamComponent, RequestBodyComponent, RequestDataComponent, RouteParamsComponent ) from .dependency_injection import Component, DependencyInjector from .errors import RequestParserNotAvailable from .http import HTTP_204, HTTP_404, HTTP_415, HTTP_500, Headers, QueryParams, Request, Response from .middleware import ResponseRendererMiddleware from .parsers import JSONParser, RequestParser, URLEncodingParser from .renderers import JSONRenderer, ResponseRenderer from .router import RouteLike, Router from .typing import ( Environ, Host, Method, Middleware, Port, QueryString, RequestInput, Scheme, StartResponse ) LOGGER = logging.getLogger(__name__) class BaseApp: """Base class for App implementations. """ def __init__( self, routes: Optional[List[RouteLike]] = None, middleware: Optional[List[Middleware]] = None, components: Optional[List[Component]] = None, parsers: Optional[List[RequestParser]] = None, renderers: Optional[List[ResponseRenderer]] = None, ) -> None: self.router = Router(routes) self.add_route = self.router.add_route self.add_routes = self.router.add_routes self.reverse_uri = self.router.reverse_uri self.parsers = parsers or [ JSONParser(), URLEncodingParser(), ] self.renderers = renderers or [ JSONRenderer(), ] self.middleware = middleware or [ ResponseRendererMiddleware(self.renderers) ] self.components = (components or []) + [ HeaderComponent(), CookiesComponent(), QueryParamComponent(), RequestBodyComponent(), RequestDataComponent(self.parsers), ] self.injector = DependencyInjector(self.components) def handle_404(self) -> Response: """Called whenever a route cannot be found. """ return Response(HTTP_404, content="Not Found") def handle_415(self) -> Response: """Called whenever a request comes in with an unsupported content type. """ return Response(HTTP_415, content="Unsupported Media Type") def handle_exception(self, exception: BaseException) -> Response: """Called whenever an unhandled exception occurs in middleware or a handler. Dependencies are injected into this just like a normal handler. """ LOGGER.exception("An unhandled exception occurred.") return Response(HTTP_500, content="Internal Server Error") def __call__(self, environ: Environ, start_response: StartResponse) -> Iterable[bytes]: # pragma: no cover raise NotImplementedError("apps must implement '__call__'") class App(BaseApp): """An application that implements the WSGI interface. """ def __call__(self, environ: Environ, start_response: StartResponse) -> Iterable[bytes]: request = Request.from_environ(environ) resolver = self.injector.get_resolver({ Request: request, Method: Method(request.method), Scheme: Scheme(request.scheme), Host: Host(request.host), Port: Port(request.port), QueryString: QueryString(environ["QUERY_STRING"]), QueryParams: request.params, Headers: request.headers, RequestInput: RequestInput(request.body_file), }) try: handler: Callable[..., Any] route_and_params = self.router.match(request.method, request.path) if route_and_params is not None: route, params = route_and_params handler = route.handler resolver.add_component(RouteParamsComponent(params)) else: handler = self.handle_404 handler = resolver.resolve(handler) for middleware in reversed(self.middleware): handler = resolver.resolve(middleware(handler)) exc_info = None response = handler() except RequestParserNotAvailable: exc_info = None response = resolver.resolve(self.handle_415)() except Exception as e: exc_info = sys.exc_info() response = resolver.resolve(self.handle_exception, {"exception": e})() response.headers.add("content-length", str(response.content_length)) start_response(response.status, list(response.headers), exc_info) if response.status != HTTP_204: wrapper = environ.get("wsgi.file_wrapper", FileWrapper) return wrapper(response.stream) else: return [] PK!r vmolten/common.pyfrom collections import defaultdict from typing import Dict, Iterable, Iterator, List, Optional, Tuple, Union Mapping = Union[Dict[str, Union[str, List[str]]], List[Tuple[str, str]]] class MultiDict(Iterable[Tuple[str, str]]): """A mapping from param names to lists of values. Once constructed, these instances cannot be modified. """ __slots__ = ["_data"] def __init__(self, mapping: Optional[Mapping] = None) -> None: self._data: Dict[str, List[str]] = defaultdict(list) self._add_all(mapping or {}) def _add(self, name: str, value: Union[str, List[str]]) -> None: """Add values for a particular key. """ if isinstance(value, list): self._data[name].extend(value) else: self._data[name].append(value) def _add_all(self, mapping: Mapping) -> None: """Add a group of values. """ items: Iterable[Tuple[str, Union[str, List[str]]]] if isinstance(mapping, dict): items = mapping.items() else: items = mapping for name, value_or_values in items: self._add(name, value_or_values) def get(self, name: str, default: Optional[str] = None) -> Optional[str]: """Get the last value for a given key. """ try: return self[name] except KeyError: return default def get_all(self, name: str) -> List[str]: """Get all the values for a given key. """ return self._data[name] def __getitem__(self, name: str) -> str: """Get the last value for a given key. Raises: KeyError: When the key is missing. """ try: return self._data[name][-1] except IndexError: raise KeyError(name) def __iter__(self) -> Iterator[Tuple[str, str]]: """Iterate over all the parameters. """ for name, values in self._data.items(): for value in values: yield name, value def __repr__(self) -> str: mapping = ", ".join(f"{repr(name)}: {repr(value)}" for name, value in self._data.items()) return f"{type(self).__name__}({{{mapping}}})" PK! CCmolten/components.pyfrom inspect import Parameter from typing import Any, Dict, List, Optional, TypeVar from .dependency_injection import DependencyResolver from .errors import HeaderMissing, HTTPError, ParamMissing, RequestParserNotAvailable from .http import HTTP_400, Cookies, Headers, QueryParams from .parsers import RequestParser from .typing import ( Header, QueryParam, RequestBody, RequestData, RequestInput, extract_optional_annotation ) _T = TypeVar("_T") class HeaderComponent: """Retrieves a named header from the request. Examples: def handle(content_type: Header) -> Response: ... def handle(content_type: Optional[Header]) -> Response: ... """ is_cacheable = False is_singleton = False def can_handle_parameter(self, parameter: Parameter) -> bool: _, annotation = extract_optional_annotation(parameter.annotation) return annotation is Header def resolve(self, parameter: Parameter, headers: Headers) -> Optional[str]: is_optional, _ = extract_optional_annotation(parameter.annotation) header_name = parameter.name.replace("_", "-") try: return headers[header_name] except HeaderMissing: if is_optional: return None raise HTTPError(HTTP_400, {header_name: "missing"}) class QueryParamComponent: """Retrieves a named query param from the request. Examples: def handle(x: QueryParam) -> Response: ... def handle(x: Optional[QueryParam]) -> Response: ... """ is_cacheable = False is_singleton = False def can_handle_parameter(self, parameter: Parameter) -> bool: _, annotation = extract_optional_annotation(parameter.annotation) return annotation is QueryParam def resolve(self, parameter: Parameter, params: QueryParams) -> Optional[str]: is_optional, _ = extract_optional_annotation(parameter.annotation) try: return params[parameter.name] except ParamMissing: if is_optional: return None raise HTTPError(HTTP_400, {parameter.name: "missing"}) class RequestBodyComponent: """A component that reads the entire request body into a string. Examples: def handle(body: RequestBody) -> Response: ... """ is_cacheable = True is_singleton = False def can_handle_parameter(self, parameter: Parameter) -> bool: return parameter.annotation is RequestBody def resolve(self, content_length: Header, body_file: RequestInput) -> RequestBody: return RequestBody(body_file.read(int(content_length))) class RequestDataComponent: """A component that parses request data based on the content-type header and the set of registered request parsers. If no request parser is available, then an HTTP 415 response is returned. Examples: def handle(data: RequestData) -> Response: ... """ __slots__ = ["parsers"] is_cacheable = True is_singleton = False def __init__(self, parsers: List[RequestParser]) -> None: self.parsers = parsers def can_handle_parameter(self, parameter: Parameter) -> bool: return parameter.annotation is RequestData def resolve(self, content_type: Optional[Header], resolver: DependencyResolver) -> RequestData: content_type_str = (content_type or "").lower() for parser in self.parsers: if parser.can_parse_content(content_type_str): return RequestData(resolver.resolve(parser.parse)()) raise RequestParserNotAvailable(content_type_str) class CookiesComponent: """A component that parses request cookies. Examples: def handle(cookies: Cookies) -> Response: cookies["some-cookie"] """ is_cacheable = True is_singleton = False def can_handle_parameter(self, parameter: Parameter) -> bool: return parameter.annotation is Cookies def resolve(self, cookie: Optional[Header]) -> Cookies: if cookie is None: return Cookies() return Cookies.parse(cookie) class RouteParamsComponent: """A component that resolves route params. Examples: def handle(name: str, age: int) -> Response: ... app.add_route(Route("/{name}/{age}", handle)) """ __slots__ = ["params"] is_cacheable = False is_singleton = False def __init__(self, params: Dict[str, str]) -> None: self.params = params def can_handle_parameter(self, parameter: Parameter) -> bool: return parameter.name in self.params def resolve(self, parameter: Parameter) -> Any: try: return parameter.annotation(self.params[parameter.name]) except (TypeError, ValueError): raise HTTPError(HTTP_400, {parameter.name: f"expected {parameter.annotation.__name__} value"}) PK! molten/contrib/settings.pyimport os from inspect import Parameter from typing import Optional try: import toml except ImportError: # pragma: no cover raise ImportError("'toml' package missing. Run 'pip install toml'.") class Settings(dict): """A dictionary of settings parsed from a TOML file. """ @classmethod def from_path(cls, path: str, environment: str) -> "Settings": """Load a TOML file into a dictionary. Raises: FileNotFoundError: When the settings file does not exist. """ all_settings = toml.load(open(path)) common_settings = all_settings.get("common", {}) environment_settings = all_settings.get(environment, {}) return Settings({**common_settings, **environment_settings}) class SettingsComponent: """A component that loads settings from a TOML file. The settings file should have a "common" section and one section for each environment the application is expected to run in. The environment-specific settings are merged on top of the "common" settings on load. The current environment is determined by the ``ENVIRONMENT`` config variable. Example settings file:: [common] conn_pooling = true conn_pool_size = 1 [dev] [prod] con_pool_size = 32 """ __slots__ = ["path", "environment"] is_cacheable = True is_singleton = True def __init__(self, path: str = "./settings.toml", environment: Optional[str] = None) -> None: self.path = path self.environment = environment or os.getenv("ENVIRONMENT", "dev") def can_handle_parameter(self, parameter: Parameter) -> bool: return parameter.annotation is Settings def resolve(self) -> Settings: return Settings.from_path(self.path, self.environment) PK!?molten/contrib/templates.pyfrom inspect import Parameter from molten import HTTP_200, Response try: import jinja2 except ImportError: # pragma: no cover raise ImportError("'jinja2' missing. Run 'pip install jinja2'.") class Templates: """Renders jinja2 templates. """ __slots__ = ["environment"] def __init__(self, path: str) -> None: self.environment = jinja2.Environment( loader=jinja2.FileSystemLoader(path), ) def render(self, template_name: str, **context) -> Response: template = self.environment.get_template(template_name) rendered_template = template.render(**context) return Response(HTTP_200, content=rendered_template, headers={ "content-type": "text/html", }) class TemplatesComponent: """A component that builds a jinja2 template renderer. """ __slots__ = ["path"] is_cacheable = True is_singleton = True def __init__(self, path: str) -> None: self.path = path def can_handle_parameter(self, parameter: Parameter) -> bool: return parameter.annotation is Templates def resolve(self) -> Templates: return Templates(self.path) PK!n1">>molten/dependency_injection.pyimport functools import inspect from inspect import Parameter from typing import Any, Callable, List, Optional, TypeVar, no_type_check from typing_extensions import Protocol from .errors import DIError _T = TypeVar("_T", covariant=True) class Component(Protocol[_T]): # pragma: no cover """The component protocol. """ @property def is_cacheable(self) -> bool: """If True, then the component will be cached within a resolver. This should be True for most components. Defaults to True. """ ... @property def is_singleton(self) -> bool: """If True, then the component will be treated as a singleton and cached after its first use. Defaults to False. """ ... def can_handle_parameter(self, parameter: Parameter) -> bool: """Returns True when parameter represents the desired component. """ ... @no_type_check def resolve(self) -> _T: """Returns an instance of the component. """ ... class DependencyInjector: """The dependency injector maintains component state and instantiates the resolver. """ __slots__ = [ "components", "singletons", ] def __init__(self, components: List[Component]) -> None: self.components = components or [] self.singletons = singletons = {} # type: dict for component in components: if getattr(component, "is_singleton", False) and component not in singletons: resolver = self.get_resolver() resolved_component = resolver.resolve(component.resolve) singletons[component] = resolved_component() def get_resolver( self, instances: Optional[dict] = None, ) -> "DependencyResolver": """Get the resolver for this Injector. """ return DependencyResolver( self.components, {**self.singletons, **(instances or {})}, ) class DependencyResolver: """The resolver does the work of actually filling in all of a function's dependencies and returning a thunk. """ __slots__ = [ "components", "instances", ] def __init__(self, components: List[Component], instances: dict) -> None: self.components = components[:] self.instances = instances def add_component(self, component: Component) -> None: self.components.append(component) def resolve( self, fn: Callable[..., Any], params: Optional[dict] = None, resolving_parameter: Optional[Parameter] = None, ) -> Callable[..., Any]: """Resolve a function's dependencies. """ @functools.wraps(fn) def resolved_fn(): nonlocal params params = params or {} signature = inspect.signature(fn) for parameter in signature.parameters.values(): if parameter.name in params: continue # When Parameter is requested then we assume that the # caller actually wants the parameter that resolved the # current component that's being resolved. See QueryParam # or Header components for an example. if parameter.annotation is Parameter: params[parameter.name] = resolving_parameter continue # When a DependencyResolver is requested, then we assume # the caller wants this instance. if parameter.annotation is DependencyResolver: params[parameter.name] = self continue # If our instances contains an exact match for a type, # then we return that. This is used to inject the current # Request object among other things. try: params[parameter.name] = self.instances[parameter.annotation] continue except KeyError: pass for component in self.components: if component.can_handle_parameter(parameter): try: params[parameter.name] = self.instances[component] except KeyError: factory = self.resolve(component.resolve, resolving_parameter=parameter) params[parameter.name] = instance = factory() if getattr(component, "is_cacheable", True): self.instances[component] = instance break else: raise DIError(f"cannot resolve parameter {parameter} of function {fn}") return fn(**params) return resolved_fn PK!molten/errors.pyfrom typing import Any class MoltenError(Exception): """Base class for all Molten exceptions. """ def __init__(self, message: str) -> None: self.message = message def __str__(self) -> str: return self.message class DIError(MoltenError): """Raised when a dependency cannot be resolved. """ class HTTPError(MoltenError): """Base class for HTTP errors. Handlers and middleware can raise these to short-circuit execution. """ def __init__(self, status: str, response: Any) -> None: self.status = status self.response = response def __str__(self) -> str: # pragma: no cover return self.status class HeaderMissing(MoltenError): """Raised by Headers.__getitem__ when a header does not exist. """ class ParamMissing(MoltenError): """Raised by QueryParams.__getitem__ when a param is missing. """ class RequestParserNotAvailable(MoltenError): """Raised when no request parser can handle the incoming request. """ class RouteNotFound(MoltenError): """Raised when trying to reverse route to a route that doesn't exist. """ class RouteParamMissing(MoltenError): """Raised when a param is missing while reversing a route. """ PK!NNmolten/http/__init__.pyfrom .cookies import Cookies from .headers import Headers from .query_params import QueryParams from .request import Request from .response import Response from .status_codes import * __all__ = [ "Request", "Response", "Headers", "QueryParams", "Cookies", # 1xx "HTTP_100", "HTTP_101", "HTTP_102", # 2xx "HTTP_200", "HTTP_201", "HTTP_202", "HTTP_203", "HTTP_204", "HTTP_205", "HTTP_206", "HTTP_207", "HTTP_208", # 3xx "HTTP_300", "HTTP_301", "HTTP_302", "HTTP_303", "HTTP_304", "HTTP_305", "HTTP_307", "HTTP_308", # 4xx "HTTP_400", "HTTP_401", "HTTP_402", "HTTP_403", "HTTP_404", "HTTP_405", "HTTP_406", "HTTP_407", "HTTP_408", "HTTP_409", "HTTP_410", "HTTP_411", "HTTP_412", "HTTP_413", "HTTP_414", "HTTP_415", "HTTP_416", "HTTP_417", "HTTP_418", "HTTP_421", "HTTP_422", "HTTP_423", "HTTP_424", "HTTP_426", "HTTP_428", "HTTP_429", "HTTP_431", "HTTP_444", "HTTP_451", "HTTP_499", # 5xx "HTTP_500", "HTTP_501", "HTTP_502", "HTTP_503", "HTTP_504", "HTTP_505", "HTTP_506", "HTTP_507", "HTTP_508", "HTTP_510", "HTTP_511", "HTTP_599", ] PK!molten/http/cookies.pyclass Cookies(dict): """A set of request cookies. """ @classmethod def parse(cls, cookie_header: str) -> "Cookies": """Turn a cookie header into a Cookies instance. """ cookies = cls() cookie_strings = cookie_header.split(";") for cookie in cookie_strings: try: name, value = cookie.lstrip().split("=", 1) cookies[name] = value except ValueError: continue return cookies PK!D,molten/http/headers.pyfrom collections import defaultdict from typing import Dict, Iterable, Iterator, List, Optional, Tuple, Union from ..errors import HeaderMissing from ..typing import Environ #: An alias representing a dictionary of headers. HeadersDict = Dict[str, Union[str, List[str]]] #: WSGI keeps these separate from other headers. CONTENT_VARS = {"CONTENT_LENGTH", "CONTENT_TYPE"} class Headers(Iterable[Tuple[str, str]]): """A mapping from case-insensitive header names to lists of values. """ __slots__ = ["_headers"] def __init__(self, mapping: Optional[HeadersDict] = None) -> None: self._headers: Dict[str, List[str]] = defaultdict(list) self.add_all(mapping or {}) @classmethod def from_environ(cls, environ: Environ) -> "Headers": """Construct a Headers instance from a WSGI environ. """ headers = {} for name, value in environ.items(): if name in CONTENT_VARS: headers[name.replace("_", "-")] = value elif name.startswith("HTTP_"): headers[_parse_environ_header(name)] = [value] return cls(headers) def add(self, header: str, value: Union[str, List[str]]) -> None: """Add values for a particular header. """ if isinstance(value, list): self._headers[header.lower()].extend(value) else: self._headers[header.lower()].append(value) def add_all(self, mapping: HeadersDict) -> None: """Add a group of headers. """ for header, value_or_values in mapping.items(): self.add(header, value_or_values) def get(self, header: str, default: Optional[str] = None) -> Optional[str]: """Get the last value for a given header. """ try: return self[header] except HeaderMissing: return default def get_all(self, header: str) -> List[str]: """Get all the values for a given header. """ return self._headers[header.lower()] def get_int(self, header: str, default: Optional[int] = None) -> Optional[int]: """Get the last value for a given header as an integer. """ try: return int(self[header]) except HeaderMissing: return default def __delitem__(self, header: str) -> None: """Delete all the values for a given header. """ del self._headers[header.lower()] def __getitem__(self, header: str) -> str: """Get the last value for a given header. Raises: HeaderMissing: When the header is missing. """ try: return self._headers[header.lower()][-1] except IndexError: raise HeaderMissing(header) def __setitem__(self, header: str, value: str) -> None: """Replace a header's values. """ self._headers[header.lower()] = [value] def __iter__(self) -> Iterator[Tuple[str, str]]: """Iterate over all the headers. """ for header, values in self._headers.items(): for value in values: yield header, value def __repr__(self) -> str: mapping = ", ".join(f"{repr(name)}: {repr(value)}" for name, value in self._headers.items()) return f"Headers({{{mapping}}})" #: The number of characters that are stripped from the beginning of #: every header name in a WSGI environ. HEADER_PREFIX_LEN = len("HTTP_") def _parse_environ_header(header: str) -> str: return header[HEADER_PREFIX_LEN:].replace("_", "-") PK!O#molten/http/query_params.pyfrom typing import Dict, List, Optional, Union from urllib.parse import parse_qsl from ..common import MultiDict from ..errors import ParamMissing from ..typing import Environ ParamsDict = Dict[str, Union[str, List[str]]] class QueryParams(MultiDict): """A mapping from param names to lists of values. Once constructed, these instances cannot be modified. """ @classmethod def from_environ(cls, environ: Environ) -> "QueryParams": """Construct a QueryParams instance from a WSGI environ. """ return cls.parse(environ["QUERY_STRING"]) @classmethod def parse(cls, query_string: str) -> "QueryParams": """Construct a QueryParams instance from a query string. """ return cls(parse_qsl(query_string)) def get(self, name: str, default: Optional[str] = None) -> Optional[str]: """Get the last value for a given key. """ try: return self[name] except ParamMissing: return default def __getitem__(self, name: str) -> str: """Get the last value for a given key. Raises: ParamMissing: When the key is missing. """ try: return self._data[name][-1] except IndexError: raise ParamMissing(name) PK!Tmolten/http/request.pyfrom io import BytesIO from typing import BinaryIO, Optional, Union from ..typing import Environ from .headers import Headers from .query_params import ParamsDict, QueryParams class Request: """Represents an individual HTTP request. """ __slots__ = [ "method", "scheme", "host", "port", "path", "params", "headers", "body_file", ] def __init__( self, *, method: str = "GET", scheme: str = "", host: str = "", port: int = 0, path: str = "/", params: Optional[Union[ParamsDict, QueryParams]] = None, headers: Optional[Union[dict, Headers]] = None, body_file: Optional[BinaryIO] = None, ) -> None: self.method = method self.scheme = scheme self.host = host self.port = port self.path = path self.body_file = body_file or BytesIO() if isinstance(headers, dict): self.headers: Headers = Headers(headers) else: self.headers = headers or Headers() if not params: self.params = QueryParams() elif isinstance(params, dict): self.params = QueryParams(params) else: self.params = params @classmethod def from_environ(cls, environ: Environ) -> "Request": """Construct a Request object from a WSGI environ. """ return Request( method=environ["REQUEST_METHOD"], scheme=environ["wsgi.url_scheme"], host=environ.get("HTTP_HOST", ""), port=environ.get("SERVER_PORT", 0), path=environ.get("SCRIPT_NAME", "") + environ.get("PATH_INFO", ""), params=QueryParams.from_environ(environ), headers=Headers.from_environ(environ), body_file=environ["wsgi.input"], ) def __repr__(self) -> str: return ( f"Request(method={repr(self.method)}, scheme={repr(self.scheme)}, host={repr(self.host)}, " f"port={repr(self.port)}, path={repr(self.path)}, params={repr(self.params)}, " f"headers={repr(self.headers)}, body_file={repr(self.body_file)})" ) PK!9Xmolten/http/response.pyimport io import os from typing import BinaryIO, Optional, Union from .headers import Headers, HeadersDict class Response: """An HTTP response. """ __slots__ = [ "status", "headers", "stream", ] def __init__( self, status: str, headers: Optional[Union[HeadersDict, Headers]] = None, content: Optional[str] = None, stream: Optional[BinaryIO] = None, encoding: str = "utf-8", ) -> None: self.status = status if isinstance(headers, dict): self.headers = Headers(headers) else: self.headers = headers or Headers() if content is not None: self.stream: BinaryIO = io.BytesIO(content.encode(encoding)) elif stream is not None: self.stream: BinaryIO = stream else: self.stream: BinaryIO = io.BytesIO() @property def content_length(self): content_length = self.headers.get_int("content_length") if content_length is None: try: stream_stat = os.fstat(self.stream.fileno()) content_length = stream_stat.st_size except OSError: self.stream.seek(0, os.SEEK_END) content_length = self.stream.tell() self.stream.seek(0, os.SEEK_SET) return content_length def __repr__(self) -> str: return f"Response(status={repr(self.status)}, headers={repr(self.headers)})" PK!d molten/http/status_codes.py# 1xx HTTP_100 = "100 Continue" HTTP_101 = "101 Switching Protocols" HTTP_102 = "102 Processing" # 2xx HTTP_200 = "200 OK" HTTP_201 = "201 Created" HTTP_202 = "202 Accepted" HTTP_203 = "203 Non-Authoritative Information" HTTP_204 = "204 No Content" HTTP_205 = "205 Reset Content" HTTP_206 = "206 Partial Content" HTTP_207 = "207 Multi-Status" HTTP_208 = "208 Already Reported" # 3xx HTTP_300 = "300 Multiple Choices" HTTP_301 = "301 Moved Permanently" HTTP_302 = "302 Found" HTTP_303 = "303 See Other" HTTP_304 = "304 Not Modified" HTTP_305 = "305 Use Proxy" HTTP_307 = "307 Temporary Redirect" HTTP_308 = "308 Permanent Redirect" # 4xx HTTP_400 = "400 Bad Request" HTTP_401 = "401 Unauthorized" HTTP_402 = "402 Payment Required" HTTP_403 = "403 Forbidden" HTTP_404 = "404 Not Found" HTTP_405 = "405 Method Not Allowed" HTTP_406 = "406 Not Acceptable" HTTP_407 = "407 Proxy Authentication Required" HTTP_408 = "408 Request Timeout" HTTP_409 = "409 Conflict" HTTP_410 = "410 Gone" HTTP_411 = "411 Length Required" HTTP_412 = "412 Precondition Failed" HTTP_413 = "413 Payload Too Large" HTTP_414 = "414 Request-URI Too Long" HTTP_415 = "415 Unsupported Media Type" HTTP_416 = "416 Requested Range Not Satisfiable" HTTP_417 = "417 Expectation Failed" HTTP_418 = "418 I'm a teapot" HTTP_421 = "421 Misdirected Request" HTTP_422 = "422 Unprocessable Entity" HTTP_423 = "423 Locked" HTTP_424 = "424 Failed Dependency" HTTP_426 = "426 Upgrade Required" HTTP_428 = "428 Precondition Required" HTTP_429 = "429 Too Many Requests" HTTP_431 = "431 Request Header Fields Too Large" HTTP_444 = "444 Connection Closed Without Response" HTTP_451 = "451 Unavailable For Legal Reasons" HTTP_499 = "499 Client Closed Request" HTTP_415 = "415 Unsupported Media Type" # 5xx HTTP_500 = "500 Internal Server Error" HTTP_501 = "501 Not Implmeneted" HTTP_502 = "502 Bad Gateway" HTTP_503 = "503 Service Unavailable" HTTP_504 = "504 Gateway Timeout" HTTP_505 = "505 HTTP Version Not Supported" HTTP_506 = "506 Variant Also Negotiates" HTTP_507 = "507 Insufficient Storage" HTTP_508 = "508 Loop Detected" HTTP_510 = "510 Not Extended" HTTP_511 = "511 Network Authentication Required" HTTP_599 = "599 Network Connect Timeout Error" PK!2llmolten/middleware.pyfrom typing import Any, Callable, List from .errors import HeaderMissing, HTTPError from .http import HTTP_200, HTTP_406, Request, Response from .renderers import ResponseRenderer class ResponseRendererMiddleware: """A middleware that renders responses. """ def __init__(self, renderers: List[ResponseRenderer]) -> None: self.renderers = renderers def __call__(self, handler: Callable[..., Any]) -> Callable[..., Response]: def handle(request: Request) -> Response: try: response = handler() if isinstance(response, Response): return response if isinstance(response, tuple): status, response = response else: status, response = HTTP_200, response except HTTPError as e: status, response = e.status, e.response try: accept = request.headers["accept"] except HeaderMissing: accept = "*/*" for renderer in self.renderers: if accept == "*/*" or renderer.can_render_response(accept): return renderer.render(status, response) return Response(HTTP_406, content="Not Acceptable", headers={ "content-type": "text/plain", }) return handle PK!molten/parsers.pyimport json from typing import Any, no_type_check from urllib.parse import parse_qsl from typing_extensions import Protocol from .common import MultiDict from .typing import RequestBody class RequestParser(Protocol): # pragma: no cover """Protocol for request parsers. """ def can_parse_content(self, content_type: str) -> bool: ... @no_type_check def parse(self): ... class JSONParser: """A JSON request parser. """ def can_parse_content(self, content_type: str) -> bool: return content_type.startswith("application/json") def parse(self, data: RequestBody) -> Any: return json.loads(data) class URLEncodingParser: """A parser for urlencoded requests. """ def can_parse_content(self, content_type: str) -> bool: return content_type.startswith("application/x-www-form-urlencoded") def parse(self, data: RequestBody) -> Any: return MultiDict(parse_qsl(data.decode("utf-8"))) PK!о//molten/py.typed# This package uses inline types as per PEP561.PK!amolten/renderers.pyimport json from typing import Any from typing_extensions import Protocol from .http import Response class ResponseRenderer(Protocol): # pragma: no cover """Protocol for response renderers. """ def can_render_response(self, accept: str) -> bool: ... def render(self, status: str, response_data: Any) -> Response: ... class JSONRenderer: """A JSON response renderer. """ def can_render_response(self, accept: str) -> bool: return accept.startswith("application/json") def render(self, status: str, response_data: Any) -> Response: return Response(status, content=json.dumps(response_data), headers={ "content-type": "application/json; charset=utf-8", }) PK!ɢCmolten/router.pyimport re from collections import defaultdict from typing import Any, Callable, Dict, Iterator, List, Optional, Pattern, Tuple, Union from .errors import RouteNotFound, RouteParamMissing #: Alias for things that can be added to a router. RouteLike = Union["Route", "Include"] class Route: """An individual route. """ __slots__ = [ "template", "handler", "method", "name", ] def __init__(self, template: str, handler: Callable[..., Any], method: str = "GET", name: Optional[str] = None) -> None: self.template = template self.handler = handler self.method = method self.name = name or handler.__name__ class Include: """Groups of routes prefixed by a common path. """ __slots__ = [ "prefix", "routes", ] def __init__(self, prefix: str, routes: List[RouteLike]) -> None: self.prefix = prefix self.routes = routes class Router: """A collection of routes. """ __slots__ = [ "_routes_by_name", "_routes_by_method", "_route_res_by_method", ] def __init__(self, routes: Optional[List[RouteLike]] = None) -> None: self._routes_by_name: Dict[str, Route] = {} self._routes_by_method: Dict[str, List[Route]] = defaultdict(list) self._route_res_by_method: Dict[str, List[Pattern[str]]] = defaultdict(list) self.add_routes(routes or []) def add_route(self, route_like: RouteLike, prefix: str = "") -> None: """Add a Route to this instance. """ if isinstance(route_like, Include): self.add_routes(route_like.routes, prefix + route_like.prefix) elif isinstance(route_like, Route): if route_like.name in self._routes_by_name: raise ValueError(f"a route named {route_like.name} is already registered") route = Route( template=prefix + route_like.template, handler=route_like.handler, method=route_like.method, name=route_like.name, ) self._routes_by_name[route.name] = route self._routes_by_method[route.method].insert(0, route) self._route_res_by_method[route.method].insert(0, compile_route_template(route.template)) else: # pragma: no cover raise NotImplementedError(f"unhandled type {type(route_like)}") def add_routes(self, route_likes: List[RouteLike], prefix: str = "") -> None: """Add a set of routes to this instance. """ for route_like in route_likes: self.add_route(route_like, prefix) def match(self, method: str, path: str) -> Union[None, Tuple[Route, Dict[str, str]]]: """Look up the route matching the given method and path. Returns the route and any path params. """ routes = self._routes_by_method[method] route_res = self._route_res_by_method[method] for route, route_re in zip(routes, route_res): match = route_re.match(path) if match is not None: return route, match.groupdict() return None def reverse_uri(self, route_name: str, **params: str) -> str: """Build a URI from a Route. Raises: RouteNotFound: When the route doesn't exist. RouteParamMissing: When a required parameter was not provided. """ try: route = self._routes_by_name[route_name] except KeyError: raise RouteNotFound(route_name) uri = [] for kind, token in tokenize_route_template(route.template): if kind == "binding" or kind == "glob": try: uri.append(str(params[token])) except KeyError: raise RouteParamMissing(token) elif kind == "chunk": uri.append(token) return "".join(uri) def compile_route_template(template: str) -> Pattern[str]: """Convert a route template into a regular expression. """ re_template = "" for kind, token in tokenize_route_template(template): if kind == "binding": re_template += f"(?P<{token}>[^/]+)" elif kind == "glob": re_template += f"(?P<{token}>.+)" elif kind == "chunk": re_template += token.replace(".", r"\.") else: # pragma: no cover raise NotImplementedError(f"unhandled token kind {kind!r}") return re.compile(f"^{re_template}$") def tokenize_route_template(template: str) -> Iterator[Tuple[str, str]]: """Convert a route template into a stream of tokens. """ k, i = 0, 0 while i < len(template): if template[i] == "{": yield "chunk", template[k:i] k = i kind = "binding" if template[i:i + 2] == "{*": kind = "glob" i += 1 for j in range(i + 1, len(template)): if template[j] == "}": yield kind, template[i + 1:j] k = j + 1 i = j break else: raise SyntaxError(f"unmatched {{ in route template {template!r}") i += 1 if k != i: yield "chunk", template[k:i] PK!@tffmolten/testing/__init__.pyfrom .client import TestClient from .common import to_environ __all__ = ["TestClient", "to_environ"] PK! ɥmolten/testing/client.pyimport json from functools import partial from io import BytesIO from json import dumps as to_json from typing import Any, Callable, Dict, Optional, Union from urllib.parse import urlencode from ..app import BaseApp from ..http import HTTP_200 from ..http.headers import Headers, HeadersDict from ..http.query_params import ParamsDict, QueryParams from ..http.request import Request from ..http.response import Response from .common import to_environ HTTP_METHODS = {"delete", "head", "get", "patch", "post", "put"} class TestResponse: """A wrapper around Response objects that adds a few additional helper methods for testing. """ __slots__ = ["_response"] def __init__(self, response: Response) -> None: self._response = response @property def data(self) -> str: """Rewinds the output stream and returns all its data. """ self._response.stream.seek(0) return self._response.stream.read().decode("utf-8") @property def status_code(self) -> int: """Returns the HTTP status code as an integer. """ code, _, _ = self._response.status.partition(" ") return int(code) def json(self) -> Any: """Convert the response data to JSON. """ return json.loads(self.data) def __getattr__(self, name: str) -> Any: return getattr(self._response, name) class TestClient: """Test clients are used to simulate requests against an application instance. """ __slots__ = ["app"] def __init__(self, app: BaseApp) -> None: self.app = app def request( self, method: str, path: str, headers: Optional[Union[HeadersDict, Headers]] = None, params: Optional[Union[ParamsDict, QueryParams]] = None, data: Optional[Dict[str, str]] = None, json: Optional[Any] = None, auth: Optional[Callable[[Request], Request]] = None, ) -> TestResponse: """Simulate a request against the application. Raises: RuntimeError: If both 'data' and 'json' are provided. """ if data is not None and json is not None: raise RuntimeError("either 'data' or 'json' should be provided, not both") request = Request( method=method.upper(), path=path, headers=headers, params=params, ) if data is not None: request_content = urlencode(data).encode("utf-8") request.headers["content-type"] = "application/x-www-form-urlencoded" request.headers["content-length"] = f"{len(request_content)}" request.body_file = BytesIO(request_content) elif json is not None: request_content = to_json(json).encode("utf-8") request.headers["content-type"] = "application/json; charset=utf-8" request.headers["content-length"] = f"{len(request_content)}" request.body_file = BytesIO(request_content) if auth is not None: request = auth(request) response = Response(HTTP_200) def start_response(status, response_headers, exc_info=None): nonlocal response response.status = status response.headers = Headers(dict(response_headers)) chunks = self.app(to_environ(request), start_response) for chunk in chunks: response.stream.write(chunk) response.stream.seek(0) return TestResponse(response) def __getattr__(self, name: str) -> Any: if name in HTTP_METHODS: return partial(self.request, name) raise AttributeError(f"unknown attribute {name}") PK!5smolten/testing/common.pyfrom typing import Union from urllib.parse import urlencode from ..http import Headers, QueryParams, Request from ..typing import Environ def to_environ(ob: Union[Request, Headers, QueryParams]) -> Environ: """Convert request abstractions to WSGI environ dicts. """ if isinstance(ob, Request): return { "HTTP_HOST": ob.host, "PATH_INFO": ob.path, "REQUEST_METHOD": ob.method, "SERVER_PORT": ob.port, "wsgi.input": ob.body_file, "wsgi.url_scheme": ob.scheme, **to_environ(ob.params), **to_environ(ob.headers), } elif isinstance(ob, Headers): headers = {f"HTTP_{name.upper().replace('-', '_')}": value for name, value in ob} try: headers["CONTENT_TYPE"] = headers.pop("HTTP_CONTENT_TYPE") except KeyError: pass try: headers["CONTENT_LENGTH"] = headers.pop("HTTP_CONTENT_LENGTH") except KeyError: pass return headers elif isinstance(ob, QueryParams): return {"QUERY_STRING": urlencode(list(ob))} else: # pragma: no cover raise NotImplementedError(f"to_environ cannot handle type {type(ob)}") PK! Z_molten/typing.pyfrom typing import Any, BinaryIO, Callable, Dict, List, NewType, Tuple, Union #: An alias representing a WSGI environment dictionary. Environ = Dict[str, Any] #: An alias representing a WSGI start_response callback. StartResponse = Callable[[str, List[Tuple[str, str]], Any], None] #: The type of middleware functions. Middleware = Callable[[Callable[..., Any]], Callable[..., Any]] #: The request method. Method = NewType("Method", str) #: The request URI scheme. Scheme = NewType("Scheme", str) #: The request hostname. Host = NewType("Host", str) #: The request port. Port = NewType("Port", int) #: The request query string. QueryString = NewType("QueryString", str) #: A query string parameter. QueryParam = NewType("QueryParam", str) #: A header. Header = NewType("Header", str) #: A file-like object representing the request input data. RequestInput = NewType("RequestInput", BinaryIO) #: A bytestring representing the request input data. RequestBody = NewType("RequestBody", bytes) #: Parsed request data. RequestData = NewType("RequestData", dict) def extract_optional_annotation(annotation) -> Tuple[bool, Any]: """Returns a tuple denoting whether or not the annotation is an Optional type and the inner annotation. """ if is_optional_annotation(annotation): return True, annotation.__args__[0] return False, annotation def is_optional_annotation(annotation) -> bool: """Returns True if the given annotation represents an Optional type. """ try: return getattr(annotation, "__origin__", None) is Union and \ issubclass(annotation.__args__[1], type(None)) except TypeError: # pragma: no cover return False PK!HƣxSTmolten-0.0.0.dist-info/WHEEL A н#J;/"d&F]xzw>@Zpy3F ]n2H%_60{8&baPa>PK!Hgɜvmolten-0.0.0.dist-info/METADATAAK1sT0R!TQAb=O1vdVK73/DwXQ)9W 2YU _9gp~PK!HZmolten-0.0.0.dist-info/RECORDuɲy= di5oE@A&}#/y+)ͪ ~/i3ʶA_qjWݸ֋vrOes w0s3N_ToJ=\Z ԏ)xx&z;DCyCH$k}D8R}/z*&P|$V8ޠko3 x$f1IVJ{62۪zZ81lwR\kWQ! `3;M9Rr#Gw'J~'auwk*}> 0(T:CR־:M>{L>mmUcEcQj%5ōY_AJGxp2F /'9CgU<Ҟ!'Т,<{K$[]w/?7{P>W:j*Γx_w`z94Lҏf́5. y S:zN ' WӬDv[Nxb >ǔkЇCؼzd*GFY%4SYkmBc6`Z9"sH7uolYbE*7ق^Y[ifRp] +Y߸q ʘy {[hn}xUsf&L6sڱFi .XK[p$zv0^#X=x=o5bg)Npx/pEXPnNlvW ?ˑEY2+.YI6?1~nGS%H{p_4dGX,tL/۫;_b\9~\;gS6܂|(~9WYoދCV Ur]lS߁_6 a4$9H7 t%MkYx«(tf8a_% 'KD?PK!e| | molten/__init__.pyPK!|gPP molten/app.pyPK!r v'molten/common.pyPK! CC &molten/components.pyPK! 9molten/contrib/settings.pyPK!?@molten/contrib/templates.pyPK!n1">>Emolten/dependency_injection.pyPK!"Ymolten/errors.pyPK!NN;^molten/http/__init__.pyPK!bmolten/http/cookies.pyPK!D,dmolten/http/headers.pyPK!O#%smolten/http/query_params.pyPK!Tvxmolten/http/request.pyPK!9X~molten/http/response.pyPK!d molten/http/status_codes.pyPK!2llmolten/middleware.pyPK!$molten/parsers.pyPK!о//0molten/py.typedPK!amolten/renderers.pyPK!ɢCmolten/router.pyPK!@tff̲molten/testing/__init__.pyPK! ɥjmolten/testing/client.pyPK!5s=molten/testing/common.pyPK! Z_Smolten/typing.pyPK!HƣxST)molten-0.0.0.dist-info/WHEELPK!Hgɜvmolten-0.0.0.dist-info/METADATAPK!HZmolten-0.0.0.dist-info/RECORDPK/