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@8: terom@8: def get_script_dir (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@10: def get_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@10: 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@7: terom@7: class Response (object) : terom@7: """ terom@7: HTTP Response with headers and data terom@7: """ terom@7: terom@7: def __init__ (self, data, content_type='text/html', status='200 OK', charset='utf8') : terom@7: """ terom@7: Create the response. The Content-type header is built from the given values. The given \a data must be terom@7: either a str (which is sent plain), an unicode object (which is encoded with the relevant charset), or terom@7: None, whereupon an empty response body is sent. The content_type argument can also be forced to None to terom@7: not send a Content-type header (e.g. for redirects) 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@7: return self.data.encode(self.charset) terom@7: 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: