wip

src/chains/signals.py

@@ -1,48 +1,44 @@
 import logging
 from typing import Any
 
-from django.conf import settings
 from django.db.models.signals import post_delete, post_save
 from django.dispatch import receiver
 
-import clients.safe_client_gateway
+from clients.safe_client_gateway import HookEvent, flush, hook_event
 
 from .models import Chain, Feature, GasPrice, Wallet
 
 logger = logging.getLogger(__name__)
 
 
-def _flush_cgw_chains() -> None:
-    clients.safe_client_gateway.flush(
-        cgw_url=settings.CGW_URL,
-        cgw_flush_token=settings.CGW_FLUSH_TOKEN,
-        json={"invalidate": "Chains"},
-    )
+def trigger_chain_update_event(chain: Chain):
+    flush()
+    hook_event(HookEvent(type=HookEvent.Type.CHAIN_UPDATE, chain_id=chain.id))
 
 
 @receiver(post_save, sender=Chain)
 @receiver(post_delete, sender=Chain)
-def on_chain_update(sender: Chain, **kwargs: Any) -> None:
+def on_chain_update(sender: Chain, instance, **kwargs: Any) -> None:
     logger.info("Chain update. Triggering CGW webhook")
-    _flush_cgw_chains()
+    trigger_chain_update_event(instance)
 
 
 @receiver(post_save, sender=GasPrice)
 @receiver(post_delete, sender=GasPrice)
-def on_gas_price_update(sender: GasPrice, **kwargs: Any) -> None:
+def on_gas_price_update(sender: GasPrice, instance, **kwargs: Any) -> None:
     logger.info("GasPrice update. Triggering CGW webhook")
-    _flush_cgw_chains()
+    trigger_chain_update_event(instance)
 
 
 @receiver(post_save, sender=Feature)
 @receiver(post_delete, sender=Feature)
-def on_feature_update(sender: Feature, **kwargs: Any) -> None:
+def on_feature_update(sender: Feature, instance, **kwargs: Any) -> None:
     logger.info("Feature update. Triggering CGW webhook")
-    _flush_cgw_chains()
+    trigger_chain_update_event(instance)
 
 
 @receiver(post_save, sender=Wallet)
 @receiver(post_delete, sender=Wallet)
-def on_wallet_update(sender: Wallet, **kwargs: Any) -> None:
+def on_wallet_update(sender: Wallet, instance, **kwargs: Any) -> None:
     logger.info("Wallet update. Triggering CGW webhook")
-    _flush_cgw_chains()
+    trigger_chain_update_event(instance)

src/clients/safe_client_gateway.py

@@ -1,6 +1,8 @@
 import logging
+from dataclasses import dataclass
+from enum import Enum
 from functools import cache
-from typing import Any, Dict, Optional
+from typing import Any, Dict
 from urllib.parse import urljoin
 
 import requests
@@ -9,6 +11,16 @@
 logger = logging.getLogger(__name__)
 
 
+@dataclass
+class HookEvent:
+    class Type(str, Enum):
+        CHAIN_UPDATE = "CHAIN_UPDATE"
+        SAFE_APPS_UPDATE = "SAFE_APPS_UPDATE"
+
+    type: Type
+    chain_id: int
+
+
 @cache
 def setup_session() -> requests.Session:
     session = requests.Session()
@@ -20,24 +32,36 @@ def setup_session() -> requests.Session:
     return session
 
 
-def flush(
-    cgw_url: Optional[str], cgw_flush_token: Optional[str], json: Dict[str, Any]
-) -> None:
-    if cgw_url is None:
-        logger.error("CGW_URL is not set. Skipping hook call")
-        return
-    if cgw_flush_token is None:
-        logger.error("CGW_FLUSH_TOKEN is not set. Skipping hook call")
-        return
+def check_cgw_setup():
+    if settings.CGW_URL is None:
+        raise ValueError("CGW_URL is not set. Skipping hook call")
+    if settings.CGW_FLUSH_TOKEN is None:
+        raise ValueError("CGW_FLUSH_TOKEN is not set. Skipping hook call")
+
+
+def flush() -> None:
+    try:
+        check_cgw_setup()
+        url = urljoin(settings.CGW_URL, "/v2/flush")
+        post(url, json={"invalidate": "Chains"})
+    except Exception as error:
+        logger.error(error)
+
 
-    url = urljoin(cgw_url, "/v2/flush")
+def hook_event(event: HookEvent) -> None:
     try:
-        post = setup_session().post(
-            url,
-            json=json,
-            headers={"Authorization": f"Basic {cgw_flush_token}"},
-            timeout=settings.CGW_SESSION_TIMEOUT_SECONDS,
-        )
-        post.raise_for_status()
+        check_cgw_setup()
+        url = urljoin(settings.CGW_URL, "/v1/hooks/events")
+        post(url, json={"type": event.type, "chain_id": event.chain_id})
     except Exception as error:
         logger.error(error)
+
+
+def post(url, json: Dict[str, Any]):
+    post = setup_session().post(
+        url,
+        json=json,
+        headers={"Authorization": f"Basic {settings.CGW_FLUSH_TOKEN}"},
+        timeout=settings.CGW_SESSION_TIMEOUT_SECONDS,
+    )
+    post.raise_for_status()