Python bottle 模块，app() 实例源码

def purge(args, test=False):
    urlpath = '/VolumeDriver.Snapshots.Purge'
    param = {'Name': args.name[0],
             'Pattern': args.pattern[0],
             'Dryrun': args.dryrun}
    if test:
        param['Test'] = True
        resp = TestApp(app).post(urlpath, json.dumps(param))
    else:
        resp = Session().post(
            'http+unix://{}{}'
            .format(urllib.parse.quote_plus(SOCKET), urlpath),
            json.dumps(param))
    res = get_from(resp, '')
    if res:
        print(res)
    return res

def setup_app() -> (SessionMiddleware, int, bool):
    bottle.debug(True)
    p = get_config(section='api', key='port', default=5080)
    bottle.install(error_translation)
    session_opts = {
        'session.cookie_expires': True,
        'session.encrypt_key': get_config('api', 'encrypt_key', 'softfire'),
        'session.httponly': True,
        'session.timeout': 3600 * 24,  # 1 day
        'session.type': 'cookie',
        'session.validate_key': True,
    }
    a = SessionMiddleware(bottle.app(), session_opts)
    qb = get_config('api', 'quiet', 'true').lower() == 'true'
    logger.debug("Bootlepy quiet mode: %s" % qb)
    return a, p, qb

def examples():
    http_host = bottle.request.environ.get('HTTP_HOST', 'localhost:3350')
    configs = restful_rfcat.example_configs.get(http_host)
    if is_cli():
        bottle.response.content_type = 'text/plain'
        lines = []
        for software, software_configs in configs.items():
            for variant, config in software_configs.items():
                lines.append("# %s - %s" % (software, variant))
                lines.append('')
                lines.append(config)
                lines.append('')
        return '\n'.join(lines)
    else:
        page = markup.page()
        page.init(script=['app.js'], css=['style.css'])
        for software, software_configs in configs.items():
            page.h2(software)
            for variant, config in software_configs.items():
                page.h3('%s - %s' % (software, variant))
                page.pre(config)
        return str(page)

def snapshot(args, test=False):
    urlpath = '/VolumeDriver.Snapshot'
    param = json.dumps({'Name': args.name[0]})
    if test:
        resp = TestApp(app).post(urlpath, param)
    else:
        resp = Session().post(
            'http+unix://{}{}'
            .format(urllib.parse.quote_plus(SOCKET), urlpath),
            param)
    res = get_from(resp, 'Snapshot')
    if res:
        print(res)
    return res

def send(args, test=False):
    urlpath = '/VolumeDriver.Snapshot.Send'
    param = {'Name': args.snapshot[0], 'Host': args.host[0]}
    if test:
        param['Test'] = True
        resp = TestApp(app).post(urlpath, json.dumps(param))
    else:
        resp = Session().post(
            'http+unix://{}{}'
            .format(urllib.parse.quote_plus(SOCKET), urlpath),
            json.dumps(param))
    res = get_from(resp, '')
    if res:
        print(res)
    return res

def sync(args, test=False):
    urlpath = '/VolumeDriver.Volume.Sync'
    param = {'Volumes': args.volumes, 'Hosts': args.hosts}
    if test:
        param['Test'] = True
        resp = TestApp(app).post(urlpath, json.dumps(param))
    else:
        resp = Session().post(
            'http+unix://{}{}'
            .format(urllib.parse.quote_plus(SOCKET), urlpath),
            json.dumps(param))
    res = get_from(resp, '')
    if res:
        print(res)
    return res

def run(args):
    if not os.path.exists(SNAPSHOTS_PATH):
        log.info('Creating %s', SNAPSHOTS_PATH)
        os.makedirs(SNAPSHOTS_PATH, exist_ok=True)
    # run a thread for the scheduled snapshots
    print('Starting scheduler job every {}s'.format(TIMER))
    Timer(1, scheduler).start()
    # listen to requests
    serve(app, unix_socket=SOCKET, unix_socket_perms='660')

def __init__(self, host, port):
        # associate the session to our app
        local_app = SessionMiddleware(bottle.app())

        bottle.run(
            app=local_app,
            server='cherrypy',
            host=host,
            port=port,
            reloader=True
        )

