Async & concurrent network startup

bento_beacon/app.py

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 import os
 from flask import Flask, current_app, request
@@ -10,7 +11,6 @@
 from .endpoints.cohorts import cohorts
 from .endpoints.datasets import datasets
 from .network.network import network
-from .network.utils import init_network_service_registry
 from .utils.exceptions import APIException
 from werkzeug.exceptions import HTTPException
 from .authz.middleware import authz_middleware
@@ -65,11 +65,6 @@
     # load blueprint for network
     if current_app.config["USE_BEACON_NETWORK"]:
         app.register_blueprint(network)
-        try:
-            init_network_service_registry()
-        except APIException:
-            # trouble setting up network, swallow for now
-            current_app.logger.error("API Error when initializing beacon network")
 
     # get censorship settings from katsu
     max_filters = None
@@ -79,7 +74,7 @@
     for tries in range(max_retries + 1):
         current_app.logger.info(f"calling katsu for censorship parameters (try={tries})")
         try:
-            max_filters, count_threshold = katsu_censorship_settings()
+            max_filters, count_threshold = asyncio.run(katsu_censorship_settings())
             # If we got values successfully, without an API exception being raised, exit early - even if they're None
             break
         except APIException as e:
@@ -102,11 +97,11 @@
 
 
 @app.before_request
-def before_request():
+async def before_request():
     if request.blueprint != "info":
         validate_request()
-        verify_permissions()
-        save_request_data()
+        await verify_permissions()
+        await save_request_data()
         reject_query_if_not_permitted()
         init_response_data()
 

bento_beacon/authz/access.py

@@ -1,66 +1,72 @@
-import functools
-import requests
+import aiocache
+import aiohttp
 from flask import current_app
-
 from .headers import auth_header_from_request
+from ..utils.http import tcp_connector
 
 __all__ = [
     "get_access_token",
     "create_access_header_or_fall_back",
 ]
 
 
-@functools.cache
-def get_token_endpoint_from_openid_config_url(url: str, validate_ssl: bool = True):
-    r = requests.get(url, verify=validate_ssl)
+@aiocache.cached()
+async def get_token_endpoint_from_openid_config_url(url: str):
+    async with aiohttp.ClientSession(connector=tcp_connector(current_app.config)) as s:
+        r = await s.get(url)
+
     if not r.ok:
         raise Exception(f"Received not-OK response from OIDC config URL: {r.status_code}")
-    return r.json()["token_endpoint"]
 
+    response = await r.json()
+    return response["token_endpoint"]
 
-def get_access_token() -> str | None:
+
+async def get_access_token() -> str | None:
     logger = current_app.logger
 
     oidc_config_url = current_app.config["OPENID_CONFIG_URL"]
     client_id = current_app.config["CLIENT_ID"]
     client_secret = current_app.config["CLIENT_SECRET"]
-    validate_ssl = current_app.config["BENTO_VALIDATE_SSL"]
 
     if not all((oidc_config_url, client_id, client_secret)):
         logger.error("Could not retrieve access token; one of OPENID_CONFIG_URL | CLIENT_ID | CLIENT_SECRET is not set")
         return None
 
     try:
-        token_endpoint = get_token_endpoint_from_openid_config_url(oidc_config_url, validate_ssl=validate_ssl)
+        token_endpoint = await get_token_endpoint_from_openid_config_url(oidc_config_url)
+        current_app.logger.info(f"token_endpoint: {token_endpoint}")
     except Exception as e:
         logger.error(f"Could not retrieve access token; got exception from OpenID config URL: {e}")
         return None
 
-    token_res = requests.post(
-        token_endpoint,
-        verify=validate_ssl,
-        data={
-            "grant_type": "client_credentials",
-            "client_id": client_id,
-            "client_secret": client_secret,
-        },
-    )
+    async with aiohttp.ClientSession(connector=tcp_connector(current_app.config)) as s:
+        token_res = await s.post(
+            token_endpoint,
+            data={
+                "grant_type": "client_credentials",
+                "client_id": client_id,
+                "client_secret": client_secret,
+            },
+        )
+
+    res = await token_res.json()
 
     if not token_res.ok:
-        logger.error(f"Could not retrieve access token; got error response: {token_res.json()}")
+        logger.error(f"Could not retrieve access token; got error response: {res}")
         return None
 
-    return token_res.json()["access_token"]
+    return res["access_token"]
 
 
-def create_access_header_or_fall_back():
+async def create_access_header_or_fall_back():
     logger = current_app.logger
 
     if not current_app.config["AUTHZ_BENTO_REQUESTS_ENABLED"]:
         logger.warning("AUTHZ_BENTO_REQUESTS_ENABLED is false; falling back to request headers")
         return auth_header_from_request()
 
