Added data update callbacks config (#219)

* Added data update callbacks config

* Fixed comment

Author: danyi1212

horizon/config.py

@@ -1,4 +1,8 @@
+from typing import Any
+
 from opal_common.confi import Confi, confi
+from opal_common.schemas.data import CallbackEntry
+from pydantic import parse_obj_as, parse_raw_as
 
 MOCK_API_KEY = "MUST BE DEFINED"
 
@@ -271,6 +275,21 @@ def __new__(cls, prefix=None, is_model=True):
         description="The path to the file that contains the PDP version",
     )
 
+    @staticmethod
+    def parse_callbacks(value: Any) -> list[CallbackEntry]:
+        if isinstance(value, str):
+            return parse_raw_as(list[CallbackEntry], value)
+        else:
+            return parse_obj_as(list[CallbackEntry], value)
+
+    DATA_UPDATE_CALLBACKS: list[CallbackEntry] = confi.str(
+        "DATA_UPDATE_CALLBACKS",
+        [],
+        description="List of callbacks to be triggered when data is updated",
+        cast=parse_callbacks,
+        cast_from_json=parse_callbacks,
+    )
+
     # non configurable values -------------------------------------------------
 
     # redoc configuration (openapi schema)

horizon/factdb/client.py

@@ -18,6 +18,10 @@
 from opal_client.policy_store.schemas import PolicyStoreTypes
 from opal_common.authentication.deps import JWTAuthenticator
 from opal_common.authentication.verifier import JWTVerifier
+from opal_common.fetcher.providers.http_fetch_provider import (
+    HttpFetcherConfig,
+    HttpMethods,
+)
 from starlette import status
 from starlette.responses import JSONResponse
 
@@ -39,6 +43,11 @@ async def check_healthy(self) -> bool:
     async def check_ready(self) -> bool:
         return self._backup_loaded or await self.policy_store.is_ready()
 
+    def _init_fast_api_app(self) -> FastAPI:
+        # Called at the end of OPALClient.__init__
+        self._inject_extra_callbacks()
+        return super()._init_fast_api_app()
+
     def _configure_api_routes(self, app: FastAPI):
         """mounts the api routes on the app object."""
 
@@ -48,13 +57,14 @@ def _configure_api_routes(self, app: FastAPI):
         policy_router = init_policy_router(policy_updater=self.policy_updater)
         data_router = init_data_router(data_updater=self.data_updater)
         policy_store_router = init_policy_store_router(authenticator)
-        callbacks_router = init_callbacks_api(authenticator, self._callbacks_register)
 
         # mount the api routes on the app object
         app.include_router(policy_router, tags=["Policy Updater"])
         app.include_router(data_router, tags=["Data Updater"])
         app.include_router(policy_store_router, tags=["Policy Store"])
-        app.include_router(callbacks_router, tags=["Callbacks"])
+
+        # excluded callbacks api from the main api, since we use it internally.
+        # Use the DATA_UPDATE_CALLBACKS config to configure callbacks instead
 
         # top level routes (i.e: healthchecks)
         @app.get("/healthcheck", include_in_schema=False)
@@ -90,6 +100,28 @@ async def ready():
 
         return app
 
+    def _inject_extra_callbacks(self) -> None:
+        register = self._callbacks_register
+        default_config = HttpFetcherConfig(
+            method=HttpMethods.POST,
+            headers={"content-type": "application/json"},
+            process_data=False,
+            fetcher=None,
+        )
+        for entry in sidecar_config.DATA_UPDATE_CALLBACKS:
+            entry.config = entry.config or default_config
+            entry.key = entry.key or register.calc_hash(entry.url, entry.config)
+
+            if register.get(entry.key):
+                raise RuntimeError(
+                    f"Callback with key '{entry.key}' already exists. Please specify a different key."
+                )
+
+            logger.info(
+                f"Registering data update callback to url '{entry.url}' with key '{entry.key}'"
+            )
+            register.put(entry.url, entry.config, entry.key)
+
 
 class FactDBClient(ExtendedOpalClient):
     def __init__(