Python cherrypy 模块，expose() 实例源码

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)

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)

def setup_server():
        class Root:

            @cherrypy.expose
            def index(self):
                return "This is public."

        class BasicProtected:

            @cherrypy.expose
            def index(self):
                return "Hello %s, you've been authorized." % (
                    cherrypy.request.login)

        class BasicProtected2:

            @cherrypy.expose
            def index(self):
                return "Hello %s, you've been authorized." % (
                    cherrypy.request.login)

        userpassdict = {'xuser': 'xpassword'}
        userhashdict = {'xuser': md5(ntob('xpassword')).hexdigest()}

        def checkpasshash(realm, user, password):
            p = userhashdict.get(user)
            return p and p == md5(ntob(password)).hexdigest() or False

        basic_checkpassword_dict = auth_basic.checkpassword_dict(userpassdict)
        conf = {
            '/basic': {
                'tools.auth_basic.on': True,
                'tools.auth_basic.realm': 'wonderland',
                'tools.auth_basic.checkpassword': basic_checkpassword_dict
            },
            '/basic2': {
                'tools.auth_basic.on': True,
                'tools.auth_basic.realm': 'wonderland',
                'tools.auth_basic.checkpassword': checkpasshash
            },
        }

        root = Root()
        root.basic = BasicProtected()
        root.basic2 = BasicProtected2()
        cherrypy.tree.mount(root, config=conf)

def preflight(allowed_methods, allowed_headers=None, allow_credentials=False,
              max_age=None, origins=None):
    """Adds CORS `preflight`_ support to a `HTTP OPTIONS` request.

    :param allowed_methods: List of supported `HTTP` methods
                            (see `Access-Control-Allow-Methods`_).
    :type allowed_methods: list or NoneType
    :param allowed_headers: List of supported `HTTP` headers
                            (see `Access-Control-Allow-Headers`_).
    :type allowed_headers: list or NoneType
    :param allow_credentials: Use credentials to make cookies work
                              (see `Access-Control-Allow-Credentials`_).
    :type allow_credentials: bool
    :param max_age: Seconds to cache the preflight request
                    (see `Access-Control-Max-Age`_).
    :type max_age: int
    :param origins: List of allowed origins clients must reference.
    :type origins: list or NoneType

    :returns: Whether the preflight is allowed.
    :rtype: bool

    - Used as a decorator with the `Method Dispatcher`_

        .. code-block:: python

                @cherrypy_cors.tools.preflight(
                    allowed_methods=["GET", "DELETE", "PUT"])
                def OPTIONS(self):
                    pass

    - Function call with the `Object Dispatcher`_

        .. code-block:: python

                @cherrypy.expose
                @cherrypy.tools.allow(
                    methods=["GET", "DELETE", "PUT", "OPTIONS"])
                def thing(self):
                    if cherrypy.request.method == "OPTIONS":
                        cherrypy_cors.preflight(
                            allowed_methods=["GET", "DELETE", "PUT"])
                    else:
                        self._do_other_things()

    """
    if _get_cors().preflight(allowed_methods, allowed_headers,
                             allow_credentials, max_age, origins):
        _safe_caching_headers()
        return True
    return False