Source code for biothings.web.applications

""" 
    Biothings Web Applications -
    
    define the routes and handlers a supported web framework would consume
    basing on a config file, typically named `config.py`, enhanced by
    :py:mod:`biothings.web.settings.configs`. 
    
    The currently supported web frameworks are
    `Tornado <https://www.tornadoweb.org/en/stable/index.html>`_,
    `Flask <https://flask.palletsprojects.com/en/2.0.x/>`_, and
    `FastAPI <https://fastapi.tiangolo.com/>`_.

    The :py:mod:`biothings.web.launcher` can start the compatible HTTP servers
    basing on their interface. And the web applications delegate routes defined
    in the config file to handlers typically in :py:mod:`biothings.web.handlers`. 

    +----------------+------------+--------------------------------+
    | Web Framework  | Interface  | Handlers                       |
    +================+============+================================+
    | Tornado        | Tornado    | biothings.web.handlers.*       |
    +----------------+------------+--------------------------------+
    | Flask          | WSGI       | biothings.web.handlers._flask  |
    +----------------+------------+--------------------------------+
    | FastAPI        | ASGI       | biothings.web.handlers._fastapi|
    +----------------+------------+--------------------------------+

"""

import inspect
import logging
from pprint import pformat
from pydoc import locate
from types import SimpleNamespace

import tornado.httpserver
import tornado.ioloop
import tornado.log
import tornado.web
from biothings.web.handlers import BaseAPIHandler, BaseQueryHandler
from biothings.web.settings import configs
from biothings.web.services.namespace import BiothingsNamespace

try:
    from raven.contrib.tornado import AsyncSentryClient
except ImportError:
    __SENTRY_INSTALLED__ = False
else:
    __SENTRY_INSTALLED__ = True

logger = logging.getLogger(__name__)


[docs]def load_class(kls): if inspect.isclass(kls): return kls if isinstance(kls, str): return locate(kls) raise ValueError()
[docs]class TornadoBiothingsAPI(tornado.web.Application): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.biothings = SimpleNamespace() @staticmethod def _get_settings(biothings, override=None): """ Generates settings for tornado.web.Application. This result and the method below can define a tornado application to start a web server. """ settings = { 'biothings': biothings.config, 'autoreload': False, 'debug': False, } settings.update(override or {}) supported_keywords = ( 'default_handler_class', 'default_handler_args', 'template_path', 'log_function', 'compress_response', 'cookie_secret', 'login_url', 'static_path', 'static_url_prefix') for setting in supported_keywords: if hasattr(biothings.config, setting.upper()): if setting in settings: logging.warning( "Override config setting %s to %s.", setting, settings[setting]) continue settings[setting] = getattr(biothings.config, setting.upper()) if __SENTRY_INSTALLED__ and biothings.config.SENTRY_CLIENT_KEY: # Setup error monitoring with Sentry. More on: # https://docs.sentry.io/clients/python/integrations/#tornado settings['sentry_client'] = AsyncSentryClient(biothings.config.SENTRY_CLIENT_KEY) return settings @staticmethod def _get_handlers(biothings, addons=None): ''' Generates the tornado.web.Application `(regex, handler_class, options) tuples <http://www.tornadoweb.org/en/stable/web.html#application-configuration>`_. ''' handlers = {} addons = addons or [] for rule in biothings.config.APP_LIST + addons: pattern = rule[0] handler = load_class(rule[1]) setting = rule[2] if len(rule) == 3 else {} assert handler, rule[1] if '{typ}' in pattern or '{tps}' in pattern: if not issubclass(handler, BaseQueryHandler): raise TypeError("Not a biothing_type-aware handler.") if '{tps}' in pattern and len(biothings.metadata.types) <= 1: continue # '{tps}' routes only valid for multi-type apps for biothing_type in biothings.metadata.types: _pattern = pattern.format( pre=biothings.config.APP_PREFIX, ver=biothings.config.APP_VERSION, typ=biothing_type, tps=biothing_type ).replace('//', '/') _setting = dict(setting) _setting['biothing_type'] = biothing_type handlers[_pattern] = (_pattern, handler, _setting) elif '{pre}' in pattern or '{ver}' in pattern: pattern = pattern.format( pre=biothings.config.APP_PREFIX, ver=biothings.config.APP_VERSION).replace('//', '/') if '()' not in pattern: handlers[pattern] = (pattern, handler, setting) else: # no pattern translation handlers[pattern] = (pattern, handler, setting) handlers = list(handlers.values()) logger.info('API Handlers:\n%s', pformat(handlers, width=200)) return handlers
[docs] @classmethod def get_app(cls, config, settings=None, handlers=None): """ Return the tornado.web.Application defined by this config. **Additional** settings and handlers are accepted as parameters. """ if isinstance(config, configs.ConfigModule): biothings = BiothingsNamespace(config) _handlers = BiothingsAPI._get_handlers(biothings, handlers) _settings = BiothingsAPI._get_settings(biothings, settings) app = cls(_handlers, **_settings) app.biothings = biothings app._populate_optionsets(config, _handlers) app._populate_handlers(_handlers) return app if isinstance(config, configs.ConfigPackage): biothings = BiothingsNamespace(config.root) _handlers = [(f'/{c.APP_PREFIX}/.*', cls.get_app(c, settings)) for c in config.modules] _settings = BiothingsAPI._get_settings(biothings, settings) app = cls(_handlers + handlers or [], **_settings) app.biothings = biothings app._populate_optionsets(config, handlers or []) app._populate_handlers(handlers or []) return app raise TypeError()
def _populate_optionsets(self, config, handlers): for handler in handlers: handler = handler[1] # handler[0] is a matching pattern if issubclass(handler, BaseAPIHandler) and handler.name: handler_name = handler.name handler_options = handler.kwargs setting_attr = '_'.join((handler_name, 'kwargs')).upper() setting_options = getattr(config, setting_attr, {}) self.biothings.optionsets.add(handler_name, setting_options) self.biothings.optionsets.add(handler_name, handler_options) def _populate_handlers(self, handlers): for handler in handlers: self.biothings.handlers[handler[0]] = handler[1]
try: from flask import Flask class FlaskBiothingsAPI(Flask): @classmethod def get_app(cls, config): app = cls(__name__) app.config['JSON_SORT_KEYS'] = False app.url_map.strict_slashes = False app.biothings = BiothingsNamespace(config) from biothings.web.handlers._flask import routes for route in routes: setting_attr = '_'.join((route.name, 'kwargs')).upper() setting_options = getattr(config, setting_attr, {}) app.biothings.optionsets.add(route.name, setting_options) if isinstance(route.pattern, str): route.pattern = [route.pattern] for pattern in route.pattern: if '{typ}' in pattern: assert len(app.biothings.metadata.types) == 1, ( "Currently Biothings API on Flask only " "supports single biothings_type configuration." ) pattern = pattern.replace('{typ}', app.biothings.metadata.types[0]) pattern = pattern.replace('{ver}', app.biothings.config.APP_VERSION) app.add_url_rule(pattern, route.name, route, methods=route.methods) app.biothings.handlers[pattern] = route return app except Exception as exc:
[docs] class FlaskBiothingsAPI():
[docs] @classmethod def get_app(cls, config): raise exc
try: from fastapi import FastAPI from fastapi.middleware.wsgi import WSGIMiddleware class FastAPIBiothingsAPI(FastAPI): @classmethod def get_app(cls, config): app = cls() app.mount("/", WSGIMiddleware(FlaskBiothingsAPI.get_app(config))) return app # Native Implementation # ------------------------------------------------------------------------- # class FastAPIBiothingsAPI(FastAPI): # @classmethod # def get_app(cls, config): # from biothings.web.handlers import _fastapi # _fastapi.biothings = BiothingsNamespace(config) # app = cls() # for route in _fastapi.routes: # setting_attr = '_'.join((route.name, 'kwargs')).upper() # setting_options = getattr(config, setting_attr, {}) # _fastapi.biothings.optionsets.add(route.name, setting_options) # app.get(*route.args, **route.kwargs)(route) # return app except Exception as exc:
[docs] class FastAPIBiothingsAPI():
[docs] @classmethod def get_app(cls, config): raise exc
BiothingsAPI = TornadoBiothingsAPI # default