def main():
    if len(sys.argv) > 1:
        config_file = sys.argv[1]
    else:
        config_file = DEFAULT_CONFIG

    app = make_app(config_file)

    with codecs.open(config_file, 'r', 'utf8') as f:
        config = konfig.Config(f)

    run(app=app,
        host=config['henet'].get('host', DEFAULT_HOST),
        port=config['henet'].get('port', DEFAULT_PORT),
        server='waitress')

def start_listening():
    logger.info("Running bottle app: quiet=%s, port=%s, host='0.0.0.0'" % (quiet_bottle, port))
    bottle.run(app=app, quiet=quiet_bottle, port=port, host='0.0.0.0')

def run_simple(host="0.0.0.0", port="8000"):
    run(app=web, host=host, port=port, quiet=True)

def run_lightweight(host="0.0.0.0", port="8000"):
    run(app=web, host=host, port=port, quiet=True, server="bjoern")

def run_threaded(host="0.0.0.0", port="8000", theads=3, cert="", key=""):
    from wsgiserver import CherryPyWSGIServer

    if cert and key:
        CherryPyWSGIServer.ssl_certificate = cert
        CherryPyWSGIServer.ssl_private_key = key

    CherryPyWSGIServer.numthreads = theads

    from utils import CherryPyWSGI

    run(app=web, host=host, port=port, server=CherryPyWSGI, quiet=True)

def run_fcgi(host="0.0.0.0", port="8000"):
    from bottle import FlupFCGIServer

    run(app=web, host=host, port=port, server=FlupFCGIServer, quiet=True)

def test_request_attrs(self):
        """ WSGI: POST routes"""
        @bottle.route('/')
        def test():
            self.assertEqual(bottle.request.app,
                             bottle.default_app())
            self.assertEqual(bottle.request.route,
                             bottle.default_app().routes[0])
            return 'foo'
        self.assertBody('foo', '/')

def test_routebuild(self):
        """ WSGI: Test route builder """
        def foo(): pass
        bottle.route('/a/:b/c', name='named')(foo)
        bottle.request.environ['SCRIPT_NAME'] = ''
        self.assertEqual('/a/xxx/c', bottle.url('named', b='xxx'))
        self.assertEqual('/a/xxx/c', bottle.app().get_url('named', b='xxx'))
        bottle.request.environ['SCRIPT_NAME'] = '/app'
        self.assertEqual('/app/a/xxx/c', bottle.url('named', b='xxx'))
        bottle.request.environ['SCRIPT_NAME'] = '/app/'
        self.assertEqual('/app/a/xxx/c', bottle.url('named', b='xxx'))
        bottle.request.environ['SCRIPT_NAME'] = 'app/'
        self.assertEqual('/app/a/xxx/c', bottle.url('named', b='xxx'))

def test_autoroute(self):
        app = bottle.Bottle()
        def a(): pass
        def b(x): pass
        def c(x, y): pass
        def d(x, y=5): pass
        def e(x=5, y=6): pass
        self.assertEqual(['/a'],list(bottle.yieldroutes(a)))
        self.assertEqual(['/b/<x>'],list(bottle.yieldroutes(b)))
        self.assertEqual(['/c/<x>/<y>'],list(bottle.yieldroutes(c)))
        self.assertEqual(['/d/<x>','/d/<x>/<y>'],list(bottle.yieldroutes(d)))
        self.assertEqual(['/e','/e/<x>','/e/<x>/<y>'],list(bottle.yieldroutes(e)))

def test_module_shortcuts(self):
        for name in '''route get post put delete error mount
                       hook install uninstall'''.split():
            short = getattr(bottle, name)
            original = getattr(bottle.app(), name)
            self.assertWraps(short, original)

def test_module_shortcuts_with_different_name(self):
        self.assertWraps(bottle.url, bottle.app().get_url)

def start():
    #http://stackoverflow.com/questions/13760109/ec2-bottle-py-connection
    app.run(host='0.0.0.0', port=18080)
    logging.info('completed %.3f' % time.clock())

def start2(number_of_process=4):
    from tornado.wsgi import WSGIContainer
    from tornado.httpserver import HTTPServer
    from tornado.ioloop import IOLoop

    http_server = HTTPServer(WSGIContainer(app))
    http_server.listen(18080)
    IOLoop.instance().start()

    logging.info('completed %.3f' % time.clock())

