def create_app_with_config()

in services/sse-api/src/sse_api/app.py [0:0]

• 35 lines of code
• 2 McCabe index (conditional complexity)

def create_app_with_config(app_config: AppConfig) -> Starlette:
    init_logging(level=app_config.log.level)
    # ^ set first to have logs as soon as possible

    # ensure the collection has changeStreamPreAndPostImages enabled (required to report the delete events)
    with CacheMongoResource(database=app_config.cache.mongo_database, host=app_config.cache.mongo_url) as resource:
        if not resource.is_available():
            raise Exception("MongoDB is not available")
        resource.create_collection(CachedResponseDocument)
        resource.enable_pre_and_post_images(CACHE_COLLECTION_RESPONSES)

    hub_cache_watcher = HubCacheWatcher(
        client=AsyncIOMotorClient(host=app_config.cache.mongo_url, io_loop=asyncio.get_running_loop()),
        db_name=app_config.cache.mongo_database,
        collection_name=CACHE_COLLECTION_RESPONSES,
    )

    middleware = [
        Middleware(
            CORSMiddleware,
            allow_origins=["*"],
            allow_methods=["*"],
            allow_headers=["*"],
            allow_credentials=True,
            expose_headers=EXPOSED_HEADERS,
        ),
        # https://github.com/sysid/sse-starlette
        # > Caveat: SSE streaming does not work in combination with GZipMiddleware.
        Middleware(PrometheusMiddleware, filter_unhandled_paths=True),
    ]

    routes = [
        Route("/sse/hub-cache", endpoint=create_hub_cache_endpoint(hub_cache_watcher=hub_cache_watcher)),
        Route("/healthcheck", endpoint=healthcheck_endpoint),
        # ^ called by ALB
        Route("/sse/healthcheck", endpoint=healthcheck_endpoint),
        # ^ called by Kubernetes
        Route("/sse/metrics", endpoint=create_metrics_endpoint()),
        # ^ called by Prometheus
    ]

    return Starlette(
        routes=routes,
        middleware=middleware,
        on_startup=[hub_cache_watcher.start_watching],
        on_shutdown=[hub_cache_watcher.stop_watching],
    )