new types, updated rate limit service (#102)

* new types, updated rate limit service
* more typed stuff, also updated openapi spec and regenerated types
* frontend linting fix

Author: HardMax71

backend/app/api/routes/dlq.py

@@ -63,15 +63,7 @@ async def retry_dlq_messages(
     retry_request: ManualRetryRequest, dlq_manager: FromDishka[DLQManager]
 ) -> DLQBatchRetryResponse:
     result = await dlq_manager.retry_messages_batch(retry_request.event_ids)
-    return DLQBatchRetryResponse(
-        total=result.total,
-        successful=result.successful,
-        failed=result.failed,
-        details=[
-            {"event_id": d.event_id, "status": d.status, **({"error": d.error} if d.error else {})}
-            for d in result.details
-        ],
-    )
+    return DLQBatchRetryResponse.model_validate(result, from_attributes=True)
 
 
 @router.post("/retry-policy", response_model=MessageResponse)

backend/app/api/routes/events.py

@@ -1,7 +1,7 @@
 import asyncio
 import logging
 from datetime import datetime, timedelta, timezone
-from typing import Annotated, Any, Dict, List
+from typing import Annotated, Any
 
 from dishka import FromDishka
 from dishka.integrations.fastapi import DishkaRoute
@@ -74,7 +74,7 @@ async def get_execution_events(
 async def get_user_events(
     current_user: Annotated[UserResponse, Depends(current_user)],
     event_service: FromDishka[EventService],
-    event_types: List[EventType] | None = Query(None),
+    event_types: list[EventType] | None = Query(None),
     start_time: datetime | None = Query(None),
     end_time: datetime | None = Query(None),
     limit: int = Query(100, ge=1, le=1000),
@@ -259,12 +259,12 @@ async def publish_custom_event(
     return PublishEventResponse(event_id=event_id, status="published", timestamp=datetime.now(timezone.utc))
 
 
-@router.post("/aggregate", response_model=List[Dict[str, Any]])
+@router.post("/aggregate", response_model=list[dict[str, Any]])
 async def aggregate_events(
     current_user: Annotated[UserResponse, Depends(current_user)],
     aggregation: EventAggregationRequest,
     event_service: FromDishka[EventService],
-) -> List[Dict[str, Any]]:
+) -> list[dict[str, Any]]:
     result = await event_service.aggregate_events(
         user_id=current_user.user_id,
         user_role=current_user.role,
@@ -275,10 +275,10 @@ async def aggregate_events(
     return result.results
 
 
-@router.get("/types/list", response_model=List[str])
+@router.get("/types/list", response_model=list[str])
 async def list_event_types(
     current_user: Annotated[UserResponse, Depends(current_user)], event_service: FromDishka[EventService]
-) -> List[str]:
+) -> list[str]:
     event_types = await event_service.list_event_types(user_id=current_user.user_id, user_role=current_user.role)
     return event_types
 

backend/app/api/routes/execution.py

@@ -14,6 +14,7 @@
 from app.domain.enums.user import UserRole
 from app.domain.events.typed import BaseEvent, DomainEvent, EventMetadata
 from app.domain.exceptions import DomainError
+from app.domain.idempotency import KeyStrategy
 from app.schemas_pydantic.execution import (
     CancelExecutionRequest,
     CancelResponse,
@@ -88,15 +89,15 @@ async def create_execution(
         # Check for duplicate request using custom key
         idempotency_result = await idempotency_manager.check_and_reserve(
             event=pseudo_event,
-            key_strategy="custom",
+            key_strategy=KeyStrategy.CUSTOM,
             custom_key=f"http:{current_user.user_id}:{idempotency_key}",
             ttl_seconds=86400,  # 24 hours TTL for HTTP idempotency
         )
 
         if idempotency_result.is_duplicate:
             cached_json = await idempotency_manager.get_cached_json(
                 event=pseudo_event,
-                key_strategy="custom",
+                key_strategy=KeyStrategy.CUSTOM,
                 custom_key=f"http:{current_user.user_id}:{idempotency_key}",
             )
             return ExecutionResponse.model_validate_json(cached_json)
@@ -119,7 +120,7 @@ async def create_execution(
             await idempotency_manager.mark_completed_with_json(
                 event=pseudo_event,
                 cached_json=response_model.model_dump_json(),
-                key_strategy="custom",
+                key_strategy=KeyStrategy.CUSTOM,
                 custom_key=f"http:{current_user.user_id}:{idempotency_key}",
             )
 
@@ -131,7 +132,7 @@ async def create_execution(
             await idempotency_manager.mark_failed(
                 event=pseudo_event,
                 error=str(e),
-                key_strategy="custom",
+                key_strategy=KeyStrategy.CUSTOM,
                 custom_key=f"http:{current_user.user_id}:{idempotency_key}",
             )
         raise
@@ -141,7 +142,7 @@ async def create_execution(
             await idempotency_manager.mark_failed(
                 event=pseudo_event,
                 error=str(e),
-                key_strategy="custom",
+                key_strategy=KeyStrategy.CUSTOM,
                 custom_key=f"http:{current_user.user_id}:{idempotency_key}",
             )
         raise HTTPException(status_code=500, detail="Internal server error during script execution") from e

backend/app/core/adaptive_sampling.py

@@ -2,7 +2,7 @@
 import threading
 import time
 from collections import deque
-from typing import Sequence, Tuple
+from collections.abc import Sequence
 
 from opentelemetry.context import Context
 from opentelemetry.sdk.trace.sampling import Decision, Sampler, SamplingResult
@@ -170,7 +170,7 @@ def _is_error(self, attributes: Attributes | None) -> bool:
 
         return False
 
-    def _calculate_metrics(self) -> Tuple[float, int]:
+    def _calculate_metrics(self) -> tuple[float, int]:
         """Calculate current error rate and request rate"""
         now = time.time()
         minute_ago = now - 60

backend/app/core/correlation.py

@@ -1,6 +1,6 @@
 import uuid
 from datetime import datetime, timezone
-from typing import Any, Dict
+from typing import Any
 
 from starlette.datastructures import MutableHeaders
 from starlette.types import ASGIApp, Message, Receive, Scope, Send
@@ -23,11 +23,11 @@ def get_correlation_id() -> str:
         return correlation_id_context.get() or ""
 
     @staticmethod
-    def set_request_metadata(metadata: Dict[str, Any]) -> None:
+    def set_request_metadata(metadata: dict[str, Any]) -> None:
         request_metadata_context.set(metadata)
 
     @staticmethod
-    def get_request_metadata() -> Dict[str, Any]:
+    def get_request_metadata() -> dict[str, Any]:
         return request_metadata_context.get() or {}
 
     @staticmethod

backend/app/core/dishka_lifespan.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import asyncio
 import logging
 from contextlib import AsyncExitStack, asynccontextmanager

backend/app/core/logging.py

@@ -3,13 +3,13 @@
 import logging
 import re
 from datetime import datetime, timezone
-from typing import Any, Dict
+from typing import Any
 
 from opentelemetry import trace
 
 correlation_id_context: contextvars.ContextVar[str | None] = contextvars.ContextVar("correlation_id", default=None)
 
-request_metadata_context: contextvars.ContextVar[Dict[str, Any] | None] = contextvars.ContextVar(
+request_metadata_context: contextvars.ContextVar[dict[str, Any] | None] = contextvars.ContextVar(
     "request_metadata", default=None
 )
 

backend/app/core/metrics/rate_limit.py

@@ -5,103 +5,40 @@ class RateLimitMetrics(BaseMetrics):
     """Metrics for rate limiting system."""
 
     def _create_instruments(self) -> None:
-        # Request metrics
-        self.requests_total = self._meter.create_counter(
+        self._requests = self._meter.create_counter(
             name="rate_limit.requests.total",
             description="Total number of rate limit checks",
             unit="1",
         )
-        self.allowed = self._meter.create_counter(
+        self._allowed = self._meter.create_counter(
             name="rate_limit.allowed.total",
             description="Number of allowed requests",
             unit="1",
         )
-        self.rejected = self._meter.create_counter(
+        self._rejected = self._meter.create_counter(
             name="rate_limit.rejected.total",
             description="Number of rejected requests",
             unit="1",
         )
-        self.bypass = self._meter.create_counter(
+        self._bypass = self._meter.create_counter(
             name="rate_limit.bypass.total",
             description="Number of bypassed rate limit checks",
             unit="1",
         )
 
-        # Performance metrics
-        self.check_duration = self._meter.create_histogram(
-            name="rate_limit.check.duration",
-            description="Duration of rate limit checks",
-            unit="ms",
-        )
-        self.redis_duration = self._meter.create_histogram(
-            name="rate_limit.redis.duration",
-            description="Duration of Redis operations",
-            unit="ms",
-        )
-        self.algorithm_duration = self._meter.create_histogram(
-            name="rate_limit.algorithm.duration",
-            description="Time to execute rate limit algorithm",
-            unit="ms",
-        )
-
-        # Usage metrics
-        self.remaining = self._meter.create_histogram(
-            name="rate_limit.remaining",
-            description="Remaining requests in rate limit window",
-            unit="1",
-        )
-        self.quota_usage = self._meter.create_histogram(
-            name="rate_limit.quota.usage",
-            description="Percentage of quota used",
-            unit="%",
-        )
-        self.window_size = self._meter.create_histogram(
-            name="rate_limit.window.size",
-            description="Size of rate limit window",
-            unit="s",
-        )
-
-        # Configuration metrics - using histograms to record absolute values
-        # We record the current value, and Grafana queries the latest
-        self.active_rules = self._meter.create_histogram(
-            name="rate_limit.active.rules",
-            description="Number of active rate limit rules",
-            unit="1",
-        )
-        self.custom_users = self._meter.create_histogram(
-            name="rate_limit.custom.users",
-            description="Number of users with custom rate limits",
-            unit="1",
-        )
-        self.bypass_users = self._meter.create_histogram(
-            name="rate_limit.bypass.users",
-            description="Number of users with rate limit bypass",
-            unit="1",
-        )
+    def record_request(self, endpoint: str, authenticated: bool, algorithm: str) -> None:
+        """Record a rate limit check request."""
+        attrs = {"endpoint": endpoint, "authenticated": str(authenticated).lower(), "algorithm": algorithm}
+        self._requests.add(1, attrs)
 
-        # Token bucket specific metrics
-        self.token_bucket_tokens = self._meter.create_histogram(
-            name="rate_limit.token_bucket.tokens",
-            description="Current tokens in token bucket",
-            unit="1",
-        )
-        self.token_bucket_refill_rate = self._meter.create_histogram(
-            name="rate_limit.token_bucket.refill_rate",
-            description="Token bucket refill rate",
-            unit="tokens/s",
-        )
+    def record_allowed(self, endpoint: str, group: str) -> None:
+        """Record an allowed request."""
+        self._allowed.add(1, {"endpoint": endpoint, "group": group})
 
-        # Error metrics
-        self.redis_errors = self._meter.create_counter(
-            name="rate_limit.redis.errors.total",
-            description="Number of Redis errors",
-            unit="1",
-        )
-        self.config_errors = self._meter.create_counter(
-            name="rate_limit.config.errors.total",
-            description="Number of configuration load errors",
-            unit="1",
-        )
+    def record_rejected(self, endpoint: str, group: str) -> None:
+        """Record a rejected request."""
+        self._rejected.add(1, {"endpoint": endpoint, "group": group})
 
-        # Authenticated vs anonymous checks can be derived from labels on requests_total
-        # No separate ip/user counters to avoid duplication and complexity.
+    def record_bypass(self, endpoint: str) -> None:
+        """Record a bypassed rate limit check."""
+        self._bypass.add(1, {"endpoint": endpoint})

backend/app/core/middlewares/cache.py

@@ -1,13 +1,11 @@
-from typing import Dict
-
 from starlette.datastructures import MutableHeaders
 from starlette.types import ASGIApp, Message, Receive, Scope, Send
 
 
 class CacheControlMiddleware:
     def __init__(self, app: ASGIApp):
         self.app = app
-        self.cache_policies: Dict[str, str] = {
+        self.cache_policies: dict[str, str] = {
             "/api/v1/k8s-limits": "public, max-age=300",  # 5 minutes
             "/api/v1/example-scripts": "public, max-age=600",  # 10 minutes
             "/api/v1/auth/verify-token": "private, no-cache",  # 30 seconds

backend/app/core/providers.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import logging
 from typing import AsyncIterator
 
@@ -111,12 +113,12 @@ async def get_redis_client(self, settings: Settings, logger: logging.Logger) ->
             socket_timeout=5,
         )
         # Test connection
-        await client.ping()  # type: ignore[misc]  # redis-py dual sync/async return type
+        await client.ping()  # type: ignore[misc]  # redis-py returns Awaitable[bool] | bool
         logger.info(f"Redis connected: {settings.REDIS_HOST}:{settings.REDIS_PORT}/{settings.REDIS_DB}")
         try:
             yield client
         finally:
-            await client.aclose()
+            await client.close()
 
     @provide
     def get_rate_limit_service(
@@ -383,7 +385,9 @@ class SSEProvider(Provider):
     scope = Scope.APP
 
     @provide
-    async def get_sse_redis_bus(self, redis_client: redis.Redis, logger: logging.Logger) -> AsyncIterator[SSERedisBus]:
+    async def get_sse_redis_bus(
+        self, redis_client: redis.Redis, logger: logging.Logger
+    ) -> AsyncIterator[SSERedisBus]:
         bus = SSERedisBus(redis_client, logger)
         yield bus
 
@@ -480,9 +484,10 @@ async def get_user_settings_service(
             repository: UserSettingsRepository,
             kafka_event_service: KafkaEventService,
             event_bus_manager: EventBusManager,
+            settings: Settings,
             logger: logging.Logger,
     ) -> UserSettingsService:
-        service = UserSettingsService(repository, kafka_event_service, logger)
+        service = UserSettingsService(repository, kafka_event_service, settings, logger)
         await service.initialize(event_bus_manager)
         return service