terom@149: """ terom@149: WSGI HTTP utility code terom@149: """ terom@149: terom@149: # for utility functions terom@149: import cgi terom@149: terom@149: # for header handling terom@149: import wsgiref.headers terom@149: terom@150: # for path handling terom@150: import os.path terom@150: terom@149: class Request (object) : terom@149: """ terom@149: HTTP Request with associated metadata terom@149: """ terom@149: terom@149: def __init__ (self, env) : terom@149: """ terom@149: Parse env data terom@149: """ terom@149: terom@149: # store env terom@149: self.env = env terom@149: terom@149: # get the querystring terom@149: self.arg_str = env.get('QUERY_STRING', '') terom@149: terom@149: # parse query args terom@149: self.arg_dict = cgi.parse_qs(self.arg_str, True) terom@150: terom@150: def get_script_dir (self) : terom@150: """ terom@150: Returns the URL path to the requested script's directory with no trailing slash, i.e. terom@150: terom@150: / -> terom@150: /foo.cgi -> terom@150: /foo/bar.cgi -> /foo terom@150: """ terom@150: terom@156: return os.path.dirname(self.env['SCRIPT_NAME']).rstrip('/') terom@150: terom@152: def get_page_prefix (self) : terom@152: """ terom@152: Returns the URL path root for page URLs, based on REQUEST_URI with PATH_INFO removed terom@152: terom@152: / -> terom@152: /foo.cgi -> /foo.cgi terom@152: /foo.cgi/index -> /foo.cgi terom@152: /foo.cgi/quux/bar -> /foo.cgi terom@152: /quux/foo.cgi/bar -> /quux/foo.cgi terom@152: /bar -> terom@152: """ terom@152: terom@152: # XXX: request uri path without the query string terom@152: request_path = self.env.get('REQUEST_URI', '').split('?', 1)[0].rstrip('/') terom@152: terom@152: # path info terom@152: page_name = self.get_page_name() terom@152: terom@152: # special-case for empty page_name terom@152: if not page_name : terom@152: return request_path terom@152: terom@152: # sanity-check terom@152: assert request_path.endswith(page_name) terom@152: terom@152: # trim terom@152: return request_path[:-len(page_name)].rstrip('/') terom@152: terom@150: def get_page_name (self) : terom@150: """ terom@150: Returns the requested page path with no leading slash, i.e. terom@150: terom@150: /foo.cgi -> terom@150: /foo.cgi/ -> terom@150: /foo.cgi/bar -> bar terom@150: /foo.cgi/quux/ -> quux/ terom@150: """ terom@150: terom@151: # the raw PATH_INFO terom@151: path_info = self.env.get('PATH_INFO') terom@151: terom@151: # avoid nasty '.' paths terom@151: if path_info : terom@151: return os.path.normpath(path_info).lstrip('/') terom@151: terom@151: else : terom@151: return '' terom@149: terom@149: class Response (object) : terom@149: """ terom@149: HTTP Response with headers and data terom@149: """ terom@149: terom@149: def __init__ (self, data, content_type='text/html', status='200 OK', charset='utf8') : terom@149: """ terom@149: Create the response. The Content-type header is built from the given values. The given \a data must be terom@149: either a str (which is sent plain), an unicode object (which is encoded with the relevant charset), or terom@149: None, whereupon an empty response body is sent. The content_type argument can also be forced to None to terom@149: not send a Content-type header (e.g. for redirects) terom@149: """ terom@149: terom@149: # store info terom@149: self.status = status terom@149: self.data = data terom@149: self.charset = charset terom@149: terom@149: # headers terom@149: self.headers = wsgiref.headers.Headers([]) terom@149: terom@149: # add Content-type header? terom@149: if content_type : terom@149: self.add_header('Content-type', content_type, charset=charset) terom@149: terom@149: def add_header (self, name, value, **params) : terom@149: """ terom@149: Add response header with the given name/value, plus option params terom@149: terom@149: XXX: uses the wsgiref.headers code, not sure how that behaves re multiple headers with the same name, etc terom@149: """ terom@149: terom@149: self.headers.add_header(name, value, **params) terom@149: terom@149: def get_status (self) : terom@149: """ terom@149: Returns response status string (XXX Foo) terom@149: """ terom@149: terom@149: return self.status terom@149: terom@149: def get_headers (self) : terom@149: """ terom@149: Returns the list of header (name, value) pairs terom@149: """ terom@149: terom@149: return self.headers.items() terom@149: terom@149: def get_data (self) : terom@149: """ terom@149: Returns the response data - as an encoded string terom@149: """ terom@149: terom@149: if self.data : terom@149: return self.data.encode(self.charset) terom@149: terom@149: else : terom@149: return '' terom@149: terom@149: class ErrorResponse (Response) : terom@149: """ terom@149: A response with an error code / message terom@149: """ terom@149: terom@150: def __init__ (self, status, message, details=None) : terom@149: """ terom@149: Build a plain error message response with the given status/message terom@150: terom@150: @param status HTTP status code terom@150: @param message short message to describe errors terom@150: @param details optional details, plaintext terom@149: """ terom@149: terom@149: data = """\ terom@149:
%(message)s
terom@150: %(details)s terom@149: terom@149: """ % dict( terom@149: title = status, terom@150: message = message, terom@150: details = '%s' % details if details else '' terom@149: ) terom@149: terom@149: super(ErrorResponse, self).__init__(data, status=status) terom@149: terom@149: class ResponseError (Exception) : terom@149: """ terom@149: An exception that results in a specfic 4xx ErrorResponse message to the client terom@149: """ terom@149: terom@150: def __init__ (self, message, status='400 Bad Request', details=None) : terom@149: self.status = status terom@149: self.message = message terom@150: self.details = details terom@149: terom@149: super(ResponseError, self).__init__(message) terom@149: terom@149: def get_response (self) : terom@150: return ErrorResponse(self.status, self.message, self.details) terom@149: terom@149: class Redirect (Response) : terom@149: """ terom@149: Redirect response terom@149: """ terom@149: terom@149: def __init__ (self, url) : terom@149: """ terom@149: Redirect to given *absolute* URL terom@149: """ terom@149: terom@149: # no content-type or data terom@149: super(Redirect, self).__init__(None, content_type=None, status='302 Found') terom@149: terom@149: # add Location: header terom@149: self.add_header("Location", url) terom@149: terom@149: