fixes

Author: HardMax71

backend/app/api/routes/admin/events.py

@@ -43,7 +43,7 @@ async def browse_events(request: EventBrowseRequest, service: FromDishka[AdminEv
         event_filter = EventFilterMapper.from_admin_pydantic(request.filters)
 
         result = await service.browse_events(
-            filter=event_filter,
+            event_filter=event_filter,
             skip=request.skip,
             limit=request.limit,
             sort_by=request.sort_by,
@@ -92,7 +92,7 @@ async def export_events_csv(
                 end_time=end_time,
             )
         )
-        result = await service.export_events_csv_content(filter=export_filter, limit=limit)
+        result = await service.export_events_csv_content(event_filter=export_filter, limit=limit)
         return StreamingResponse(
             iter([result.content]),
             media_type=result.media_type,
@@ -128,7 +128,7 @@ async def export_events_json(
                 end_time=end_time,
             )
         )
-        result = await service.export_events_json_content(filter=export_filter, limit=limit)
+        result = await service.export_events_json_content(event_filter=export_filter, limit=limit)
         return StreamingResponse(
             iter([result.content]),
             media_type=result.media_type,

backend/app/db/repositories/admin/admin_events_repository.py

@@ -58,14 +58,14 @@ def __init__(self, db: Database):
 
     async def browse_events(
         self,
-        filter: EventFilter,
+        event_filter: EventFilter,
         skip: int = 0,
         limit: int = 50,
         sort_by: str = EventFields.TIMESTAMP,
         sort_order: int = SortDirection.DESCENDING,
     ) -> EventBrowseResult:
         """Browse events with filters using domain models."""
-        query = EventFilterMapper.to_mongo_query(filter)
+        query = EventFilterMapper.to_mongo_query(event_filter)
 
         # Get total count
         total = await self.events_collection.count_documents(query)
@@ -191,9 +191,9 @@ async def get_event_stats(self, hours: int = 24) -> EventStatistics:
 
         return statistics
 
-    async def export_events_csv(self, filter: EventFilter) -> List[EventExportRow]:
+    async def export_events_csv(self, event_filter: EventFilter) -> List[EventExportRow]:
         """Export events as CSV data."""
-        query = EventFilterMapper.to_mongo_query(filter)
+        query = EventFilterMapper.to_mongo_query(event_filter)
 
         cursor = self.events_collection.find(query).sort(EventFields.TIMESTAMP, SortDirection.DESCENDING).limit(10000)
 

backend/app/db/repositories/dlq_repository.py

@@ -112,10 +112,8 @@ async def get_messages(
         limit: int = 50,
         offset: int = 0,
     ) -> DLQMessageListResult:
-        # Create filter
-        filter = DLQMessageFilter(status=status, topic=topic, event_type=event_type)
-
-        query = DLQMapper.filter_to_query(filter)
+        msg_filter = DLQMessageFilter(status=status, topic=topic, event_type=event_type)
+        query = DLQMapper.filter_to_query(msg_filter)
         total_count = await self.dlq_collection.count_documents(query)
 
         cursor = self.dlq_collection.find(query).sort(DLQFields.FAILED_AT, -1).skip(offset).limit(limit)

backend/app/db/repositories/replay_repository.py

@@ -89,16 +89,16 @@ async def update_replay_session(self, session_id: str, updates: ReplaySessionUpd
         result = await self.replay_collection.update_one({"session_id": session_id}, {"$set": mongo_updates})
         return result.modified_count > 0
 
-    async def count_events(self, filter: ReplayFilter) -> int:
+    async def count_events(self, replay_filter: ReplayFilter) -> int:
         """Count events matching the given filter"""
-        query = filter.to_mongo_query()
+        query = replay_filter.to_mongo_query()
         return await self.events_collection.count_documents(query)
 
     async def fetch_events(
-        self, filter: ReplayFilter, batch_size: int = 100, skip: int = 0
+        self, replay_filter: ReplayFilter, batch_size: int = 100, skip: int = 0
     ) -> AsyncIterator[List[Dict[str, Any]]]:
         """Fetch events in batches based on filter"""
-        query = filter.to_mongo_query()
+        query = replay_filter.to_mongo_query()
         cursor = self.events_collection.find(query).sort("timestamp", 1).skip(skip)
 
         batch = []

backend/app/db/repositories/saga_repository.py

@@ -55,8 +55,8 @@ async def get_sagas_by_execution(self, execution_id: str, state: SagaState | Non
         docs = await cursor.to_list(length=None)
         return [self.mapper.from_mongo(doc) for doc in docs]
 
-    async def list_sagas(self, filter: SagaFilter, limit: int = 100, skip: int = 0) -> SagaListResult:
-        query = self.filter_mapper.to_mongodb_query(filter)
+    async def list_sagas(self, saga_filter: SagaFilter, limit: int = 100, skip: int = 0) -> SagaListResult:
+        query = self.filter_mapper.to_mongodb_query(saga_filter)
 
         # Get total count
         total = await self.sagas.count_documents(query)
@@ -108,8 +108,8 @@ async def find_timed_out_sagas(
         docs = await cursor.to_list(length=limit)
         return [self.mapper.from_mongo(doc) for doc in docs]
 
-    async def get_saga_statistics(self, filter: SagaFilter | None = None) -> dict[str, object]:
-        query = self.filter_mapper.to_mongodb_query(filter) if filter else {}
+    async def get_saga_statistics(self, saga_filter: SagaFilter | None = None) -> dict[str, object]:
+        query = self.filter_mapper.to_mongodb_query(saga_filter) if saga_filter else {}
 
         # Basic counts
         total = await self.sagas.count_documents(query)

backend/app/domain/events/query_builders.py

@@ -51,9 +51,9 @@ def size(field: str) -> dict[str, str]:
         return {"$size": field}
 
     @staticmethod
-    def date_to_string(date_field: str, format: str = "%Y-%m-%d-%H") -> dict[str, Any]:
+    def date_to_string(date_field: str, date_format: str = "%Y-%m-%d-%H") -> dict[str, Any]:
         """Create a $dateToString expression."""
-        return {"$dateToString": {"format": format, "date": date_field}}
+        return {"$dateToString": {"format": date_format, "date": date_field}}
 
 
 class EventStatsAggregation:

backend/app/infrastructure/mappers/saga_mapper.py

@@ -214,31 +214,34 @@ def to_cancelled_event(
 class SagaFilterMapper:
     """Maps saga filters to MongoDB queries."""
 
-    def to_mongodb_query(self, filter: SagaFilter) -> dict[str, Any]:
+    def to_mongodb_query(self, saga_filter: SagaFilter | None) -> dict[str, Any]:
         """Convert filter to MongoDB query."""
         query: dict[str, Any] = {}
 
-        if filter.state:
-            query["state"] = filter.state.value
+        if not saga_filter:
+            return query
 
-        if filter.execution_ids:
-            query["execution_id"] = {"$in": filter.execution_ids}
+        if saga_filter.state:
+            query["state"] = saga_filter.state.value
 
-        if filter.saga_name:
-            query["saga_name"] = filter.saga_name
+        if saga_filter.execution_ids:
+            query["execution_id"] = {"$in": saga_filter.execution_ids}
 
-        if filter.error_status is not None:
-            if filter.error_status:
+        if saga_filter.saga_name:
+            query["saga_name"] = saga_filter.saga_name
+
+        if saga_filter.error_status is not None:
+            if saga_filter.error_status:
                 query["error_message"] = {"$ne": None}
             else:
                 query["error_message"] = None
 
-        if filter.created_after or filter.created_before:
+        if saga_filter.created_after or saga_filter.created_before:
             time_query: dict[str, Any] = {}
-            if filter.created_after:
-                time_query["$gte"] = filter.created_after
-            if filter.created_before:
-                time_query["$lte"] = filter.created_before
+            if saga_filter.created_after:
+                time_query["$gte"] = saga_filter.created_after
+            if saga_filter.created_before:
+                time_query["$lte"] = saga_filter.created_before
             query["created_at"] = time_query
 
         return query

backend/app/services/admin/admin_events_service.py

@@ -59,14 +59,14 @@ def __init__(self, repository: AdminEventsRepository, replay_service: ReplayServ
     async def browse_events(
         self,
         *,
-        filter: EventFilter,
+        event_filter: EventFilter,
         skip: int,
         limit: int,
         sort_by: str,
         sort_order: int,
     ) -> EventBrowseResult:
         return await self._repo.browse_events(
-            filter=filter, skip=skip, limit=limit, sort_by=sort_by, sort_order=sort_order
+            event_filter=event_filter, skip=skip, limit=limit, sort_by=sort_by, sort_order=sort_order
         )
 
     async def get_event_detail(self, event_id: str) -> EventDetail | None:
@@ -180,12 +180,12 @@ async def get_replay_status(self, session_id: str) -> ReplaySessionStatusDetail
         status = await self._repo.get_replay_status_with_progress(session_id)
         return status
 
-    async def export_events_csv(self, filter: EventFilter) -> List[EventExportRow]:
-        rows = await self._repo.export_events_csv(filter)
+    async def export_events_csv(self, event_filter: EventFilter) -> List[EventExportRow]:
+        rows = await self._repo.export_events_csv(event_filter)
         return rows
 
-    async def export_events_csv_content(self, *, filter: EventFilter, limit: int) -> ExportResult:
-        rows = await self._repo.export_events_csv(filter)
+    async def export_events_csv_content(self, *, event_filter: EventFilter, limit: int) -> ExportResult:
+        rows = await self._repo.export_events_csv(event_filter)
         output = StringIO()
         writer = csv.DictWriter(
             output,
@@ -216,8 +216,10 @@ async def export_events_csv_content(self, *, filter: EventFilter, limit: int) ->
         )
         return ExportResult(file_name=filename, content=output.getvalue(), media_type="text/csv")
 
-    async def export_events_json_content(self, *, filter: EventFilter, limit: int) -> ExportResult:
-        result = await self._repo.browse_events(filter=filter, skip=0, limit=limit, sort_by="timestamp", sort_order=-1)
+    async def export_events_json_content(self, *, event_filter: EventFilter, limit: int) -> ExportResult:
+        result = await self._repo.browse_events(
+            event_filter=event_filter, skip=0, limit=limit, sort_by="timestamp", sort_order=-1
+        )
         event_mapper = EventMapper()
         events_data: list[dict[str, Any]] = []
         for event in result.events:
@@ -232,13 +234,13 @@ async def export_events_json_content(self, *, filter: EventFilter, limit: int) -
                 "exported_at": datetime.now(timezone.utc).isoformat(),
                 "total_events": len(events_data),
                 "filters_applied": {
-                    "event_types": filter.event_types,
-                    "aggregate_id": filter.aggregate_id,
-                    "correlation_id": filter.correlation_id,
-                    "user_id": filter.user_id,
-                    "service_name": filter.service_name,
-                    "start_time": filter.start_time.isoformat() if filter.start_time else None,
-                    "end_time": filter.end_time.isoformat() if filter.end_time else None,
+                    "event_types": event_filter.event_types,
+                    "aggregate_id": event_filter.aggregate_id,
+                    "correlation_id": event_filter.correlation_id,
+                    "user_id": event_filter.user_id,
+                    "service_name": event_filter.service_name,
+                    "start_time": event_filter.start_time.isoformat() if event_filter.start_time else None,
+                    "end_time": event_filter.end_time.isoformat() if event_filter.end_time else None,
                 },
                 "export_limit": limit,
             },

backend/tests/unit/dlq/test_dlq_models.py

@@ -1,19 +1,18 @@
-from datetime import datetime, timezone
 import json
-
-import pytest
+from datetime import datetime, timezone
 
 from app.dlq import (
     AgeStatistics,
     DLQFields,
     DLQMessageFilter,
     DLQMessageStatus,
+    DLQStatistics,
     EventTypeStatistic,
     RetryPolicy,
     RetryStrategy,
     TopicStatistic,
-    DLQStatistics,
 )
+from app.domain.enums.events import EventType
 from app.events.schema.schema_registry import SchemaRegistryManager
 from app.infrastructure.kafka.events.metadata import EventMetadata
 from app.infrastructure.kafka.events.user import UserLoggedInEvent
@@ -22,6 +21,7 @@
 
 def _make_event() -> UserLoggedInEvent:
     from app.domain.enums.auth import LoginMethod
+
     return UserLoggedInEvent(
         user_id="u1",
         login_method=LoginMethod.PASSWORD,
@@ -95,15 +95,15 @@ def test_retry_policy_bounds() -> None:
 
 
 def test_filter_and_stats_models() -> None:
-    f = DLQMessageFilter(status=DLQMessageStatus.PENDING, topic="t", event_type="X")
+    f = DLQMessageFilter(status=DLQMessageStatus.PENDING, topic="t", event_type=EventType.EXECUTION_REQUESTED)
     q = DLQMapper.filter_to_query(f)
     assert q[DLQFields.STATUS] == DLQMessageStatus.PENDING
     assert q[DLQFields.ORIGINAL_TOPIC] == "t"
+    assert q[DLQFields.EVENT_TYPE] == EventType.EXECUTION_REQUESTED
 
     ts = TopicStatistic(topic="t", count=2, avg_retry_count=1.5)
     es = EventTypeStatistic(event_type="X", count=3)
     ages = AgeStatistics(min_age_seconds=1, max_age_seconds=10, avg_age_seconds=5)
     stats = DLQStatistics(by_status={"pending": 1}, by_topic=[ts], by_event_type=[es], age_stats=ages)
     assert stats.by_status["pending"] == 1
     assert isinstance(stats.timestamp, datetime)
-