def method_not_allowed(res):
    if request.method == 'OPTIONS':
        new_res = bottle.HTTPResponse()
        new_res.set_header('Access-Control-Allow-Origin', '*')
        new_res.set_header('Access-Control-Allow-Methods', 'GET, POST, DELETE, PUT')
        new_res.set_header('Access-Control-Allow-Headers', 'X-Session-Token, X-Impersonate-Username, Content-Type')
        return new_res
    res.headers['Allow'] += ', OPTIONS'
    return request.app.default_error_handler(res)

def log_message(self, format, *args):
        code = args[1]
        # Logging HTTP 200 responses is super verbose
        if code == '200':
            return
        # But any non 200 reponse is suspicious and should be logger
        logger.info(
            "%s %s\n" % (
                self.client_address[0], format % args
            )
        )

# TODO: do not spawn the API ourselves ; instead, create an application.wsgi file defining
# the WSGI app and let Apache or some other webserver serve it
# TODO: or at least, use a better webserver than the standard WSGIServer

def __init__(self, config_path, config, notifier, websocket_notifier, authentificator, health):
        app = bottle.app()
        app.config.load_config(config_path)
        engine = database.engine()
        plugin = sabottle.Plugin(
            engine,
            None,
            keyword='db',
            use_kwargs=True,
            create=False,
        )
        app.install(plugin)
        self._check_for_index_html(app)
        conn = beanstalkc.Connection(
            host=config.get("general", "beanstalk_host"),
            port=11300
        )
        conn.use('deployer-deployments')
        app.config["deployer.engine"] = engine
        app.config["deployer.beanstalk"] = conn
        app.config["deployer.notifier"] = notifier
        app.config["deployer.websocket_notifier"] = websocket_notifier
        app.config["deployer.bcrypt_log_rounds"] = 12
        app.config["deployer.authentificator"] = authentificator
        app.config["health"] = health
        self.httpd = make_server("0.0.0.0", config.getint('general', 'api_port'), app,
                                 server_class=ThreadingWSGIServer,
                                 handler_class=LoggingWSGIRequestHandler)

def _check_for_index_html(self, app):
        index_path = os.path.abspath(os.path.join(app.config['general.web_path'], "html", "index.html"))
        if not os.path.exists(index_path):
            raise ValueError("Could not find the index.html file at {}, check your configuration."
                             .format(index_path))

def appjs():
    return bottle.static_file('app.js', root=script_path)

def run_webserver():
    app = bottle.app()
    app.catchall = False
    app = Sentry(app, raven.Client(SENTRY_DSN))
    bottle.run(app, server='paste', host='0.0.0.0', port=3350)

def index():
    if is_cli():
        bottle.response.content_type = 'text/plain'
        lines = []
        for klass in sorted(device_list.keys()):
            klass_devices = device_list[klass]
            for name in sorted(klass_devices.keys()):
                device = klass_devices[name]
                path = device._state_path()
                state = device.get_state()
                if state is None:
                    state = "Unknown"
                lines.append("%s - %s" % (path, state))
                for name,subdev in device.subdevices.items():
                    path = subdev._state_path()
                    state = subdev.get_state()
                    if state is None:
                        state = "Unknown"
                    lines.append("%s - %s" % (path, state))
        lines.append('')
        return '\n'.join(lines)
    else:
        page = markup.page()
        page.init(script=['app.js'], css=['style.css'])
        for klass in sorted(device_list.keys()):
            klass_devices = device_list[klass]
            page.h2(klass)
            for name in sorted(klass_devices.keys()):
                device = klass_devices[name]
                path = device._state_path()
                state = device.get_state()
                if state is None:
                    state = "Unknown"
                page.li.open()
                page.span("%s - " % (path,))
                page.span(state, id='%s-state'%(path,), class_='state')
                page.li.close()
                for name,subdev in device.subdevices.items():
                    path = subdev._state_path()
                    state = subdev.get_state()
                    if state is None:
                        state = "Unknown"
                    page.li.open()
                    page.span("%s - " % (path,))
                    page.span(state, id='%s-state'%(path,), class_='state')
                    page.li.close()
        page.br()
        page.a('Example configurations', href='examples')
        return str(page)