我们从Python开源项目中，提取了以下32个代码示例，用于说明如何使用cherrypy.response()。
def clean_headers(status): """Remove any headers which should not apply to an error response.""" import cherrypy response = cherrypy.serving.response # Remove headers which applied to the original content, # but do not apply to the error page. respheaders = response.headers for key in ['Accept-Ranges', 'Age', 'ETag', 'Location', 'Retry-After', 'Vary', 'Content-Encoding', 'Content-Length', 'Expires', 'Content-Location', 'Content-MD5', 'Last-Modified']: if key in respheaders: del respheaders[key] if status != 416: # A server sending a response with status code 416 (Requested # range not satisfiable) SHOULD include a Content-Range field # with a byte-range-resp-spec of "*". The instance-length # specifies the current length of the selected resource. # A response with status code 206 (Partial Content) MUST NOT # include a Content-Range field with a byte-range- resp-spec of "*". if 'Content-Range' in respheaders: del respheaders['Content-Range']
def _populate_known_types(self): b = [x for x in vars(builtins).values() if type(x) is type(str)] def traverse(obj, namespace): for name in dir(obj): # Hack for 3.2's warning about body_params if name == 'body_params': continue vtype = type(getattr(obj, name, None)) if vtype in b: self.known_config_types[namespace + '.' + name] = vtype traverse(cherrypy.request, 'request') traverse(cherrypy.response, 'response') traverse(cherrypy.server, 'server') traverse(cherrypy.engine, 'engine') traverse(cherrypy.log, 'log')
def run_standard_benchmarks(): print("") print("Client Thread Report (1000 requests, 14 byte response body, " "%s server threads):" % cherrypy.server.thread_pool) print_report(thread_report()) print("") print("Client Thread Report (1000 requests, 14 bytes via staticdir, " "%s server threads):" % cherrypy.server.thread_pool) print_report(thread_report("%s/static/index.html" % SCRIPT_NAME)) print("") print("Size Report (1000 requests, 50 client threads, " "%s server threads):" % cherrypy.server.thread_pool) print_report(size_report()) # modpython and other WSGI #
def clean_headers(status): """Remove any headers which should not apply to an error response.""" import cherrypy response = cherrypy.serving.response # Remove headers which applied to the original content, # but do not apply to the error page. respheaders = response.headers for key in ["Accept-Ranges", "Age", "ETag", "Location", "Retry-After", "Vary", "Content-Encoding", "Content-Length", "Expires", "Content-Location", "Content-MD5", "Last-Modified"]: if key in respheaders: del respheaders[key] if status != 416: # A server sending a response with status code 416 (Requested # range not satisfiable) SHOULD include a Content-Range field # with a byte-range-resp-spec of "*". The instance-length # specifies the current length of the selected resource. # A response with status code 206 (Partial Content) MUST NOT # include a Content-Range field with a byte-range- resp-spec of "*". if "Content-Range" in respheaders: del respheaders["Content-Range"]
def _be_ie_unfriendly(status): import cherrypy response = cherrypy.serving.response # For some statuses, Internet Explorer 5+ shows "friendly error # messages" instead of our response.body if the body is smaller # than a given size. Fix this by returning a body over that size # (by adding whitespace). # See http://support.microsoft.com/kb/q218155/ s = _ie_friendly_error_sizes.get(status, 0) if s: s += 1 # Since we are issuing an HTTP error status, we assume that # the entity is short, and we should just collapse it. content = response.collapse_body() l = len(content) if l and l < s: # IN ADDITION: the response must be written to IE # in one chunk or it will still get replaced! Bah. content = content + (ntob(" ") * (s - l)) response.body = content response.headers['Content-Length'] = str(len(content))
def _populate_known_types(self): b = [x for x in vars(builtins).values() if type(x) is type(str)] def traverse(obj, namespace): for name in dir(obj): # Hack for 3.2's warning about body_params if name == 'body_params': continue vtype = type(getattr(obj, name, None)) if vtype in b: self.known_config_types[namespace + "." + name] = vtype traverse(cherrypy.request, "request") traverse(cherrypy.response, "response") traverse(cherrypy.server, "server") traverse(cherrypy.engine, "engine") traverse(cherrypy.log, "log")
def jsonify_error(status, message, traceback, version): response = cherrypy.response response.headers['Content-Type'] = 'application/json' return json.dumps({'status': int(status.split()[0]), 'message': message})
def cleanup_headers(self): """ Some clients aren't that smart when it comes to headers lookup. """ response = cherrypy.response if not response.header_list: return headers = response.header_list[:] for (k, v) in headers: if k[:7] == 'Sec-Web': response.header_list.remove((k, v)) response.header_list.append((k.replace('Sec-Websocket', 'Sec-WebSocket'), v))
def _set_response(body): # The XML-RPC spec (http://www.xmlrpc.com/spec) says: # "Unless there's a lower-level error, always return 200 OK." # Since Python's xmlrpclib interprets a non-200 response # as a "Protocol Error", we'll just return 200 every time. response = cherrypy.response response.status = '200 OK' response.body = ntob(body, 'utf-8') response.headers['Content-Type'] = 'text/xml' response.headers['Content-Length'] = len(body)
def __init__(self, urls, status=None, encoding=None): import cherrypy request = cherrypy.serving.request if isinstance(urls, text_or_bytes): urls = [urls] abs_urls = [] for url in urls: url = tonative(url, encoding or self.encoding) # Note that urljoin will "do the right thing" whether url is: # 1. a complete URL with host (e.g. "http://www.example.com/test") # 2. a URL relative to root (e.g. "/dummy") # 3. a URL relative to the current path # Note that any query string in cherrypy.request is discarded. url = _urljoin(cherrypy.url(), url) abs_urls.append(url) self.urls = abs_urls # RFC 2616 indicates a 301 response code fits our goal; however, # browser support for 301 is quite messy. Do 302/303 instead. See # http://www.alanflavell.org.uk/www/post-redirect.html if status is None: if request.protocol >= (1, 1): status = 303 else: status = 302 else: status = int(status) if status < 300 or status > 399: raise ValueError('status must be between 300 and 399.') self.status = status CherryPyException.__init__(self, abs_urls, status)
def set_response(self): """Modify cherrypy.response status, headers, and body to represent self. CherryPy uses this internally, but you can also use it to create an HTTPError object and set its output without *raising* the exception. """ import cherrypy response = cherrypy.serving.response clean_headers(self.code) # In all cases, finalize will be called after this method, # so don't bother cleaning up response values here. response.status = self.status tb = None if cherrypy.serving.request.show_tracebacks: tb = format_exc() response.headers.pop('Content-Length', None) content = self.get_error_page(self.status, traceback=tb, message=self._message) response.body = content _be_ie_unfriendly(self.code)
def run(self, method, path, query_string, protocol, headers, rfile): cherrypy.response.status = "200 OK" cherrypy.response.header_list = [("Content-Type", 'text/html'), ("Server", "Null CherryPy"), ("Date", httputil.HTTPDate()), ("Content-Length", "0"), ] cherrypy.response.body = [""] return cherrypy.response
def __init__(self, urls, status=None, encoding=None): import cherrypy request = cherrypy.serving.request if isinstance(urls, text_or_bytes): urls = [urls] abs_urls = [] for url in urls: url = tonative(url, encoding or self.encoding) # Note that urljoin will "do the right thing" whether url is: # 1. a complete URL with host (e.g. "http://www.example.com/test") # 2. a URL relative to root (e.g. "/dummy") # 3. a URL relative to the current path # Note that any query string in cherrypy.request is discarded. url = _urljoin(cherrypy.url(), url) abs_urls.append(url) self.urls = abs_urls # RFC 2616 indicates a 301 response code fits our goal; however, # browser support for 301 is quite messy. Do 302/303 instead. See # http://www.alanflavell.org.uk/www/post-redirect.html if status is None: if request.protocol >= (1, 1): status = 303 else: status = 302 else: status = int(status) if status < 300 or status > 399: raise ValueError("status must be between 300 and 399.") self.status = status CherryPyException.__init__(self, abs_urls, status)
def default_error_page(**kwargs): """This function is registered as the default error page for CherryPy errors. This sets the response headers to be uncacheable, and then returns a HTTP response that is identical to the default CherryPy message format.""" response = cherrypy.response for key in ('Cache-Control', 'Pragma'): if key in response.headers: del response.headers[key] return _HTTPErrorTemplate % kwargs
def file_2(self, *tokens): """Outputs the contents of the file, named by the SHA hash name in the request path, directly to the client.""" method = cherrypy.request.method if method == "HEAD": try: fhash = tokens[0] except IndexError: fhash = None try: fpath = self.repo.file(fhash, pub=self._get_req_pub()) except srepo.RepositoryFileNotFoundError as e: raise cherrypy.HTTPError(http_client.NOT_FOUND, str(e)) except srepo.RepositoryError as e: # Treat any remaining repository error as a 404, # but log the error and include the real failure # information. cherrypy.log("Request failed: {0}".format( str(e))) raise cherrypy.HTTPError(http_client.NOT_FOUND, str(e)) csize, chashes = misc.compute_compressed_attrs(fhash, file_path=fpath) response = cherrypy.response for i, attr in enumerate(chashes): response.headers["X-Ipkg-Attr-{0}".format(i)] = \ "{0}={1}".format(attr, chashes[attr]) # set expiration of response to one day self.__set_response_expires("file", 86400, 86400) return serve_file(fpath, "application/data") return self.file_1(*tokens)
def open_0(self, *tokens): """Starts a transaction for the package name specified in the request path. Returns no output.""" request = cherrypy.request response = cherrypy.response client_release = request.headers.get("Client-Release", None) try: pfmri = tokens[0] except IndexError: pfmri = None # XXX Authentication will be handled by virtue of possessing a # signed certificate (or a more elaborate system). if not pfmri: raise cherrypy.HTTPError(http_client.BAD_REQUEST, _("A valid package FMRI must be specified.")) try: pfmri = fmri.PkgFmri(pfmri, client_release) trans_id = self.repo.open(client_release, pfmri) except (fmri.FmriError, srepo.RepositoryError) as e: # Assume a bad request was made. A 404 can't be # returned here as misc.versioned_urlopen will interpret # that to mean that the server doesn't support this # operation. cherrypy.log("Request failed: {0}".format(e)) raise cherrypy.HTTPError(http_client.BAD_REQUEST, str(e)) if pfmri.publisher and not self._get_req_pub(): self.__map_pub_ops(pfmri.publisher) # Set response headers before returning. response.headers["Content-type"] = "text/plain; charset=utf-8" response.headers["Transaction-ID"] = trans_id
def append_0(self, *tokens): """Starts an append transaction for the package name specified in the request path. Returns no output.""" request = cherrypy.request response = cherrypy.response client_release = request.headers.get("Client-Release", None) try: pfmri = tokens[0] except IndexError: pfmri = None # XXX Authentication will be handled by virtue of possessing a # signed certificate (or a more elaborate system). if not pfmri: raise cherrypy.HTTPError(http_client.BAD_REQUEST, _("A valid package FMRI must be specified.")) try: pfmri = fmri.PkgFmri(pfmri, client_release) trans_id = self.repo.append(client_release, pfmri) except (fmri.FmriError, srepo.RepositoryError) as e: # Assume a bad request was made. A 404 can't be # returned here as misc.versioned_urlopen will interpret # that to mean that the server doesn't support this # operation. cherrypy.log("Request failed: {0}".format(e)) raise cherrypy.HTTPError(http_client.BAD_REQUEST, str(e)) if pfmri.publisher and not self._get_req_pub(): self.__map_pub_ops(pfmri.publisher) # Set response headers before returning. response.headers["Content-type"] = "text/plain; charset=utf-8" response.headers["Transaction-ID"] = trans_id
def __upload_file(self, *tokens): """Adds a file to an in-flight transaction for the Transaction ID specified in the request path. The content is expected to be in the request body. Returns no output.""" try: # cherrypy decoded it, but we actually need it encoded. trans_id = quote(tokens[0], "") except IndexError: raise trans_id = None request = cherrypy.request response = cherrypy.response size = int(request.headers.get("Content-Length", 0)) if size < 0: raise cherrypy.HTTPError(http_client.BAD_REQUEST, _("file/1 must be sent a file.")) data = request.rfile attrs = dict( val.split("=", 1) for hdr, val in request.headers.items() if hdr.lower().startswith("x-ipkg-setattr") ) basename = attrs.get("basename", None) try: self.repo.add_file(trans_id, data, basename, size) except srepo.RepositoryError as e: # Assume a bad request was made. A 404 can't be # returned here as misc.versioned_urlopen will interpret # that to mean that the server doesn't support this # operation. raise cherrypy.HTTPError(http_client.BAD_REQUEST, str(e)) response.headers["Content-Length"] = "0" return response.body
def __upload_manifest(self, *tokens): """Adds a file to an in-flight transaction for the Transaction ID specified in the request path. The content is expected to be in the request body. Returns no output.""" try: # cherrypy decoded it, but we actually need it encoded. trans_id = quote(tokens[0], "") except IndexError: raise trans_id = None request = cherrypy.request response = cherrypy.response size = int(request.headers.get("Content-Length", 0)) if size < 0: raise cherrypy.HTTPError(http_client.BAD_REQUEST, _("manifest/1 must be sent a file.")) data = request.rfile try: self.repo.add_manifest(trans_id, data) except srepo.RepositoryError as e: # Assume a bad request was made. A 404 can't be # returned here as misc.versioned_urlopen will interpret # that to mean that the server doesn't support this # operation. raise cherrypy.HTTPError(http_client.BAD_REQUEST, str(e)) response.headers["Content-Length"] = "0" return response.body
def index_0(self, *tokens): """Provides an administrative interface for search indexing. Returns no output if successful; otherwise the response body will contain the failure details. """ try: cmd = tokens[0] except IndexError: cmd = "" # These commands cause the operation requested to be queued # for later execution. This does mean that if the operation # fails, the client won't know about it, but this is necessary # since these are long running operations (are likely to exceed # connection timeout limits). try: if cmd == "refresh": # Update search indexes. self.__bgtask.put(self.repo.refresh_index, pub=self._get_req_pub()) else: err = "Unknown index subcommand: {0}".format( cmd) cherrypy.log(err) raise cherrypy.HTTPError(http_client.NOT_FOUND, err) except queue.Full: raise cherrypy.HTTPError(http_client.SERVICE_UNAVAILABLE, "Another operation is already in progress; try " "again later.")
def nasty_before_handler(nasty_depot, maxroll=100): """Cherrypy Tool callable which generates various problems prior to a request. Possible outcomes: retryable HTTP error, short nap.""" # Must be set in _cp_config on associated request handler. assert nasty_depot # Adjust nastiness values once per incoming request. nasty_depot.nasty_housekeeping() # Just roll the main nasty dice once. if not nasty_depot.need_nasty(maxroll=maxroll): return False while True: roll = random.randint(0, 10) if roll == 0: nasty_depot.nasty_nap() if random.randint(0, 1) == 1: # Nap was enough. Let the normal handler run. return False if 1 <= roll <= 8: nasty_depot.nasty_raise_error() else: cherrypy.log("NASTY: return bogus or empty response") response = cherrypy.response response.body = random.choice(['', 'set this is a bogus action', 'Instead of office chair, ' 'package contained bobcat.', '{"this is a": "fragment of json"}']) return True return False
def info(self, *tokens): """Use a DepotHTTP to return an info response.""" dh = self.__build_depot_http() tokens = self.__strip_pub(tokens, dh.repo) # In Python 3, a WSGI application must return bytes as its output return misc.force_bytes(dh.info_0(*tokens[3:]))
def p5i(self, *tokens): """Use a DepotHTTP to return a p5i response.""" dh = self.__build_depot_http() tokens = self.__strip_pub(tokens, dh.repo) headers = cherrypy.response.headers headers["Content-Type"] = pkg.p5i.MIME_TYPE return dh.p5i_0(*tokens[3:])
def search_1(self, *tokens, **params): """Use a DepotHTTP to return a search/1 response.""" toks = cherrypy.request.path_info.lstrip("/").split("/") dh = self.__build_depot_http() toks = self.__strip_pub(tokens, dh.repo) query_str = "/".join(toks[3:]) return dh.search_1(query_str)
def default_error_page(status=http_client.NOT_FOUND, message="oops", traceback=None, version=None): """This function is registered as the default error page for CherryPy errors. This sets the response headers to be uncacheable, and then returns a HTTP response.""" response = cherrypy.response for key in ('Cache-Control', 'Pragma'): if key in response.headers: del response.headers[key] # Server errors are interesting, so let's log them. In the case # of an internal server error, we send a 404 to the client. but # log the full details in the server log. if (status == http_client.INTERNAL_SERVER_ERROR or status.startswith("500 ")): # Convert the error to a 404 to obscure implementation # from the client, but log the original error to the # server logs. error = cherrypy._cperror._HTTPErrorTemplate % \ {"status": http_client.NOT_FOUND, "message": http_client.responses[http_client.NOT_FOUND], "traceback": "", "version": cherrypy.__version__} print("Path that raised exception was {0}".format( cherrypy.request.path_info)) print(message) return error else: error = cherrypy._cperror._HTTPErrorTemplate % \ {"status": http_client.NOT_FOUND, "message": message, "traceback": "", "version": cherrypy.__version__} return error
def set_response(self): """Modify cherrypy.response status, headers, and body to represent self. CherryPy uses this internally, but you can also use it to create an HTTPRedirect object and set its output without *raising* the exception. """ import cherrypy response = cherrypy.serving.response response.status = status = self.status if status in (300, 301, 302, 303, 307): response.headers['Content-Type'] = 'text/html;charset=utf-8' # "The ... URI SHOULD be given by the Location field # in the response." response.headers['Location'] = self.urls[0] # "Unless the request method was HEAD, the entity of the response # SHOULD contain a short hypertext note with a hyperlink to the # new URI(s)." msg = { 300: 'This resource can be found at ', 301: 'This resource has permanently moved to ', 302: 'This resource resides temporarily at ', 303: 'This resource can be found at ', 307: 'This resource has moved temporarily to ', }[status] msg += '<a href=%s>%s</a>.' msgs = [msg % (saxutils.quoteattr(u), u) for u in self.urls] response.body = ntob('<br />\n'.join(msgs), 'utf-8') # Previous code may have set C-L, so we have to reset it # (allow finalize to set it). response.headers.pop('Content-Length', None) elif status == 304: # Not Modified. # "The response MUST include the following header fields: # Date, unless its omission is required by section 14.18.1" # The "Date" header should have been set in Response.__init__ # "...the response SHOULD NOT include other entity-headers." for key in ('Allow', 'Content-Encoding', 'Content-Language', 'Content-Length', 'Content-Location', 'Content-MD5', 'Content-Range', 'Content-Type', 'Expires', 'Last-Modified'): if key in response.headers: del response.headers[key] # "The 304 response MUST NOT contain a message-body." response.body = None # Previous code may have set C-L, so we have to reset it. response.headers.pop('Content-Length', None) elif status == 305: # Use Proxy. # self.urls[0] should be the URI of the proxy. response.headers['Location'] = ntob(self.urls[0], 'utf-8') response.body = None # Previous code may have set C-L, so we have to reset it. response.headers.pop('Content-Length', None) else: raise ValueError('The %s status code is unknown.' % status)
def testHookErrors(self): self.getPage("/demo/?id=1") # If body is "razdrez", then on_end_request is being called too early. self.assertBody("A horrorshow lomtick of cherry 3.14159") # If this fails, then on_end_request isn't being called at all. time.sleep(0.1) self.getPage("/demo/ended/1") self.assertBody("True") valerr = '\n raise ValueError()\nValueError' self.getPage("/demo/err?id=3") # If body is "razdrez", then on_end_request is being called too early. self.assertErrorPage(502, pattern=valerr) # If this fails, then on_end_request isn't being called at all. time.sleep(0.1) self.getPage("/demo/ended/3") self.assertBody("True") # If body is "razdrez", then on_end_request is being called too early. if (cherrypy.server.protocol_version == "HTTP/1.0" or getattr(cherrypy.server, "using_apache", False)): self.getPage("/demo/errinstream?id=5") # Because this error is raised after the response body has # started, the status should not change to an error status. self.assertStatus("200 OK") self.assertBody("nonconfidential") else: # Because this error is raised after the response body has # started, and because it's chunked output, an error is raised by # the HTTP client when it encounters incomplete output. self.assertRaises((ValueError, IncompleteRead), self.getPage, "/demo/errinstream?id=5") # If this fails, then on_end_request isn't being called at all. time.sleep(0.1) self.getPage("/demo/ended/5") self.assertBody("True") # Test the "__call__" technique (compile-time decorator). self.getPage("/demo/restricted") self.assertErrorPage(401) # Test compile-time decorator with kwargs from config. self.getPage("/demo/userid") self.assertBody("Welcome!")
def __set_response_expires(self, op_name, expires, max_age=None): """Used to set expiration headers on a response dynamically based on the name of the operation. 'op_name' is a string containing the name of the depot operation as listed in the REPO_OPS_* constants. 'expires' is an integer value in seconds indicating how long from when the request was made the content returned should expire. The maximum value is 365*86400. 'max_age' is an integer value in seconds indicating the maximum length of time a response should be considered valid. For some operations, the maximum value for this parameter is equal to the repository's refresh_seconds property.""" prefix = self._get_req_pub() if not prefix: prefix = self.repo.cfg.get_property("publisher", "prefix") rs = None if prefix: try: pub = self.repo.get_publisher(prefix) except Exception as e: # Couldn't get pub. pass else: repo = pub.repository if repo: rs = repo.refresh_seconds if rs is None: rs = 14400 if max_age is None: max_age = min((rs, expires)) now = cherrypy.response.time if op_name == "publisher" or op_name == "search" or \ op_name == "catalog": # For these operations, cap the value based on # refresh_seconds. expires = now + min((rs, max_age)) max_age = min((rs, max_age)) else: expires = now + expires headers = cherrypy.response.headers headers["Cache-Control"] = \ "must-revalidate, no-transform, max-age={0:d}".format( max_age) headers["Expires"] = formatdate(timeval=expires, usegmt=True)
def repo_index(self, *tokens, **params): """Generate a page showing the list of repositories served by this Apache instance.""" self.setup(cherrypy.request) # In order to reuse the pkg.depotd shtml files, we need to use # the pkg.server.api, which means passing a DepotBUI object, # despite the fact that we're not serving content for any one # repository. For the purposes of rendering this page, we'll # use the first object we come across. depot = depot_buis[list(depot_buis.keys())[0]] accept_lang = self.get_accept_lang(cherrypy.request, depot) cherrypy.request.path_info = "/{0}".format(accept_lang) tlookup = mako.lookup.TemplateLookup( directories=[depot.web_root]) pub = None base = pkg.server.api.BaseInterface(cherrypy.request, depot, pub) # build a list of all repositories URIs and BUI links, # and a dictionary of publishers for each repository URI repo_list = [] repo_pubs = {} for repo_prefix in repositories.keys(): repo = repositories[repo_prefix] depot = depot_buis[repo_prefix] repo_url = "{0}/{1}".format(cherrypy.request.base, repo_prefix) bui_link = "{0}/{1}/index.shtml".format( repo_prefix, accept_lang) repo_list.append((repo_url, bui_link)) repo_pubs[repo_url] = \ [(pub, "{0}/{1}/{2}".format( cherrypy.request.base, repo_prefix, pub)) for pub in repo.publishers] repo_list.sort() template = tlookup.get_template("repos.shtml") # Starting in CherryPy 3.2, cherrypy.response.body only allows # bytes. return misc.force_bytes(template.render(g_vars={"base": base, "pub": None, "http_depot": "true", "lang": accept_lang, "repo_list": repo_list, "repo_pubs": repo_pubs }))
