def setup_server():
def break_header():
# Add a header after finalize that is invalid
cherrypy.serving.response.header_list.append((2, 3))
cherrypy.tools.break_header = cherrypy.Tool(
'on_end_resource', break_header)
class Root:
@cherrypy.expose
def index(self):
return "hello"
@cherrypy.config(**{'tools.break_header.on': True})
def start_response_error(self):
return "salud!"
@cherrypy.expose
def stat(self, path):
with cherrypy.HTTPError.handle(OSError, 404):
st = os.stat(path)
root = Root()
cherrypy.tree.mount(root)
python类expose()的实例源码
def setup_upload_server():
class Root:
@cherrypy.expose
def upload(self):
if not cherrypy.request.method == 'POST':
raise AssertionError("'POST' != request.method %r" %
cherrypy.request.method)
return "thanks for '%s'" % tonative(cherrypy.request.body.read())
cherrypy.tree.mount(Root())
cherrypy.config.update({
'server.max_request_body_size': 1001,
'server.socket_timeout': 10,
'server.accepted_queue_size': 5,
'server.accepted_queue_timeout': 0.1,
})
def setup_server():
class Root:
@cherrypy.expose
def resource(self):
return "Oh wah ta goo Siam."
@cherrypy.expose
def fail(self, code):
code = int(code)
if 300 <= code <= 399:
raise cherrypy.HTTPRedirect([], code)
else:
raise cherrypy.HTTPError(code)
@cherrypy.expose
# In Python 3, tools.encode is on by default
@cherrypy.config(**{'tools.encode.on': True})
def unicoded(self):
return ntou('I am a \u1ee4nicode string.', 'escape')
conf = {'/': {'tools.etags.on': True,
'tools.etags.autotags': True,
}}
cherrypy.tree.mount(Root(), config=conf)
def testExpose(self):
# Test the cherrypy.expose function/decorator
self.getPage("/exposing/base")
self.assertBody("expose works!")
self.getPage("/exposing/1")
self.assertBody("expose works!")
self.getPage("/exposing/2")
self.assertBody("expose works!")
self.getPage("/exposingnew/base")
self.assertBody("expose works!")
self.getPage("/exposingnew/1")
self.assertBody("expose works!")
self.getPage("/exposingnew/2")
self.assertBody("expose works!")
def test_syntax(self):
if sys.version_info < (3,):
return self.skip("skipped (Python 3 only)")
code = textwrap.dedent("""
class Root:
@cherrypy.expose
@cherrypy.tools.params()
def resource(self, limit: int):
return type(limit).__name__
conf = {'/': {'tools.params.on': True}}
cherrypy.tree.mount(Root(), config=conf)
""")
exec(code)
self.getPage('/resource?limit=0')
self.assertStatus(200)
self.assertBody('int')
def setup_server():
class ClassOfRoot(object):
def __init__(self, name):
self.name = name
@cherrypy.expose
def index(self):
return "Welcome to the %s website!" % self.name
default = cherrypy.Application(None)
domains = {}
for year in range(1997, 2008):
app = cherrypy.Application(ClassOfRoot('Class of %s' % year))
domains['www.classof%s.example' % year] = app
cherrypy.tree.graft(cherrypy._cpwsgi.VirtualHost(default, domains))
def setup_server():
class Root(object):
@cherrypy.expose
def count(self, clsname):
cherrypy.response.headers['Content-Type'] = 'text/plain'
return six.text_type(globals()[clsname].created)
@cherrypy.expose
def getall(self, clsname):
cherrypy.response.headers['Content-Type'] = 'text/plain'
return globals()[clsname]()
@cherrypy.expose
@cherrypy.config(**{'response.stream': True})
def stream(self, clsname):
return self.getall(clsname)
cherrypy.tree.mount(Root())
def setup_server():
class Root:
@cherrypy.expose
def multipart(self, parts):
return repr(parts)
@cherrypy.expose
def multipart_form_data(self, **kwargs):
return repr(list(sorted(kwargs.items())))
@cherrypy.expose
def flashupload(self, Filedata, Upload, Filename):
return ("Upload: %s, Filename: %s, Filedata: %r" %
(Upload, Filename, Filedata.file.read()))
cherrypy.config.update({'server.max_request_body_size': 0})
cherrypy.tree.mount(Root())
# Client-side code #
def setup_server():
class Root:
@cherrypy.expose
def index(self):
return "hello"
origins = 'http://example.com', 'https://example.com'
config = {
'/': {
'cors.expose.on': True,
'cors.expose.origins': origins,
'cors.preflight.origins': origins,
},
}
cherrypy.tree.mount(Root(), config=config)
cherrypy_cors.install()
def setup_server():
class Root:
@cherrypy.expose
def index(self):
return "hello"
pattern = re.compile(r'(http|https)://svr[1-9]\.example\.com')
config = {
'/': {
'cors.expose.on': True,
'cors.expose.origins': [pattern],
'cors.preflight.origins': [pattern],
},
}
cherrypy.tree.mount(Root(), config=config)
cherrypy_cors.install()
def __init__(self, req):
for method in self.expose:
self.__dict__[method] = getattr(req, method)
def start(self):
self._running = True
cherrypy.config.update({
'server.socket_port': self.port,
'server.socket_host': '0.0.0.0',
'engine.autoreload.on': False,
'log.screen': False
})
self.plugin = DeployerWebSocketPlugin(cherrypy.engine, self)
self.plugin.subscribe()
cherrypy.tools.websocket = WebSocketTool()
class Root(object):
@cherrypy.expose
def index(self):
pass
app = cherrypy.tree.mount(Root(), '/', config={'/': {
'tools.websocket.on': True,
'tools.websocket.handler_cls': DeployerWebSocket
}})
# CherryPy forces a log level for some reason
# let's undo that
app.log.error_log.setLevel(NOTSET)
app.log.access_log.setLevel(NOTSET)
cherrypy.engine.start()
while self._running:
time.sleep(2)
def run_custom_chpy(port, request, response):
class CustomRoot(object):
@cherrypy.expose(request.url)
@cherrypy.tools.allow(methods=[request.method])
def index():
return response
run_cherrypy(port, root=CustomRoot)
def __init__(self, req):
for method in self.expose:
self.__dict__[method] = getattr(req, method)
def test_expose_decorator(self):
# Test @expose
self.getPage("/expose_dec/no_call")
self.assertStatus(200)
self.assertBody("Mr E. R. Bradshaw")
# Test @expose()
self.getPage("/expose_dec/call_empty")
self.assertStatus(200)
self.assertBody("Mrs. B.J. Smegma")
# Test @expose("alias")
self.getPage("/expose_dec/call_alias")
self.assertStatus(200)
self.assertBody("Mr Nesbitt")
# Does the original name work?
self.getPage("/expose_dec/nesbitt")
self.assertStatus(200)
self.assertBody("Mr Nesbitt")
# Test @expose(["alias1", "alias2"])
self.getPage("/expose_dec/alias1")
self.assertStatus(200)
self.assertBody("Mr Ken Andrews")
self.getPage("/expose_dec/alias2")
self.assertStatus(200)
self.assertBody("Mr Ken Andrews")
# Does the original name work?
self.getPage("/expose_dec/andrews")
self.assertStatus(200)
self.assertBody("Mr Ken Andrews")
# Test @expose(alias="alias")
self.getPage("/expose_dec/alias3")
self.assertStatus(200)
self.assertBody("Mr. and Mrs. Watson")
def testTreeMounting(self):
class Root(object):
@cherrypy.expose
def hello(self):
return "Hello world!"
# When mounting an application instance,
# we can't specify a different script name in the call to mount.
a = Application(Root(), '/somewhere')
self.assertRaises(ValueError, cherrypy.tree.mount, a, '/somewhereelse')
# When mounting an application instance...
a = Application(Root(), '/somewhere')
# ...we MUST allow in identical script name in the call to mount...
cherrypy.tree.mount(a, '/somewhere')
self.getPage('/somewhere/hello')
self.assertStatus(200)
# ...and MUST allow a missing script_name.
del cherrypy.tree.apps['/somewhere']
cherrypy.tree.mount(a)
self.getPage('/somewhere/hello')
self.assertStatus(200)
# In addition, we MUST be able to create an Application using
# script_name == None for access to the wsgi_environ.
a = Application(Root(), script_name=None)
# However, this does not apply to tree.mount
self.assertRaises(TypeError, cherrypy.tree.mount, a, None)
def testKeywords(self):
if sys.version_info < (3,):
return self.skip("skipped (Python 3 only)")
exec("""class Root(object):
@cherrypy.expose
def hello(self, *, name='world'):
return 'Hello %s!' % name
cherrypy.tree.mount(Application(Root(), '/keywords'))""")
self.getPage('/keywords/hello')
self.assertStatus(200)
self.getPage('/keywords/hello/extra')
self.assertStatus(404)
def setup_server():
class Root(object):
@cherrypy.expose
def plain(self):
return 'hello'
@cherrypy.expose
@json_out
def json_string(self):
return 'hello'
@cherrypy.expose
@json_out
def json_list(self):
return ['a', 'b', 42]
@cherrypy.expose
@json_out
def json_dict(self):
return {'answer': 42}
@cherrypy.expose
@json_in
def json_post(self):
if cherrypy.request.json == [13, 'c']:
return 'ok'
else:
return 'nok'
@cherrypy.expose
@json_out
@cherrypy.config(**{'tools.caching.on': True})
def json_cached(self):
return 'hello there'
root = Root()
cherrypy.tree.mount(root)
def setup_server():
class Root:
@cherrypy.expose
@cherrypy.tools.params()
def resource(self, limit=None, sort=None):
return type(limit).__name__
# for testing on Py 2
resource.__annotations__ = {'limit': int}
conf = {'/': {'tools.params.on': True}}
cherrypy.tree.mount(Root(), config=conf)
def setup_server():
class Root:
@cherrypy.expose
def index(self):
return cherrypy.request.wsgi_environ['SERVER_PORT']
@cherrypy.expose
def upload(self, file):
return "Size: %s" % len(file.file.read())
@cherrypy.expose
@cherrypy.config(**{'request.body.maxbytes': 100})
def tinyupload(self):
return cherrypy.request.body.read()
cherrypy.tree.mount(Root())
cherrypy.config.update({
'server.socket_host': '0.0.0.0',
'server.socket_port': 9876,
'server.max_request_body_size': 200,
'server.max_request_header_size': 500,
'server.socket_timeout': 0.5,
# Test explicit server.instance
'server.2.instance': 'cherrypy._cpwsgi_server.CPWSGIServer',
'server.2.socket_port': 9877,
# Test non-numeric <servername>
# Also test default server.instance = builtin server
'server.yetanother.socket_port': 9878,
})
def setup_server():
class Root:
@cherrypy.expose
def index(self):
return "Hello World"
@cherrypy.expose
def ctrlc(self):
raise KeyboardInterrupt()
@cherrypy.expose
def graceful(self):
engine.graceful()
return "app was (gracefully) restarted succesfully"
@cherrypy.expose
def block_explicit(self):
while True:
if cherrypy.response.timed_out:
cherrypy.response.timed_out = False
return "broken!"
time.sleep(0.01)
@cherrypy.expose
def block_implicit(self):
time.sleep(0.5)
return "response.timeout = %s" % cherrypy.response.timeout
cherrypy.tree.mount(Root())
cherrypy.config.update({
'environment': 'test_suite',
'engine.timeout_monitor.frequency': 0.1,
})
db_connection.subscribe()
# ------------ Enough helpers. Time for real live test cases. ------------ #
def setup_server():
class Root:
@cherrypy.expose
def index(self):
return "hello"
config = {
'/': {
'cors.expose.on': True,
},
}
cherrypy.tree.mount(Root(), config=config)
cherrypy_cors.install()
def setup_server():
class Root:
@cherrypy.expose
@cherrypy_cors.tools.expose()
def index(self):
return "hello"
cherrypy.tree.mount(Root())
cherrypy_cors.install()
def run(cls):
cherrypy_cors.install()
config = {
'/': {
'cors.expose.on': True,
},
}
cherrypy.quickstart(cls(), config=config)
def expose(self, allow_credentials, expose_headers, origins):
if self._is_valid_origin(origins):
self._add_origin_and_credentials_headers(allow_credentials)
self._add_expose_headers(expose_headers)
return True
return False
def setup_server():
class Root:
@cherrypy.expose
def index(self):
return "Hello, world"
@cherrypy.expose
def dom4(self):
return "Under construction"
@cherrypy.expose
def method(self, value):
return "You sent %s" % value
class VHost:
def __init__(self, sitename):
self.sitename = sitename
@cherrypy.expose
def index(self):
return "Welcome to %s" % self.sitename
@cherrypy.expose
def vmethod(self, value):
return "You sent %s" % value
@cherrypy.expose
def url(self):
return cherrypy.url("nextpage")
# Test static as a handler (section must NOT include vhost prefix)
static = cherrypy.tools.staticdir.handler(
section='/static', dir=curdir)
root = Root()
root.mydom2 = VHost("Domain 2")
root.mydom3 = VHost("Domain 3")
hostmap = {'www.mydom2.com': '/mydom2',
'www.mydom3.com': '/mydom3',
'www.mydom4.com': '/dom4',
}
cherrypy.tree.mount(root, config={
'/': {
'request.dispatch': cherrypy.dispatch.VirtualHost(**hostmap)
},
# Test static in config (section must include vhost prefix)
'/mydom2/static2': {
'tools.staticdir.on': True,
'tools.staticdir.root': curdir,
'tools.staticdir.dir': 'static',
'tools.staticdir.index': 'index.html',
},
})
def setup_server():
class Root:
@cherrypy.expose
def index(self, *args, **kwargs):
return "Hello world!"
@cherrypy.expose
@cherrypy.config(**{'request.process_request_body': False})
def no_body(self, *args, **kwargs):
return "Hello world!"
@cherrypy.expose
def post_multipart(self, file):
"""Return a summary ("a * 65536\nb * 65536") of the uploaded
file.
"""
contents = file.file.read()
summary = []
curchar = None
count = 0
for c in contents:
if c == curchar:
count += 1
else:
if count:
if six.PY3:
curchar = chr(curchar)
summary.append("%s * %d" % (curchar, count))
count = 1
curchar = c
if count:
if six.PY3:
curchar = chr(curchar)
summary.append("%s * %d" % (curchar, count))
return ", ".join(summary)
@cherrypy.expose
def post_filename(self, myfile):
'''Return the name of the file which was uploaded.'''
return myfile.filename
cherrypy.tree.mount(Root())
cherrypy.config.update({'server.max_request_body_size': 30000000})
def setup_server():
# Set up site
cherrypy.config.update({
'tools.proxy.on': True,
'tools.proxy.base': 'www.mydomain.test',
})
# Set up application
class Root:
def __init__(self, sn):
# Calculate a URL outside of any requests.
self.thisnewpage = cherrypy.url(
"/this/new/page", script_name=sn)
@cherrypy.expose
def pageurl(self):
return self.thisnewpage
@cherrypy.expose
def index(self):
raise cherrypy.HTTPRedirect('dummy')
@cherrypy.expose
def remoteip(self):
return cherrypy.request.remote.ip
@cherrypy.expose
@cherrypy.config(**{
'tools.proxy.local': 'X-Host',
'tools.trailing_slash.extra': True,
})
def xhost(self):
raise cherrypy.HTTPRedirect('blah')
@cherrypy.expose
def base(self):
return cherrypy.request.base
@cherrypy.expose
@cherrypy.config(**{'tools.proxy.scheme': 'X-Forwarded-Ssl'})
def ssl(self):
return cherrypy.request.base
@cherrypy.expose
def newurl(self):
return ("Browse to <a href='%s'>this page</a>."
% cherrypy.url("/this/new/page"))
for sn in script_names:
cherrypy.tree.mount(Root(sn), sn)
def setup_server():
def check(username, password):
# Dummy check_username_and_password function
if username != 'test' or password != 'password':
return 'Wrong login/password'
def augment_params():
# A simple tool to add some things to request.params
# This is to check to make sure that session_auth can handle
# request params (ticket #780)
cherrypy.request.params["test"] = "test"
cherrypy.tools.augment_params = cherrypy.Tool(
'before_handler', augment_params, None, priority=30)
class Test:
_cp_config = {
'tools.sessions.on': True,
'tools.session_auth.on': True,
'tools.session_auth.check_username_and_password': check,
'tools.augment_params.on': True,
}
@cherrypy.expose
def index(self, **kwargs):
return "Hi %s, you are logged in" % cherrypy.request.login
cherrypy.tree.mount(Test())
def setup_server():
class Root:
@cherrypy.expose
def index(self):
return "hello"
@cherrypy.expose
def uni_code(self):
cherrypy.request.login = tartaros
cherrypy.request.remote.name = erebos
@cherrypy.expose
def slashes(self):
cherrypy.request.request_line = r'GET /slashed\path HTTP/1.1'
@cherrypy.expose
def whitespace(self):
# User-Agent = "User-Agent" ":" 1*( product | comment )
# comment = "(" *( ctext | quoted-pair | comment ) ")"
# ctext = <any TEXT excluding "(" and ")">
# TEXT = <any OCTET except CTLs, but including LWS>
# LWS = [CRLF] 1*( SP | HT )
cherrypy.request.headers['User-Agent'] = 'Browzuh (1.0\r\n\t\t.3)'
@cherrypy.expose
def as_string(self):
return "content"
@cherrypy.expose
def as_yield(self):
yield "content"
@cherrypy.expose
@cherrypy.config(**{'tools.log_tracebacks.on': True})
def error(self):
raise ValueError()
root = Root()
cherrypy.config.update({
'log.error_file': error_log,
'log.access_file': access_log,
})
cherrypy.tree.mount(root)