Source code for watson.http.messages

# -*- coding: utf-8 -*-
import json
from urllib.parse import parse_qsl
from wsgiref.util import request_uri
from watson.common.datastructures import ImmutableMultiDict
from watson.common.decorators import cached_property
from watson.common.imports import get_qualified_name, load_definition_from_string
from watson.http import STATUS_CODES, REQUEST_METHODS
from watson.http.cookies import CookieDict, cookies_from_environ
from watson.http.headers import HeaderCollection, ServerCollection, fix_http_headers
from watson.http.uri import Url
from watson.http.wsgi import copy_wsgi_input, get_form_vars, WSGI_BODY
from watson.http.sessions import COOKIE_KEY

[docs]class MessageMixin(object): """Base mixin for all Http Message objects. """ _version = None @property def version(self): return self._version or '1.1' @version.setter def version(self, version): self._version = version
[docs]class Request(MessageMixin): """ Provides a simple and usable interface for dealing with Http Requests. Requests are designed to be immutable and not altered after they are created, as such you should only set get/post/cookie etc attributes via the __init__. By default the session storage method is MemoryStorage which will store session in ram. See: - - Example: .. code-block:: python request = Request.from_environ(environ) print(request.method) print('my_post_var')) request = Request.from_dicts(server={'HTTP_METHOD': 'GET'}, get={'get_var': 'somevalue'}) print(request.method) # get print(request.get('get_var')) # somevalue """ _environ = None _session = None
[docs] def __init__(self, environ): fix_http_headers(environ) self._environ = environ
@property def environ(self): return self._environ @property def encoding(self): return self.headers.get_option('Content-Type', 'charset', 'utf-8') @property def raw_body(self): return self.environ[WSGI_BODY] @property def body(self): copy_wsgi_input(self.environ) return self.raw_body.decode(self.encoding) @body.setter def body(self, body): """Set the body of the Request. Args: body (string): The body of the request. """ self.environ[WSGI_BODY] = body.encode(self.encoding) @property
[docs] def json_body(self): """Returns the body encoded as JSON. """ return json.loads(self.body)
@cached_property def method(self): """The method associated with the request. If the existing method is a POST, also check for HTTP_REQUEST_METHOD in the post vars to enable custom (but valid) request methods. Returns: A string representation of the Http Request method """ method = self.environ.get('REQUEST_METHOD', 'GET').upper() if method == 'POST': post_method ='HTTP_REQUEST_METHOD', method).upper() if post_method in REQUEST_METHODS: method = post_method return method @cached_property def url(self): """Generates a watson.http.uri.Url object based on Request.server variables. Example: .. code-block:: python request = ... print(request.url.path) # / Returns: A watson.http.uri.Url object """ return Url(request_uri(self.environ)) @cached_property def get(self): """A dict of all GET variables associated with the request. Returns: A dict of GET variables """ qs = self.environ.get('QUERY_STRING', '') if qs: return ImmutableMultiDict(parse_qsl(qs, keep_blank_values=True)) return ImmutableMultiDict() @cached_property def _get_post_files_from_environ(self): if self.environ.get('REQUEST_METHOD') not in ('POST', 'PUT', 'PATCH'): post = files = ImmutableMultiDict() else: copy_wsgi_input(self.environ) post, files = get_form_vars(self.environ, ImmutableMultiDict) return post, files @property
[docs] def post(self): """A dict of all POST variables associated with the request. Returns: A dict of POST variables """ post, files = self._get_post_files_from_environ return post
@cached_property def files(self): """A dict of all files that have been uploaded as part of a enctype="multipart/form-data" request. Example: .. code-block:: python request = ... request.files['uploaded_file'] # FieldStorage object Returns: A dict of FieldStorage objects """ post, files = self._get_post_files_from_environ return files @cached_property def headers(self): return HeaderCollection.from_environ(self.environ) @cached_property def server(self): return ServerCollection.from_environ(self.environ) @cached_property def cookies(self): """A dict of all cookies from the request. Example: .. code-block:: python request = ... request.cookies.get('test') # value of cookie named 'test' Returns: A watson.http.cookies.CookieDict object """ return cookies_from_environ(self.environ) @property def session(self): session_class = self.environ.get('watson.session.class', None) if session_class and not self._session: storage = load_definition_from_string(session_class) options = self.environ['watson.session.options'] http_cookie = self.environ.get('HTTP_COOKIE', None) if (http_cookie and '{0}='.format(COOKIE_KEY) in http_cookie): session_cookie = self.cookies[COOKIE_KEY] if session_cookie: options['id'] = session_cookie.value self._session = storage(**options) return self._session # Initializers @classmethod
[docs] def from_dicts(cls, get, post, server, headers, body): """@todo Implement. """ environ = {} request = cls(environ) return request
@classmethod def from_environ(cls, environ, session_class=None, session_options=None): environ['watson.session.class'] = session_class environ['watson.session.options'] = session_options or {} request = cls(environ) return request # Convenience methods
[docs] def is_method(self, *methods): """ Determine whether or not a request was made via a specific method. Example: .. code-block:: python request = ... # request made via GET request.is_method('get') # True Args: method (string|list|tuple): the method or list of methods to check Returns: Boolean """ methods = (methods,) if isinstance(methods, str) else methods return self.method in [m.upper() for m in methods]
[docs] def is_xml_http_request(self): """ Determine whether or not a request has originated via an XmlHttpRequest, assuming the relevant header has been set by the request. Returns: Boolean """ return (self.headers.get( 'X-Requested-With', '').lower() == 'xmlhttprequest')
[docs] def is_secure(self): """ Determine whether or not the request was made via Https. Returns: Boolean """ if 'Https' in self.headers: return self.headers['Https'].lower() == 'https' return self.url.scheme.lower() == 'https'
[docs] def host(self): """Determine the real host of a request. Returns: X_FORWARDED_FOR header variable if set, otherwise a watson.http.uri.Url hostname attribute. """ return ( self.url.hostname if 'X-Forwarded-For' not in self.headers else self.headers.get('X-Forwarded-For') )
def __str__(self): return '{0} {1} HTTP/{2}\r\n{3}\r\n\r\n{4}'.format(self.method, self.url, self.version, self.headers, self.body) def __repr__(self): return '<{0} method:{1} url:{2}>'.format(get_qualified_name(self), self.method, self.url)
[docs]class Response(MessageMixin): """Provides a simple and usable interface for dealing with Http Responses. See: - Example: .. code-block:: python def app(environ, start_response): response = Response(200, body='<h1>Hello World!</h1>') response.headers.add('Content-Type', 'text/html', charset='utf-8') response.cookies.add('something', 'test') start_response(*response.start()) return [response()] """ _status_code = None _cookies = None _body = None _prepared = False
[docs] def __init__(self, status_code=None, headers=None, body=None, version=None): """ Args: status_code (int): The status code for the Response headers (watson.http.headers.HeaderCollection): Valid response headers. body (string): The content for the response version (string): The Http version for the response """ self.status_code = status_code self._headers = headers or HeaderCollection() if version: self.version = version if body: self.body = body
@property def headers(self): return self._headers @headers.setter def headers(self, headers): if not isinstance(headers, HeaderCollection): headers = HeaderCollection(headers) self._headers = headers @property def raw_body(self): if self._body: return self._body return b'' @property def body(self): """Returns the decoded body based on the response encoding type. """ return self.raw_body.decode(self.encoding) @body.setter
[docs] def body(self, body): """Set the body of the Request. Args: body (string): The body of the request. """ self._body = body.encode(self.encoding)
@property def status_code(self): """The status code for the Response. """ return self._status_code or 200 @status_code.setter
[docs] def status_code(self, code): """ Args: Code: an int representing the status code for the Response """ self._status_code = code
[docs] def status_line(self): """The formatted status line including the status code and message. """ return ( '{0} {1}'.format( self.status_code, STATUS_CODES.get(self.status_code)))
@property def cookies(self): """Returns the cookies associated with the Response. """ if not self._cookies: self.cookies = CookieDict() return self._cookies @cookies.setter
[docs] def cookies(self, cookies): """Sets the cookies associated with the Response. Args: cookies (CookieDict): The cookies to add to the response. """ self._cookies = cookies
[docs] def encoding(self): """Retrieve the encoding for the response from the headers, defaults to UTF-8. """ return self.headers.get_option('Content-Type', 'charset', 'utf-8')
[docs] def start(self): """Return the status_line and headers of the response for use in a WSGI application. Returns: The status line and headers of the response. """ self._prepare() return self.status_line, self.headers()
[docs] def raw(self): """Return the raw encoded output for the response. """ return str(self).encode(self.encoding)
def __str__(self): self._prepare() return 'HTTP/{0} {1}\r\n{2}\r\n\r\n{3}'.format(self.version, self.status_line, self.headers, self.body) def _prepare(self): if not self._prepared: if self._cookies: for cookie, morsel in self.cookies.items(): self.headers.add('Set-Cookie', str(morsel)) self._prepared = True def __call__(self, start_response): """Execute the start_response method and return the response body. Args: start_response (callable): The start_response function from a WSGI callable. """ start_response(*self.start()) return [self.raw_body]