opal_server/server.py
def _configure_api_routes(self, app: FastAPI):
        """
        mounts the api routes on the app object
        """
        authenticator = JWTAuthenticator(self.signer)

        data_update_publisher: Optional[DataUpdatePublisher] = None
        if self.publisher is not None:
            data_update_publisher = DataUpdatePublisher(self.publisher)

        # Init api routers with required dependencies
        data_updates_router = init_data_updates_router(
            data_update_publisher,
            self.data_sources_config,
            authenticator
        )
        webhook_router = init_git_webhook_router(self.pubsub.endpoint, authenticator)
        security_router = init_security_router(self.signer, StaticBearerAuthenticator(self.master_token))

        # mount the api routes on the app object
        app.include_router(bundles_router, tags=["Bundle Server"], dependencies=[Depends(authenticator)])
        app.include_router(data_updates_router, tags=["Data Updates"], dependencies=[Depends(authenticator)])
        app.include_router(webhook_router, tags=["Github Webhook"])
        app.include_router(security_router, tags=["Security"])
        app.include_router(self.pubsub.router, tags=["Pub/Sub"])

        if self.jwks_endpoint is not None:
            # mount jwts (static) route
            self.jwks_endpoint.configure_app(app)

        # top level routes (i.e: healthchecks)
        @app.get("/healthcheck", include_in_schema=False)
        @app.get("/", include_in_schema=False)
        def healthcheck():
            return {"status": "ok"}

        return app

        mounts the api routes on the app object
        
mounts the api routes on the app object
mounts the api routes on the app object
def _configure_api_routes(self, app: FastAPI):
        authenticator = JWTAuthenticator(self.signer)
        data_update_publisher: Optional[DataUpdatePublisher] = None
        if self.publisher is not None:
            data_update_publisher = DataUpdatePublisher(self.publisher)
        data_updates_router = init_data_updates_router(
            data_update_publisher,
            self.data_sources_config,
            authenticator
        )
        webhook_router = init_git_webhook_router(self.pubsub.endpoint, authenticator)
        security_router = init_security_router(self.signer, StaticBearerAuthenticator(self.master_token))
        app.include_router(bundles_router, tags=["Bundle Server"], dependencies=[Depends(authenticator)])
        app.include_router(data_updates_router, tags=["Data Updates"], dependencies=[Depends(authenticator)])
        app.include_router(webhook_router, tags=["Github Webhook"])
        app.include_router(security_router, tags=["Security"])
        app.include_router(self.pubsub.router, tags=["Pub/Sub"])
        if self.jwks_endpoint is not None:
            self.jwks_endpoint.configure_app(app)
        @app.get("/healthcheck", include_in_schema=False)
        @app.get("/", include_in_schema=False)
        def healthcheck():
            return {"status": "ok"}
        return app
def _configure_api_routes ( self , app : FastAPI ) : authenticator = JWTAuthenticator ( self . signer ) data_update_publisher : Optional [ DataUpdatePublisher ] = None if self . publisher is not None : data_update_publisher = DataUpdatePublisher ( self . publisher ) data_updates_router = init_data_updates_router ( data_update_publisher , self . data_sources_config , authenticator ) webhook_router = init_git_webhook_router ( self . pubsub . endpoint , authenticator ) security_router = init_security_router ( self . signer , StaticBearerAuthenticator ( self . master_token ) ) app . include_router ( bundles_router , tags = [ "Bundle Server" ] , dependencies = [ Depends ( authenticator ) ] ) app . include_router ( data_updates_router , tags = [ "Data Updates" ] , dependencies = [ Depends ( authenticator ) ] ) app . include_router ( webhook_router , tags = [ "Github Webhook" ] ) app . include_router ( security_router , tags = [ "Security" ] ) app . include_router ( self . pubsub . router , tags = [ "Pub/Sub" ] ) if self . jwks_endpoint is not None : self . jwks_endpoint . configure_app ( app ) @ app . get ( "/healthcheck" , include_in_schema = False ) @ app . get ( "/" , include_in_schema = False ) def healthcheck ( ) : return { "status" : "ok" } return app
mounts the api routes on the app object
mounts the api routes on the app object
"""
        mounts the api routes on the app object
        """ # Init api routers with required dependencies # mount the api routes on the app object # mount jwts (static) route # top level routes (i.e: healthchecks)
self
app
FastAPI