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)
python类request()的实例源码
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 flush(self):
"""Flushes the stream."""
try:
stream = cherrypy.serving.request.wsgi_environ.get('wsgi.errors')
except (AttributeError, KeyError):
pass
else:
stream.flush()
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 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 __call__(self):
"""Use this exception as a request.handler (raise self)."""
raise self
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])
