我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用cherrypy.request()。
def _send_email(cls): """ Check the email type and send the requested email using the email plugin. """ if not hasattr(cherrypy.request, 'email_address'): return if not hasattr(cherrypy.request, 'email_type'): return cherrypy.request.user = User.get_by_email(cherrypy.request.email_address) if 'mail' in cherrypy.request.user: if cherrypy.request.email_type == 'password': cherrypy.engine.publish('email-send-reset', cherrypy.request.user) if cherrypy.request.email_type == 'username': cherrypy.engine.publish('email-send-username', cherrypy.request.user)
def start_handler(self): """ Runs at the end of the request processing by calling the opened method of the handler. """ request = cherrypy.request if not hasattr(request, 'ws_handler'): return addr = (request.remote.ip, request.remote.port) ws_handler = request.ws_handler request.ws_handler = None delattr(request, 'ws_handler') # By doing this we detach the socket from # the CherryPy stack avoiding memory leaks detach_connection(request.rfile.rfile) cherrypy.engine.publish('handle-websocket', ws_handler, addr)
def emit(self, record): """Emit a record.""" try: stream = cherrypy.serving.request.wsgi_environ.get('wsgi.errors') except (AttributeError, KeyError): pass else: try: msg = self.format(record) fs = '%s\n' import types # if no unicode support... if not hasattr(types, 'UnicodeType'): stream.write(fs % msg) else: try: stream.write(fs % msg) except UnicodeError: stream.write(fs % msg.encode('UTF-8')) self.flush() except: self.handleError(record)
def __init__(self, path, query_string=''): import cherrypy self.request = cherrypy.serving.request self.query_string = query_string if '?' in path: # Separate any params included in the path path, self.query_string = path.split('?', 1) # Note that urljoin will "do the right thing" whether url is: # 1. a URL relative to root (e.g. "/dummy") # 2. a URL relative to the current path # Note that any query string will be discarded. path = _urljoin(self.request.path_info, path) # Set a 'path' member attribute so that code which traps this # error can have access to it. self.path = path CherryPyException.__init__(self, path, self.query_string)
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 most_similar(self, *args, **kwargs): positive = cherrypy.request.params.get('positive[]', []) if isinstance(positive, basestring): positive = [positive] negative = cherrypy.request.params.get('negative[]', []) if isinstance(negative, basestring): negative = [negative] try: result = self.model.most_similar( positive=[gensim.utils.to_utf8(word).strip() for word in positive if word], negative=[gensim.utils.to_utf8(word).strip() for word in negative if word], topn=5) except: result = [] logger.info("similars for %s vs. %s: %s" % (positive, negative, result)) return {'similars': result}
def emit(self, record): """Emit a record.""" try: stream = cherrypy.serving.request.wsgi_environ.get('wsgi.errors') except (AttributeError, KeyError): pass else: try: msg = self.format(record) fs = "%s\n" import types # if no unicode support... if not hasattr(types, "UnicodeType"): stream.write(fs % msg) else: try: stream.write(fs % msg) except UnicodeError: stream.write(fs % msg.encode("UTF-8")) self.flush() except: self.handleError(record)
def script_name(self, path=None): """The script_name of the app at the given path, or None. If path is None, cherrypy.request is used. """ if path is None: try: request = cherrypy.serving.request path = httputil.urljoin(request.script_name, request.path_info) except AttributeError: return None while True: if path in self.apps: return path if path == "": return None # Move one node up the tree and try again. path = path[:path.rfind("/")]
def __init__(self, path, query_string=""): import cherrypy self.request = cherrypy.serving.request self.query_string = query_string if "?" in path: # Separate any params included in the path path, self.query_string = path.split("?", 1) # Note that urljoin will "do the right thing" whether url is: # 1. a URL relative to root (e.g. "/dummy") # 2. a URL relative to the current path # Note that any query string will be discarded. path = _urljoin(self.request.path_info, path) # Set a 'path' member attribute so that code which traps this # error can have access to it. self.path = path CherryPyException.__init__(self, path, self.query_string)
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 _get_req_pub(self): """Private helper function to retrieve the publisher prefix for the current operation from the request path. Returns None if a publisher prefix was not found in the request path. The publisher is assumed to be the first component of the path_info string if it doesn't match the operation's name. This does mean that a publisher can't be named the same as an operation, but that isn't viewed as an unreasonable limitation. """ if self.request_pub_func: return self.request_pub_func(cherrypy.request.path_info) try: req_pub = cherrypy.request.path_info.strip("/").split( "/")[0] except IndexError: return None if req_pub not in self.REPO_OPS_DEFAULT and req_pub != "feed": # Assume that if the first component of the request path # doesn't match a known operation that it's a publisher # prefix. return req_pub return None
def catalog_1(self, *tokens): """Outputs the contents of the specified catalog file, using the name in the request path, directly to the client.""" try: name = tokens[0] except IndexError: raise cherrypy.HTTPError(http_client.FORBIDDEN, _("Directory listing not allowed.")) try: fpath = self.repo.catalog_1(name, pub=self._get_req_pub()) 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)) self.__set_response_expires("catalog", 86400, 86400) return serve_file(fpath, "text/plain; charset=utf-8")
def manifest_1(self, *tokens): """Outputs the contents of the manifest or uploads the manifest.""" method = cherrypy.request.method if method == "GET": return self.manifest_0(*tokens) elif method in ("POST", "PUT"): return self.__upload_manifest(*tokens) raise cherrypy.HTTPError(http_client.METHOD_NOT_ALLOWED, "{0} is not allowed".format(method)) # We need to prevent cherrypy from processing the request body so that # manifest can parse the request body itself. In addition, we also need # to set the timeout higher since the default is five minutes; not # really enough for a slow connection to upload content.
def file_0(self, *tokens): """Outputs the contents of the file, named by the SHA-1 hash name in the request path, directly to the client.""" 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)) self.__set_response_expires("file", 86400*365, 86400*365) return serve_file(fpath, "application/data")
def file_1(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 == "GET": return self.file_0(*tokens) elif method in ("POST", "PUT"): return self.__upload_file(*tokens) raise cherrypy.HTTPError(http_client.METHOD_NOT_ALLOWED, "{0} is not allowed".format(method)) # We need to prevent cherrypy from processing the request body so that # file can parse the request body itself. In addition, we also need to # set the timeout higher since the default is five minutes; not really # enough for a slow connection to upload content.
def abandon_0(self, *tokens): """Aborts an in-flight transaction for the Transaction ID specified in the request path. Returns no output.""" try: # cherrypy decoded it, but we actually need it encoded. trans_id = quote(tokens[0], "") except IndexError: trans_id = None try: self.repo.abandon(trans_id) 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))
def start(self): """Start the background task thread. Since we configure mod_wsgi with an inactivity-timeout, long-running background tasks which don't cause new WSGI HTTP requests can result in us hitting that inactivity-timeout. To prevent this, while background tasks are running, we periodically send a HTTP request to the server.""" self.__running = True if not self.__thread: # Create and start a thread for the caller. self.__thread = threading.Thread(target=self.run) self.__thread.start() self.__keep_busy_thread = threading.Thread( target=self.run_keep_busy) self.__keep_busy_thread.start()
def manifest(self, *tokens): """Manifest requests coming from the BUI need to be redirected back through the RewriteRules defined in the Apache configuration in order to be served directly. pkg(1) will never hit this code, as those requests don't get handled by this webapp. """ self.setup(cherrypy.request) rel_uri = cherrypy.request.path_info # we need to recover the escaped portions of the URI redir = rel_uri.lstrip("/").split("/") pub_mf = "/".join(redir[0:4]) pkg_name = "/".join(redir[4:]) # encode the URI so our RewriteRules can process them pkg_name = quote(pkg_name) pkg_name = pkg_name.replace("/", "%2F") pkg_name = pkg_name.replace("%40", "@", 1) # build a URI that we can redirect to redir = "{0}/{1}".format(pub_mf, pkg_name) redir = "/{0}".format(redir.lstrip("/")) raise cherrypy.HTTPRedirect(redir)
def __init__(self): """ The Email Tool provides functionality for sending password reset emails in the background rather than blocking the request. The user should not be made aware of success or failure, so all error handling is logged. """ cherrypy.Tool.__init__(self, 'on_end_request', self._send_email, priority=80)
def _setup(self): conf = self._merged_args() hooks = cherrypy.serving.request.hooks p = conf.pop("priority", getattr(self.callable, "priority", self._priority)) hooks.attach(self._point, self.callable, priority=p, **conf) hooks.attach('before_finalize', self.complete, priority=p) hooks.attach('on_end_resource', self.cleanup_headers, priority=70) hooks.attach('on_end_request', self.start_handler, priority=70)
def index(self): cherrypy.log("Handler created: %s" % repr(cherrypy.request.ws_handler))
def flush(self): """Flushes the stream.""" try: stream = cherrypy.serving.request.wsgi_environ.get('wsgi.errors') except (AttributeError, KeyError): pass else: stream.flush()
def _get_script_name(self): if self._script_name is not None: return self._script_name # A `_script_name` with a value of None signals that the script name # should be pulled from WSGI environ. return cherrypy.serving.request.wsgi_environ['SCRIPT_NAME'].rstrip('/')
def release_serving(self): """Release the current serving (request and response).""" req = cherrypy.serving.request cherrypy.engine.publish('after_request') try: req.close() except: cherrypy.log(traceback=True, severity=40) cherrypy.serving.clear()
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 __call__(self): """Use this exception as a request.handler (raise self).""" raise self
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 __init__(self, path=None): if path is None: import cherrypy request = cherrypy.serving.request path = request.script_name + request.path_info self.args = (path,) HTTPError.__init__(self, 404, "The path '%s' was not found." % path)
def bare_error(extrabody=None): """Produce status, headers, body for a critical error. Returns a triple without calling any other questionable functions, so it should be as error-free as possible. Call it from an HTTP server if you get errors outside of the request. If extrabody is None, a friendly but rather unhelpful error message is set in the body. If extrabody is a string, it will be appended as-is to the body. """ # The whole point of this function is to be a last line-of-defense # in handling errors. That is, it must not raise any errors itself; # it cannot be allowed to fail. Therefore, don't add to it! # In particular, don't call any other CP functions. body = ntob('Unrecoverable error in the server.') if extrabody is not None: if not isinstance(extrabody, bytes): extrabody = extrabody.encode('utf-8') body += ntob('\n') + extrabody return (ntob('500 Internal Server Error'), [(ntob('Content-Type'), ntob('text/plain')), (ntob('Content-Length'), ntob(str(len(body)), 'ISO-8859-1'))], [body])
def server_exception_wrap(func): """ Method decorator to return nicer JSON responses: handle internal server errors & report request timings. """ @wraps(func) def _wrapper(self, *args, **kwargs): try: # append "success=1" and time taken in milliseconds to the response, on success logger.debug("calling server method '%s'" % (func.func_name)) cherrypy.response.timeout = 3600 * 24 * 7 # [s] # include json data in kwargs; HACK: should be a separate decorator really, but whatever if getattr(cherrypy.request, 'json', None): kwargs.update(cherrypy.request.json) start = time.time() result = func(self, *args, **kwargs) if result is None: result = {} result['success'] = 1 result['taken'] = time.time() - start logger.info("method '%s' succeeded in %ss" % (func.func_name, result['taken'])) return result except Exception, e: logger.exception("exception serving request") result = { 'error': repr(e), 'success': 0, } cherrypy.response.status = 500 return result return _wrapper
def most_dissimilar(self, *args, **kwargs): words = cherrypy.request.params.get('words[]', '').split() try: result = self.model.doesnt_match(words) except: result = '' logger.info("dissimilar for %s: %s" % (words, result)) return {'dissimilar': result}
def render_home(self): context = common_context( cherrypy.config[setting_name('AUTHENTICATION_BACKENDS')], load_strategy(), user=getattr(cherrypy.request, 'user', None), plus_id=cherrypy.config.get(setting_name('SOCIAL_AUTH_GOOGLE_PLUS_KEY')) ) return cherrypy.tools.jinja2env \ .get_template("home.html") \ .render(**context)
def load_user(): user_id = cherrypy.session.get('user_id') if user_id: cherrypy.request.user = cherrypy.request.db.query(User).get(user_id) else: cherrypy.request.user = None
def _setup(self): super(SATool, self)._setup() cherrypy.request.hooks.attach('on_end_resource', self.commit_transaction, priority=80)
def bind_session(self): session = cherrypy.engine.publish('bind-session').pop() cherrypy.request.db = session
def commit_transaction(self): if not hasattr(cherrypy.request, 'db'): return cherrypy.request.db = None cherrypy.engine.publish('commit-session')
def test_encoded_headers(self): # First, make sure the innards work like expected. self.assertEqual( httputil.decode_TEXT(ntou("=?utf-8?q?f=C3=BCr?=")), ntou("f\xfcr")) if cherrypy.server.protocol_version == "HTTP/1.1": # Test RFC-2047-encoded request and response header values u = ntou('\u212bngstr\xf6m', 'escape') c = ntou("=E2=84=ABngstr=C3=B6m") self.getPage("/headers/ifmatch", [('If-Match', ntou('=?utf-8?q?%s?=') % c)]) # The body should be utf-8 encoded. self.assertBody(ntob("\xe2\x84\xabngstr\xc3\xb6m")) # But the Etag header should be RFC-2047 encoded (binary) self.assertHeader("ETag", ntou('=?utf-8?b?4oSrbmdzdHLDtm0=?=')) # Test a *LONG* RFC-2047-encoded request and response header value self.getPage("/headers/ifmatch", [('If-Match', ntou('=?utf-8?q?%s?=') % (c * 10))]) self.assertBody(ntob("\xe2\x84\xabngstr\xc3\xb6m") * 10) # Note: this is different output for Python3, but it decodes fine. etag = self.assertHeader( "ETag", '=?utf-8?b?4oSrbmdzdHLDtm3ihKtuZ3N0csO2beKEq25nc3Ryw7Zt' '4oSrbmdzdHLDtm3ihKtuZ3N0csO2beKEq25nc3Ryw7Zt' '4oSrbmdzdHLDtm3ihKtuZ3N0csO2beKEq25nc3Ryw7Zt' '4oSrbmdzdHLDtm0=?=') self.assertEqual(httputil.decode_TEXT(etag), u * 10)
def test_header_presence(self): # If we don't pass a Content-Type header, it should not be present # in cherrypy.request.headers self.getPage("/headers/Content-Type", headers=[]) self.assertStatus(500) # If Content-Type is present in the request, it should be present in # cherrypy.request.headers self.getPage("/headers/Content-Type", headers=[("Content-type", "application/json")]) self.assertBody("application/json")
def testEndRequestOnDrop(self): old_timeout = None try: httpserver = cherrypy.server.httpserver old_timeout = httpserver.timeout except (AttributeError, IndexError): return self.skip() try: httpserver.timeout = timeout # Test that on_end_request is called even if the client drops. self.persistent = True try: conn = self.HTTP_CONN conn.putrequest("GET", "/demo/stream?id=9", skip_host=True) conn.putheader("Host", self.HOST) conn.endheaders() # Skip the rest of the request and close the conn. This will # cause the server's active socket to error, which *should* # result in the request being aborted, and request.close being # called all the way up the stack (including WSGI middleware), # eventually calling our on_end_request hook. finally: self.persistent = False time.sleep(timeout * 2) # Test that the on_end_request hook was called. self.getPage("/demo/ended/9") self.assertBody("True") finally: if old_timeout is not None: httpserver.timeout = old_timeout
def _get_script_name(self): if self._script_name is not None: return self._script_name # A `_script_name` with a value of None signals that the script name # should be pulled from WSGI environ. return cherrypy.serving.request.wsgi_environ['SCRIPT_NAME'].rstrip("/")
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 bare_error(extrabody=None): """Produce status, headers, body for a critical error. Returns a triple without calling any other questionable functions, so it should be as error-free as possible. Call it from an HTTP server if you get errors outside of the request. If extrabody is None, a friendly but rather unhelpful error message is set in the body. If extrabody is a string, it will be appended as-is to the body. """ # The whole point of this function is to be a last line-of-defense # in handling errors. That is, it must not raise any errors itself; # it cannot be allowed to fail. Therefore, don't add to it! # In particular, don't call any other CP functions. body = ntob("Unrecoverable error in the server.") if extrabody is not None: if not isinstance(extrabody, bytes): extrabody = extrabody.encode('utf-8') body += ntob("\n") + extrabody return (ntob("500 Internal Server Error"), [(ntob('Content-Type'), ntob('text/plain')), (ntob('Content-Length'), ntob(str(len(body)), 'ISO-8859-1'))], [body])
