terom@7: """
terom@7: WSGI HTTP utility code
terom@7: """
terom@7:
terom@7: # for utility functions
terom@7: import cgi
terom@7:
terom@7: # for header handling
terom@7: import wsgiref.headers
terom@7:
terom@8: # for path handling
terom@8: import os.path
terom@8:
terom@7: class Request (object) :
terom@7: """
terom@7: HTTP Request with associated metadata
terom@7: """
terom@7:
terom@7: def __init__ (self, env) :
terom@7: """
terom@7: Parse env data
terom@7: """
terom@7:
terom@7: # store env
terom@7: self.env = env
terom@7:
terom@7: # get the querystring
terom@7: self.arg_str = env.get('QUERY_STRING', '')
terom@7:
terom@7: # parse query args
terom@7: self.arg_dict = cgi.parse_qs(self.arg_str, True)
terom@55:
terom@55: # load post data?
terom@55: if self.is_post() :
terom@55: # content-type of post data
terom@55: content_type = self.env.get('CONTENT_TYPE', 'application/x-www-form-urlencoded')
terom@55:
terom@55: # valid content-type?
terom@55: # XXX: how to handle errors?
terom@55: assert any(content_type.startswith(x) for x in (
terom@55: 'application/x-www-form-urlencoded',
terom@55: 'multipart/form-data'
terom@55: ))
terom@55:
terom@55: # input stream
terom@55: input = self.env['wsgi.input']
terom@55:
terom@55: # use cgi.FieldStorage to parse this
terom@55: self.post_data = cgi.FieldStorage(fp=input, environ=self.env, keep_blank_values=1)
terom@55:
terom@55: else :
terom@55: # no post data
terom@55: self.post_data = None
terom@42:
terom@42: @property
terom@42: def site_host (self) :
terom@42: """
terom@42: Returns the site's hostname (DNS name)
terom@42: """
terom@42:
terom@42: return self.env['HTTP_HOST']
terom@42:
terom@42: @property
terom@42: def site_root (self) :
terom@8: """
terom@8: Returns the URL path to the requested script's directory with no trailing slash, i.e.
terom@8:
terom@8: / ->
terom@8: /foo.cgi ->
terom@8: /foo/bar.cgi -> /foo
terom@8: """
terom@8:
terom@14: return os.path.dirname(self.env['SCRIPT_NAME']).rstrip('/')
terom@8:
terom@42: @property
terom@42: def page_prefix (self) :
terom@10: """
terom@10: Returns the URL path root for page URLs, based on REQUEST_URI with PATH_INFO removed
terom@10:
terom@10: / ->
terom@10: /foo.cgi -> /foo.cgi
terom@10: /foo.cgi/index -> /foo.cgi
terom@10: /foo.cgi/quux/bar -> /foo.cgi
terom@10: /quux/foo.cgi/bar -> /quux/foo.cgi
terom@10: /bar ->
terom@10: """
terom@10:
terom@10: # XXX: request uri path without the query string
terom@10: request_path = self.env.get('REQUEST_URI', '').split('?', 1)[0].rstrip('/')
terom@10:
terom@10: # path info
terom@10: page_name = self.get_page_name()
terom@10:
terom@10: # special-case for empty page_name
terom@10: if not page_name :
terom@10: return request_path
terom@10:
terom@10: # sanity-check
terom@10: assert request_path.endswith(page_name)
terom@10:
terom@10: # trim
terom@10: return request_path[:-len(page_name)].rstrip('/')
terom@42:
terom@8: def get_page_name (self) :
terom@8: """
terom@8: Returns the requested page path with no leading slash, i.e.
terom@8:
terom@8: /foo.cgi ->
terom@8: /foo.cgi/ ->
terom@8: /foo.cgi/bar -> bar
terom@8: /foo.cgi/quux/ -> quux/
terom@8: """
terom@8:
terom@9: # the raw PATH_INFO
terom@9: path_info = self.env.get('PATH_INFO')
terom@9:
terom@9: # avoid nasty '.' paths
terom@9: if path_info :
terom@9: return os.path.normpath(path_info).lstrip('/')
terom@9:
terom@9: else :
terom@9: return ''
terom@42:
terom@49: def get_arg (self, name, default=None) :
terom@49: """
terom@49: Get a single value for an argument with the given key, or the default if missing
terom@49: """
terom@49:
terom@49: if name in self.arg_dict :
terom@49: return self.arg_dict[name][0]
terom@49:
terom@49: else :
terom@49: return default
terom@49:
terom@42: def get_args (self) :
terom@42: """
terom@42: Iterate over all available (key, value) pairs from the query string
terom@42: """
terom@42:
terom@42: return cgi.parse_qsl(self.arg_str)
terom@55:
terom@55: def is_post (self) :
terom@55: """
terom@55: Is this a POST request?
terom@55: """
terom@55:
terom@55: # just check REQUEST_METHOD
terom@55: return (self.env['REQUEST_METHOD'].upper() == 'POST')
terom@55:
terom@55: def get_post (self, name) :
terom@55: """
terom@55: Get the value of the given POST field
terom@55: """
terom@55:
terom@55: # sanity-check
terom@55: assert self.post_data
terom@55:
terom@55: # return the FieldStorage value
terom@55: return self.post_data[name].value
terom@7:
terom@7: class Response (object) :
terom@7: """
terom@7: HTTP Response with headers and data
terom@7: """
terom@7:
terom@25: def __init__ (self, data, content_type='text/html', status='200 OK', charset='UTF-8') :
terom@7: """
terom@57: Create the response. The Content-type header is built from the given values.
terom@57:
terom@57: The given data must be a string-like object, which will be encoded with the given charset, or None,
terom@57: whereupon an empty response body will be sent.
terom@57:
terom@57: The content_type argument can also be forced to None to not send a Content-type header (e.g. for
terom@57: redirects).
terom@57:
terom@57: The charset can be given as None to not encode the output (for binary data).
terom@7: """
terom@7:
terom@7: # store info
terom@7: self.status = status
terom@7: self.data = data
terom@7: self.charset = charset
terom@7:
terom@7: # headers
terom@7: self.headers = wsgiref.headers.Headers([])
terom@7:
terom@7: # add Content-type header?
terom@7: if content_type :
terom@7: self.add_header('Content-type', content_type, charset=charset)
terom@7:
terom@7: def add_header (self, name, value, **params) :
terom@7: """
terom@7: Add response header with the given name/value, plus option params
terom@7:
terom@7: XXX: uses the wsgiref.headers code, not sure how that behaves re multiple headers with the same name, etc
terom@7: """
terom@7:
terom@7: self.headers.add_header(name, value, **params)
terom@7:
terom@7: def get_status (self) :
terom@7: """
terom@7: Returns response status string (XXX Foo)
terom@7: """
terom@7:
terom@7: return self.status
terom@7:
terom@7: def get_headers (self) :
terom@7: """
terom@7: Returns the list of header (name, value) pairs
terom@7: """
terom@7:
terom@7: return self.headers.items()
terom@7:
terom@7: def get_data (self) :
terom@7: """
terom@7: Returns the response data - as an encoded string
terom@7: """
terom@7:
terom@7: if self.data :
terom@57: if self.charset :
terom@57: return self.data.encode(self.charset)
terom@7:
terom@57: else :
terom@57: return self.data
terom@57:
terom@7: else :
terom@7: return ''
terom@7:
terom@7: class ErrorResponse (Response) :
terom@7: """
terom@7: A response with an error code / message
terom@7: """
terom@7:
terom@8: def __init__ (self, status, message, details=None) :
terom@7: """
terom@7: Build a plain error message response with the given status/message
terom@8:
terom@8: @param status HTTP status code
terom@8: @param message short message to describe errors
terom@8: @param details optional details, plaintext
terom@7: """
terom@7:
terom@7: data = """\
terom@7: %(title)s
terom@7: %(title)s
terom@7: %(message)s
terom@8: %(details)s
terom@7:
terom@7: """ % dict(
terom@7: title = status,
terom@8: message = message,
terom@8: details = '%s
' % details if details else ''
terom@7: )
terom@7:
terom@7: super(ErrorResponse, self).__init__(data, status=status)
terom@7:
terom@7: class ResponseError (Exception) :
terom@7: """
terom@7: An exception that results in a specfic 4xx ErrorResponse message to the client
terom@7: """
terom@7:
terom@8: def __init__ (self, message, status='400 Bad Request', details=None) :
terom@7: self.status = status
terom@7: self.message = message
terom@8: self.details = details
terom@7:
terom@7: super(ResponseError, self).__init__(message)
terom@7:
terom@7: def get_response (self) :
terom@8: return ErrorResponse(self.status, self.message, self.details)
terom@7:
terom@7: class Redirect (Response) :
terom@7: """
terom@7: Redirect response
terom@7: """
terom@7:
terom@7: def __init__ (self, url) :
terom@7: """
terom@7: Redirect to given *absolute* URL
terom@7: """
terom@7:
terom@7: # no content-type or data
terom@7: super(Redirect, self).__init__(None, content_type=None, status='302 Found')
terom@7:
terom@7: # add Location: header
terom@7: self.add_header("Location", url)
terom@7:
terom@7: