def app()

in services/ui_backend_service/ui_server.py [0:0]

• 64 lines of code
• 12 McCabe index (conditional complexity)

def app(loop=None, db_conf: DBConfiguration = None):

    loop = loop or asyncio.get_event_loop()
    _app = web.Application(loop=loop)
    app = web.Application(loop=loop) if len(PATH_PREFIX) > 0 else _app

    async_db = AsyncPostgresDB('ui')
    loop.run_until_complete(async_db._init(db_conf=db_conf, create_triggers=DB_TRIGGER_CREATE))

    event_emitter = AsyncIOEventEmitter()

    async_db_cache = AsyncPostgresDB('ui:cache')
    loop.run_until_complete(async_db_cache._init(db_conf))
    cache_store = CacheStore(app=app, db=async_db_cache, event_emitter=event_emitter)

    if FEATURE_DB_LISTEN_ENABLE:
        async_db_notify = AsyncPostgresDB('ui:notify')
        loop.run_until_complete(async_db_notify._init(db_conf))
        ListenNotify(app, db=async_db_notify, event_emitter=event_emitter)

    if FEATURE_HEARTBEAT_ENABLE:
        async_db_heartbeat = AsyncPostgresDB('ui:heartbeat')
        loop.run_until_complete(async_db_heartbeat._init(db_conf))
        RunHeartbeatMonitor(event_emitter, db=async_db_heartbeat)
        TaskHeartbeatMonitor(event_emitter, db=async_db_heartbeat, cache=cache_store)

    if FEATURE_WS_ENABLE:
        async_db_ws = AsyncPostgresDB('ui:websocket')
        loop.run_until_complete(async_db_ws._init(db_conf))
        Websocket(app, db=async_db_ws, event_emitter=event_emitter, cache=cache_store)

    AutoCompleteApi(app, async_db)
    FlowApi(app, async_db)
    RunApi(app, async_db, cache_store)
    StepApi(app, async_db)
    TaskApi(app, async_db, cache_store)
    MetadataApi(app, async_db)
    ArtificatsApi(app, async_db, cache_store)
    TagApi(app, async_db)
    SearchApi(app, async_db, cache_store)
    DagApi(app, async_db, cache_store)
    FeaturesApi(app)
    ConfigApi(app)
    PluginsApi(app)
    CardsApi(app, async_db, cache_store)

    LogApi(app, async_db, cache_store)
    AdminApi(app, cache_store)

    # Add Metadata Service as a sub application so that Metaflow Client
    # can use it as a service backend in case none provided via METAFLOW_SERVICE_URL
    #
    # Metadata service exposed through UI service is intended for read-only use only.
    # 'allow_get_requests_only' middleware will only accept GET requests.
    app.add_subapp("/metadata", metadata_service_app(
        loop=loop, db_conf=db_conf, middlewares=[allow_get_requests_only]))

    if os.environ.get("UI_ENABLED", "1") == "1":
        # Serve UI bundle only if enabled
        # This has to be placed last due to catch-all route
        Frontend(app)

    if len(PATH_PREFIX) > 0:
        _app.add_subapp(PATH_PREFIX, app)

    if ORIGIN_TO_ALLOW_CORS_FROM:
        import aiohttp_cors
        logging.info("We will allows CORS from the origin %s" % ORIGIN_TO_ALLOW_CORS_FROM)
        cors = aiohttp_cors.setup(_app, defaults={
            ORIGIN_TO_ALLOW_CORS_FROM: aiohttp_cors.ResourceOptions(
                allow_credentials=True,
                expose_headers="*",
                allow_headers="*",
                allow_methods="*",
            )
        })
        # Configure CORS on all routes.
        for route in list(_app.router.routes()):
            cors.add(route)

    logging.info("Metadata service available at {}".format(DEFAULT_METADATA_SERVICE_URL))

    async def _init_plugins():
        with concurrent.futures.ThreadPoolExecutor() as pool:
            await loop.run_in_executor(pool, init_plugins)
    asyncio.run_coroutine_threadsafe(_init_plugins(), loop)

    return _app