feat: Implement admin notification for rate limit exceeded events.

src/auth/service.py

@@ -147,6 +147,89 @@ async def internal_key(request: Request):
 
 
 
+import httpx
+import logging
+from typing import Dict, Any
+import src.config as config
+
+
+log = logging.getLogger(__name__)
+
+
+AUTH_NOTIFY_ENDPOINT = f"{config.AUTH_SERVICE_API}/admin/notify-limit"
+
+
+async def notify_admin_on_rate_limit(
+    endpoint_name: str,
+    ip_address: str,
+    method: str = "POST",
+    cooldown: int = 900,
+    timeout: int = 5
+) -> Dict[str, Any]:
+    """
+    Kirim notifikasi ke admin via be-auth service ketika rate limit terlampaui.
+   
+    Async version - gunakan di async context.
+    """
+    payload = {
+        "endpoint_name": endpoint_name,
+        "ip_address": ip_address,
+        "method": method,
+        "cooldown": cooldown,
+    }
+
+
+    try:
+        async with httpx.AsyncClient(timeout=timeout) as client:
+            response = await client.post(AUTH_NOTIFY_ENDPOINT, json=payload)
+            response.raise_for_status()
+            result = response.json()
+            log.info(f"Notifikasi admin sent | Endpoint: {endpoint_name}")
+            return result
+
+
+    except Exception as e:
+        log.error(f"Error notifying admin: {str(e)}")
+        return {"status": False, "message": str(e), "data": payload}
+
+
+
+
+def notify_admin_on_rate_limit_sync(
+    endpoint_name: str,
+    ip_address: str,
+    method: str = "POST",
+    cooldown: int = 900,
+    timeout: int = 5
+) -> Dict[str, Any]:
+    """
+    Kirim notifikasi ke admin via be-auth service.
+   
+    Sync version - gunakan di exception handler atau sync context.
+    RECOMMENDED untuk use case ini.
+    """
+    payload = {
+        "endpoint_name": endpoint_name,
+        "ip_address": ip_address,
+        "method": method,
+        "cooldown": cooldown,
+    }
+
+
+    try:
+        response = httpx.post(AUTH_NOTIFY_ENDPOINT, json=payload, timeout=timeout)
+        response.raise_for_status()
+        result = response.json()
+        log.info(f"Notifikasi admin sent | Endpoint: {endpoint_name}")
+        return result
+
+
+    except Exception as e:
+        log.error(f"Error notifying admin: {str(e)}")
+        return {"status": False, "message": str(e), "data": payload}
+
+    
+
 CurrentUser = Annotated[UserBase, Depends(get_current_user)]
 Token = Annotated[str, Depends(get_token)]
 InternalKey = Annotated[str, Depends(internal_key)]

src/exceptions.py

@@ -14,6 +14,7 @@ from sqlalchemy.exc import (DataError, DBAPIError, IntegrityError,
                             SQLAlchemyError)
 
 from src.enums import ResponseStatus
+from src.auth.service import notify_admin_on_rate_limit_sync
 
 from starlette.exceptions import HTTPException as StarletteHTTPException
 
@@ -113,6 +114,13 @@ def handle_exception(request: Request, exc: Exception):
     request.state.error_id = error_id
 
     if isinstance(exc, RateLimitExceeded):
+        # Kirim notifikasi ke admin
+        notify_admin_on_rate_limit_sync(
+            endpoint_name=request_info["endpoint"],
+            ip_address=request_info["remote_addr"],
+            method=request_info["method"],
+        )
+
         log.warning(
             f"Rate limit exceeded | Error ID: {error_id}",
             extra={

src/overhaul_activity/service.py

@@ -121,8 +121,6 @@ async def get_all(
     equipments = paginated_results["items"]
     
     material_cost = await get_cm_cost_summary(collector_db=collector_db, last_oh_date=prev_oh_scope.end_date, upcoming_oh_date=overhaul.start_date)
-    # total_equipment for cost calculation should be the global total, not just the page
-    total_count = paginated_results["total"]
     overhaul_cost = await get_oh_cost_summary(collector_db=collector_db, last_oh_date=prev_oh_scope.end_date, upcoming_oh_date=overhaul.start_date)
 
     results = []