我们从Python开源项目中，提取了以下35个代码示例，用于说明如何使用bottle.Bottle()。
def __init__(self, server_settings): """ Loads a translation model and initialises the webserver. @param server_settings: see `settings.py` """ self._style = server_settings.style self._host = server_settings.host self._port = server_settings.port self._threads = server_settings.threads self._debug = server_settings.verbose self._models = server_settings.models self._num_processes = server_settings.num_processes self._device_list = server_settings.device_list self._status = self.STATUS_LOADING # start webserver self._server = Bottle() self._server.config['logging.level'] = 'DEBUG' if server_settings.verbose else 'WARNING' self._server.config['logging.format'] = '%(levelname)s: %(message)s' self._server.install(LoggingPlugin(self._server.config)) logging.info("Starting Nematus Server") # start translation workers logging.info("Loading translation models") self._translator = Translator(server_settings) self._status = self.STATUS_OK
def set_config(self, config_instance): '''Set the config instance. By default, this is an instance of :class:`memex_dossier.web.Config`, which provides services like ``kvlclient`` and ``label_store``. Custom services should probably subclass :class:`memex_dossier.web.Config`, but it's not strictly necessary so long as it provides the same set of services (which are used for dependency injection into Bottle routes). :param config_instance: A config instance. :type config_instance: :class:`memex_dossier.web.Config` :rtype: :class:`WebBuilder` ''' self.config = config_instance return self
def add_routes(self, routes): '''Merges a Bottle application into this one. :param routes: A Bottle application or a sequence of routes. :type routes: :class:`bottle.Bottle` or `[bottle route]`. :rtype: :class:`WebBuilder` ''' # Basically the same as `self.app.merge(routes)`, except this # changes the owner of the route so that plugins on `self.app` # apply to the routes given here. if isinstance(routes, bottle.Bottle): routes = routes.routes for route in routes: route.app = self.app self.app.add_route(route) return self
def view_wrapper(fn): """Bottle view wrapper""" def _inner(): kwargs = json.loads(request.forms.get('data')) try: response = { 'data': fn(**kwargs), 'status': 'OK' } except: response = {'status': 'ERROR', 'data': {}} return response return _inner
def make_app(*apps): """Make a bottle app from the instance""" app = Bottle() for mount, instance in apps: sub_app = Bottle() for method_name in get_methods(instance): method = getattr(instance, method_name) route = '/%s/' % method_name sub_app.route(route, method='POST')(view_wrapper(method)) app.mount(mount, sub_app) return app
def new_access(): bottle_app.oauth = setup_oauth(r_c, conf_obj, store_tokens_callback) secret_keys = r_c.get('secret_keys') if not secret_keys: secret_keys = [] else: secret_keys = json.loads(str(secret_keys, encoding='utf-8', errors='strict')) if str(bottle.request.POST.get('diycrate_secret_key')) not in secret_keys: raise ValueError('No matching secret key; we are being secure!') return json.dumps([el.decode(encoding='utf-8', errors='strict') if isinstance(el, bytes) else el for el in bottle_app.oauth.refresh(bottle.request.POST.get('access_token'))]) # Create our own sub-class of Bottle's ServerAdapter # so that we can specify SSL. Using just server='cherrypy' # uses the default cherrypy server, which doesn't use SSL
def start(): if conf.quiet: print('Bottle v%s server starting up (using WSGIRefServer())...' % __version__) print('Listening on http://127.0.0.1:%s/' % conf.port) print('Hit Ctrl-C to quit.') app.run(host='127.0.0.1', port=conf.port, quiet=conf.quiet) # start bottle
def __init__(self, api_root=None, access_token=None, secret=None): super().__init__(api_root, access_token) self._secret = secret self._handlers = defaultdict(dict) self._app = Bottle() self._app.post('/')(self._handle) self._groups = [] self.on_message = self._deco_maker('message') self.on_event = self._deco_maker('event') self.on_request = self._deco_maker('request')
def setUp(self): # provide a dummy tracer self.tracer = get_dummy_tracer() self._original_tracer = ddtrace.tracer ddtrace.tracer = self.tracer # provide a Bottle app self.app = bottle.Bottle()
def patch(): """Patch the bottle.Bottle class """ if getattr(bottle, '_datadog_patch', False): return setattr(bottle, '_datadog_patch', True) wrapt.wrap_function_wrapper('bottle', 'Bottle.__init__', traced_init)
def __init__(self, context=None, httpd=None, route=None, routes=None, check=False): """ Serves web requests :param context: global context for this process :type context: Context :param httpd: actual WSGI server :param route: a route to add to this instance :type route: Route :param routes: multiple routes to add to this instance :type routes: list of Route :param check: True to check configuration settings :type check: bool """ self.context = Context() if context is None else context self.httpd = Bottle() if httpd is None else httpd self._routes = {} if route is not None: self.add_route(route) if routes is not None: self.add_routes(routes) if check: self.configure()
def start(self): self.__app = Bottle() self.__app.route('/api/tts', method="GET", callback=self.__text_to_speech) self.__app.run(host=self.__config.host, port=self.__config.port)
def setUp(self): e = dict() wsgiref.util.setup_testing_defaults(e) b = Bottle() request.bind(e) response.bind()
def setUp(self): ServerTestBase.setUp(self) self.subapp = bottle.Bottle() @self.subapp.route('/') @self.subapp.route('/test/:test') def test(test='foo'): return test
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 __init__(self, issuer_cert: str, responder_cert: str, responder_key: str, validate_func: ValidateFunc, cert_retrieve_func: CertRetrieveFunc, next_update_days: int = 7): """ Create a new OCSPResponder instance. :param issuer_cert: Path to the issuer certificate. :param responder_cert: Path to the certificate of the OCSP responder with the `OCSP Signing` extension. :param responder_key: Path to the private key belonging to the responder cert. :param validate_func: A function that - given a certificate serial - will return the appropriate :class:`CertificateStatus` and - depending on the status - a revocation datetime. :param cert_retrieve_func: A function that - given a certificate serial - will return the corresponding certificate as a string. :param next_update_days: The ``nextUpdate`` value that will be written into the response. Default: 7 days. """ # Certs and keys self._issuer_cert = asymmetric.load_certificate(issuer_cert) self._responder_cert = asymmetric.load_certificate(responder_cert) self._responder_key = asymmetric.load_private_key(responder_key) # Functions self._validate = validate_func self._cert_retrieve = cert_retrieve_func # Next update self._next_update_days = next_update_days # Bottle self._app = Bottle() # Initialize routing self._route()
def __init__(self, quiet=True, debug=False, **kwargs): super(SwitchboardClient, self).__init__(**kwargs) self._quiet = quiet self._debug = debug self._app = Bottle() self._app.route('/devices_info', method='GET', callback=self._devices_info) self._app.route('/devices_value', method='GET', callback=self._devices_value) self._app.route('/device_set', method='PUT', callback=self._device_set)
def __init__(self): self._app = Bottle() self._app.route('/', method='GET', callback=self._index) self._app.route('/websocket', method='GET', callback=self._websocket_connection, apply=[websocket]) self._clients = set() self._io_state_table = []
def test_hello_world(self): app = bottle.Bottle() br = bottlereact.BottleReact(app, prod=True) html = br.render_html(br.HelloWorld()) self.assertTrue(html.startswith('<html>')) self.assertTrue('-bottlereact.js"></script>' in html) self.assertTrue('-hello_world.js"></script>' in html) self.assertTrue('React.createElement(bottlereact.HelloWorld,{},[])' in html)
def test_kwarg(self): app = bottle.Bottle() br = bottlereact.BottleReact(app, prod=True) html = br.render_html(br.HelloWorld(), template='title', title='xyz').strip() self.assertEqual(html, 'xyz')
def test_default_kwarg(self): app = bottle.Bottle() br = bottlereact.BottleReact(app, prod=True, default_render_html_kwargs={'title':'abc'}) html = br.render_html(br.HelloWorld(), template='title').strip() self.assertEqual(html, 'abc') html = br.render_html(br.HelloWorld(), template='title', title='xyz').strip() self.assertEqual(html, 'xyz')
def test_default_kwarg_func(self): def default_render_html_kwargs(): return {'title':'abc'} app = bottle.Bottle() br = bottlereact.BottleReact(app, prod=True, default_render_html_kwargs=default_render_html_kwargs) html = br.render_html(br.HelloWorld(), template='title').strip() self.assertEqual(html, 'abc') html = br.render_html(br.HelloWorld(), template='title', title='xyz').strip() self.assertEqual(html, 'xyz')
def test_hello_world_server_side_render(self): app = bottle.Bottle() br = bottlereact.BottleReact(app, prod=True) html = br.render_html(br.HelloWorld(), render_server=True) self.assertTrue('Thanks for trying' in html)
def test_hello_world_server_side_render_died(self): app = bottle.Bottle() br = bottlereact.BottleReact(app, prod=True) html = br.render_html(br.HelloWorld(), render_server=True) self.assertTrue('Thanks for trying' in html) self.assertEqual(len(br._ctxs), 1) port, child = list(br._ctxs.values())[0] child.kill() html = br.render_html(br.HelloWorld(), render_server=True) self.assertEqual(len(br._ctxs), 1)
def __init__(self, add_default_routes=True): '''Introduce a new builder. You can use method chaining to configure your web application options. e.g., .. code-block:: python app = WebBuilder().enable_cors().get_app() app.run() This code will create a new Bottle web application that enables CORS (Cross Origin Resource Sharing). If ``add_default_routes`` is ``False``, then the default set of routes in ``memex_dossier.web.routes`` is not added. This is only useful if you want to compose multiple Bottle applications constructed through multiple instances of ``WebBuilder``. ''' self.app = bottle.Bottle() self.search_engines = { 'random': builtin_engines.random, 'plain_index_scan': builtin_engines.plain_index_scan, } self.filters = { 'already_labeled': already_labeled, } self.mount_prefix = None self.config = None if add_default_routes: self.add_routes(default_app) self.add_routes(tags_app) # DEPRECATED. Remove. ---AG self.visid_to_dbid, self.dbid_to_visid = lambda x: x, lambda x: x
def create_injector(param_name, fun_param_value): '''Dependency injection with Bottle. This creates a simple dependency injector that will map ``param_name`` in routes to the value ``fun_param_value()`` each time the route is invoked. ``fun_param_value`` is a closure so that it is lazily evaluated. This is useful for handling thread local services like database connections. :param str param_name: name of function parameter to inject into :param fun_param_value: the value to insert :type fun_param_value: a closure that can be applied with zero arguments ''' class _(object): api = 2 def apply(self, callback, route): if param_name not in inspect.getargspec(route.callback)[0]: return callback def _(*args, **kwargs): pval = fun_param_value() if pval is None: logger.error('service "%s" unavailable', param_name) bottle.abort(503, 'service "%s" unavailable' % param_name) return kwargs[param_name] = pval return callback(*args, **kwargs) return _ return _()
def log_to_logger(fn): """ Wrap a Bottle request so that a log line is emitted after it's handled. """ @wraps(fn) def _log_to_logger(*args, **kwargs): actual_response = fn(*args, **kwargs) # modify this to log exactly what you need: logger.info('%s %s %s %s' % (bottle.request.remote_addr, bottle.request.method, bottle.request.url, bottle.response.status)) return actual_response return _log_to_logger
def __init__(self, host='', porta=8080): self._h = host self._p = porta self._a = Bottle() self._rota() self._g = core
def __init__(self, host, port): serv_host = FLAGS.host serv_port = FLAGS.port model_name = FLAGS.model_name model_version = FLAGS.model_version self.request_timeout = FLAGS.request_timeout config = tf.ConfigProto(log_device_placement=False) config.gpu_options.allow_growth = True self.sess = tf.Session(config=config) init = tf.global_variables_initializer() config = Config() self.woipv = WoipvPspNetModel(config) self.input_tensor = tf.placeholder(tf.string, name="input_tensor") processed = tf.cast(tf.decode_raw(self.input_tensor, tf.uint8), tf.float32) processed = tf.reshape(processed, [288, 288, 3]) self.image_op = processed processed = tf.image.per_image_standardization(processed) processed = tf.expand_dims(processed, axis=0) self.result_op = tf.nn.sigmoid(self.woipv.inference(processed)) self.sess.run(init) ckpt = tf.train.get_checkpoint_state("/training/woipv_train") ckpt_name = os.path.basename(ckpt.model_checkpoint_path) variables_to_restore = tf.global_variables() chkpt_saver = tf.train.Saver(variables_to_restore, write_version=tf.train.SaverDef.V2) chkpt_saver.restore(self.sess, ckpt.model_checkpoint_path) self._host = host self._port = port self._app = bottle.Bottle() self._route()
def create_app(engine): app = Bottle() @app.error() @app.error(404) def handle_error(error): if issubclass(type(error.exception), ApiException): response.status = error.exception.code else: response.status = error.status_code response.set_header('Content-type', 'application/json') resp = { 'type': type(error.exception).__name__, 'message': repr(error.exception) if error.exception else '', 'traceback': error.traceback, 'status_code': response.status } log.error('Exception, type=%s, message=%s, status_code=%s, traceback=%s'\ % (resp['type'], resp['message'], resp['status_code'], resp['traceback'])) return '%s %s' % (resp['status_code'], resp['message']) @app.route('/ping', method=['GET']) def ping(): return {'name': 'xFlow', 'version': '0.1' } @app.route('/publish', method=['POST']) def publish(): data = json.loads(request.body.read()) try: publish_schema.validate(data) except jsonschema.ValidationError as err: raise BadRequest(err) stream = data['stream'] event = json.dumps(data['event']) try: engine.publish(stream, event) except core.KinesisStreamDoesNotExist as ex: raise NotFoundException(str(ex)) return {} @app.route('/track/workflows/<workflow_id>/executions/<execution_id>', method=['GET']) def track(workflow_id, execution_id): try: tracking_info = engine.track(workflow_id, execution_id) return tracking_info except (core.CloudWatchStreamDoesNotExist, core.WorkflowDoesNotExist, core.CloudWatchLogDoesNotExist) as ex: raise NotFoundException(str(ex)) raise Exception("Something went wrong!") return app
def _start_server(self): class BottleServerAdapter(bottle.ServerAdapter): proxy = self def close_session(self): self.proxy.ctx.model.log._session.remove() def run(self, app): class Server(wsgiref.simple_server.WSGIServer): allow_reuse_address = True bottle_server = self def handle_error(self, request, client_address): pass def serve_forever(self, poll_interval=0.5): try: wsgiref.simple_server.WSGIServer.serve_forever(self, poll_interval) finally: # Once shutdown is called, we need to close the session. # If the session is not closed properly, it might raise warnings, # or even lock the database. self.bottle_server.close_session() class Handler(wsgiref.simple_server.WSGIRequestHandler): def address_string(self): return self.client_address[0] def log_request(*args, **kwargs): # pylint: disable=no-method-argument if not self.quiet: return wsgiref.simple_server.WSGIRequestHandler.log_request(*args, **kwargs) server = wsgiref.simple_server.make_server( host=self.host, port=self.port, app=app, server_class=Server, handler_class=Handler) self.proxy.server = server self.proxy._started.put(True) server.serve_forever(poll_interval=0.1) def serve(): # Since task is a thread_local object, we need to patch it inside the server thread. self._ctx_patcher(self.ctx) bottle_app = bottle.Bottle() bottle_app.post('/', callback=self._request_handler) bottle.run( app=bottle_app, host='localhost', port=self.port, quiet=True, server=BottleServerAdapter) thread = threading.Thread(target=serve) thread.daemon = True thread.start() return thread
def get_app(self): '''Eliminate the builder by producing a new Bottle application. This should be the final call in your method chain. It uses all of the built up options to create a new Bottle application. :rtype: :class:`bottle.Bottle` ''' if self.config is None: # If the user never sets a config instance, then just create # a default. self.config = Config() if self.mount_prefix is None: self.mount_prefix = self.config.config.get('url_prefix') self.inject('config', lambda: self.config) self.inject('kvlclient', lambda: self.config.kvlclient) self.inject('store', lambda: self.config.store) self.inject('label_store', lambda: self.config.label_store) self.inject('tags', lambda: self.config.tags) self.inject('search_engines', lambda: self.search_engines) self.inject('filters', lambda: self.filters) self.inject('request', lambda: bottle.request) self.inject('response', lambda: bottle.response) # DEPRECATED. Remove. ---AG self.inject('visid_to_dbid', lambda: self.visid_to_dbid) self.inject('dbid_to_visid', lambda: self.dbid_to_visid) # Also DEPRECATED. self.inject('label_hooks', lambda: []) # Load routes defined in entry points. for extroute in self.config.config.get('external_routes', []): mod, fun_name = extroute.split(':') logger.info('Loading external route: %s', extroute) fun = getattr(__import__(mod, fromlist=[fun_name]), fun_name) self.add_routes(fun()) # This adds the `json=True` feature on routes, which always coerces # the output to JSON. Bottle, by default, only permits dictionaries # to be JSON, which is the correct behavior. (Because returning JSON # arrays is a hazard.) # # So we should fix the routes and then remove this. ---AG self.app.install(JsonPlugin()) # Throw away the app and return it. Because this is elimination! app = self.app self.app = None if self.mount_prefix is not None: root = bottle.Bottle() root.mount(self.mount_prefix, app) return root else: return app
