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
python类Bottle()的实例源码
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 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 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 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