-    access_token = get_access_token()
+    access_token = await get_access_token()
     if access_token is None:
         logger.error("create_access_header_or_fall_back: falling back to request headers")
         return auth_header_from_request()

bento_beacon/authz/middleware.py

@@ -19,5 +19,5 @@
 )
 
 
-def check_permission(permission: Permission) -> bool:
-    return authz_middleware.evaluate_one(request, RESOURCE_EVERYTHING, permission, mark_authz_done=True)
+async def check_permission(permission: Permission) -> bool:
+    return await authz_middleware.async_evaluate_one(request, RESOURCE_EVERYTHING, permission, mark_authz_done=True)

bento_beacon/config_files/config.py

@@ -1,6 +1,6 @@
 import json
+import logging
 import os
-import urllib3
 from ..constants import GRANULARITY_COUNT, GRANULARITY_RECORD
 
 
@@ -11,13 +11,15 @@ def str_to_bool(value: str) -> bool:
     return value.strip().lower() in ("true", "1", "t", "yes")
 
 
+def reverse_domain_id(domain):
+    return ".".join(reversed(domain.split("."))) + ".beacon"
+
+
 BENTO_DEBUG = str_to_bool(os.environ.get("BENTO_DEBUG", os.environ.get("FLASK_DEBUG", "false")))
-BENTO_VALIDATE_SSL = str_to_bool(os.environ.get("BENTO_VALIDATE_SSL", str(not BENTO_DEBUG)))
 
-if not BENTO_VALIDATE_SSL:
-    # Don't let urllib3 spam us with SSL validation warnings if we're operating with SSL validation off, most likely in
-    # a development/test context where we're using self-signed certificates.
-    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+# silence logspam
+logging.getLogger("asyncio").setLevel(logging.WARNING)
+logging.getLogger("aiocache").setLevel(logging.WARNING)
 
 
 class Config:
@@ -40,15 +42,10 @@ class Config:
     DEFAULT_PAGINATION_PAGE_SIZE = 10
 
     BENTO_DEBUG = BENTO_DEBUG
-    BENTO_VALIDATE_SSL = BENTO_VALIDATE_SSL
-
     BENTO_DOMAIN = os.environ.get("BENTOV2_DOMAIN")
     BEACON_BASE_URL = os.environ.get("BEACON_BASE_URL")
     BENTO_PUBLIC_URL = os.environ.get("BENTOV2_PUBLIC_URL")
-
-    # reverse domain id
-    BEACON_ID = ".".join(reversed(BENTO_DOMAIN.split("."))) + ".beacon"
-
+    BEACON_ID = reverse_domain_id(BENTO_DOMAIN)
     BEACON_NAME = os.environ.get("BENTO_PUBLIC_CLIENT_NAME", "Bento") + " Beacon"
     BEACON_UI_ENABLED = str_to_bool(os.environ.get("BENTO_BEACON_UI_ENABLED", ""))
     BEACON_UI_URL = BENTO_PUBLIC_URL + "/#/en/beacon"
@@ -153,9 +150,6 @@ class Config:
     KATSU_DATASETS_ENDPOINT = "/api/datasets"
     KATSU_SEARCH_ENDPOINT = "/private/search"
     KATSU_RESOURCES_ENDPOINT = "/api/resources"
-    KATSU_PHENOTYPIC_FEATURE_TERMS_ENDPOINT = "/api/phenotypic_feature_type_autocomplete"
-    KATSU_DISEASES_TERMS_ENDPOINT = "/api/disease_term_autocomplete"
-    KATSU_SAMPLED_TISSUES_TERMS_ENDPOINT = "/api/biosample_sampled_tissue_autocomplete"
     KATSU_PUBLIC_CONFIG_ENDPOINT = "/api/public_search_fields"
     KATSU_INDIVIDUAL_SCHEMA_ENDPOINT = "/api/schemas/phenopacket"
     KATSU_EXPERIMENT_SCHEMA_ENDPOINT = "/api/schemas/experiment"

bento_beacon/endpoints/datasets.py

@@ -9,16 +9,16 @@
 
 @datasets.route("/datasets", methods=["GET", "POST"])
 @authz_middleware.deco_public_endpoint  # TODO: authz - more flexibility in what is visible (?)
-def get_datasets():
-    k_datasets = katsu_datasets()
+async def get_datasets():
+    k_datasets = await katsu_datasets()
     datasets_beacon_format = list(map(katsu_to_beacon_dataset_mapping, k_datasets))
     return beacon_collections_response({"collections": datasets_beacon_format})
 
 
 @datasets.route("/datasets/<id>", methods=["GET", "POST"])
 @authz_middleware.deco_public_endpoint  # TODO: authz - more flexibility in what is visible (?)
-def get_datasets_by_id(id):
-    k_dataset = katsu_datasets(id)
+async def get_datasets_by_id(id):
+    k_dataset = await katsu_datasets(id)
     dataset_beacon_format = katsu_to_beacon_dataset_mapping(k_dataset) if k_dataset else []
     return beacon_collections_response({"collections": dataset_beacon_format})
 

bento_beacon/endpoints/individuals.py

@@ -28,7 +28,7 @@
 
 
 @individuals.route("/individuals", methods=["GET", "POST"])
-def get_individuals():
+async def get_individuals():
     variants_query = g.beacon_query_parameters["variants_query"]
     phenopacket_filters = g.beacon_query_parameters["phenopacket_filters"]
     experiment_filters = g.beacon_query_parameters["experiment_filters"]
@@ -42,20 +42,20 @@
     # TODO: return default granularity rather than count (default could be bool rather than count)
     if no_query:
         add_info_to_response("no query found, returning total count")
-        total_count = katsu_total_individuals_count()
+        total_count = await katsu_total_individuals_count()
         if summary_stats_requested():
-            add_overview_stats_to_response()
-        return build_query_response(numTotalResults=total_count)
+            await add_overview_stats_to_response()
+        return await build_query_response(numTotalResults=total_count)
 
     # ----------------------------------------------------------
     #  collect biosample ids from variant and experiment search
     # ----------------------------------------------------------
     sample_ids = []
 
     if search_sample_ids:
-        sample_ids = biosample_id_search(variants_query=variants_query, experiment_filters=experiment_filters)
+        sample_ids = await biosample_id_search(variants_query=variants_query, experiment_filters=experiment_filters)
         if not sample_ids:
-            return zero_count_response()
+            return await zero_count_response()
 
     # -------------------------------
     #  get individuals
@@ -65,38 +65,38 @@
 
     # get individuals from katsu config search
     if config_filters:
-        config_ids = search_from_config(config_filters)
+        config_ids = await search_from_config(config_filters)
         if not config_ids:
-            return zero_count_response()
+            return await zero_count_response()
         individual_results["config_ids"] = config_ids
 
     if not config_search_only:
         # retrieve all matching individuals from sample id search, filtered by any phenopacket filters
         # either of phenopacket_filters or sample_ids can be empty
-        phenopacket_ids = katsu_filters_and_sample_ids_query(phenopacket_filters, "phenopacket", sample_ids)
+        phenopacket_ids = await katsu_filters_and_sample_ids_query(phenopacket_filters, "phenopacket", sample_ids)
         if not phenopacket_ids:
-            return zero_count_response()
+            return await zero_count_response()
         individual_results["phenopacket_ids"] = phenopacket_ids
 
     # baroque syntax but covers all cases
     individual_ids = list(reduce(set.intersection, (set(ids) for ids in individual_results.values())))
 
     if summary_stats_requested():
-        add_stats_to_response(individual_ids)
+        await add_stats_to_response(individual_ids)
 
-    return build_query_response(ids=individual_ids, full_record_handler=individuals_full_results)
+    return await build_query_response(ids=individual_ids, full_record_handler=individuals_full_results)
 
 
 # TODO: pagination (ideally after katsu search gets paginated)
-def individuals_full_results(ids):
+async def individuals_full_results(ids):
 
     # temp
     # if len(ids) > 100:
     #     return {"message": "too many ids for full response"}
 
-    handover_permission = check_permission(P_DOWNLOAD_DATA)
-    handover = handover_for_ids(ids) if handover_permission else {}
-    phenopackets_by_result_set = phenopackets_for_ids(ids).get("results", {})
+    handover_permission = await check_permission(P_DOWNLOAD_DATA)
+    handover = (await handover_for_ids(ids)) if handover_permission else {}
+    phenopackets_by_result_set = (await phenopackets_for_ids(ids)).get("results", {})
     result_ids = list(phenopackets_by_result_set.keys())
     result_sets = {}
     numTotalResults = 0
@@ -123,8 +123,8 @@
 # forbidden / unauthorized if no permissions
 @individuals.route("/individuals/<id>", methods=["GET", "POST"])
 @authz_middleware.deco_require_permissions_on_resource({P_QUERY_DATA})
-def individual_by_id(id):
-    result_sets, numTotalResults = individuals_full_results([id])
+async def individual_by_id(id):
+    result_sets, numTotalResults = await individuals_full_results([id])
 
     # return 404 if not found
     # only authorized users will get 404 here, so this can't be used to probe ids