elimination of redundant mappers

Author: HardMax71

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

@@ -12,9 +12,7 @@
 from app.domain.enums.events import EventType
 from app.infrastructure.mappers import (
     AdminReplayApiMapper,
-    EventDetailMapper,
     EventFilterMapper,
-    EventMapper,
 )
 from app.schemas_pydantic.admin_events import (
     EventBrowseRequest,
@@ -48,9 +46,8 @@ async def browse_events(request: EventBrowseRequest, service: FromDishka[AdminEv
             sort_order=request.sort_order,
         )
 
-        event_mapper = EventMapper()
         return EventBrowseResponse(
-            events=[jsonable_encoder(event_mapper.to_dict(event)) for event in result.events],
+            events=[jsonable_encoder(event) for event in result.events],
             total=result.total,
             skip=result.skip,
             limit=result.limit,
@@ -144,12 +141,10 @@ async def get_event_detail(event_id: str, service: FromDishka[AdminEventsService
         if not result:
             raise HTTPException(status_code=404, detail="Event not found")
 
-        detail_mapper = EventDetailMapper()
-        serialized_result = jsonable_encoder(detail_mapper.to_dict(result))
         return EventDetailResponse(
-            event=serialized_result["event"],
-            related_events=serialized_result["related_events"],
-            timeline=serialized_result["timeline"],
+            event=jsonable_encoder(result.event),
+            related_events=[jsonable_encoder(e) for e in result.related_events],
+            timeline=[jsonable_encoder(e) for e in result.timeline],
         )
 
     except HTTPException:

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

@@ -9,8 +9,12 @@
 from app.domain.enums.user import UserRole
 from app.domain.rate_limit import UserRateLimit
 from app.domain.user import UserUpdate as DomainUserUpdate
-from app.infrastructure.mappers import AdminOverviewApiMapper
-from app.schemas_pydantic.admin_user_overview import AdminUserOverview
+from app.schemas_pydantic.admin_user_overview import (
+    AdminUserOverview,
+    DerivedCounts,
+    RateLimitSummary,
+)
+from app.schemas_pydantic.events import EventResponse, EventStatistics
 from app.schemas_pydantic.user import (
     DeleteUserResponse,
     MessageResponse,
@@ -110,8 +114,13 @@ async def get_user_overview(
         domain = await admin_user_service.get_user_overview(user_id=user_id, hours=24)
     except ValueError:
         raise HTTPException(status_code=404, detail="User not found")
-    mapper = AdminOverviewApiMapper()
-    return mapper.to_response(domain)
+    return AdminUserOverview(
+        user=UserResponse.model_validate(domain.user),
+        stats=EventStatistics.model_validate(domain.stats),
+        derived_counts=DerivedCounts.model_validate(domain.derived_counts),
+        rate_limit_summary=RateLimitSummary.model_validate(domain.rate_limit_summary),
+        recent_events=[EventResponse.model_validate(e).model_dump() for e in domain.recent_events],
+    )
 
 
 @router.put("/{user_id}", response_model=UserResponse)

backend/app/api/routes/replay.py

@@ -70,7 +70,7 @@ async def list_replay_sessions(
 @router.get("/sessions/{session_id}", response_model=ReplaySession)
 async def get_replay_session(session_id: str, service: FromDishka[ReplayService]) -> ReplaySession:
     state = service.get_session(session_id)
-    return ReplayApiMapper.session_to_response(state)
+    return ReplaySession.model_validate(state)
 
 
 @router.post("/cleanup", response_model=CleanupResponse)

backend/app/api/routes/saga.py

@@ -3,7 +3,6 @@
 from fastapi import APIRouter, Query, Request
 
 from app.domain.enums.saga import SagaState
-from app.infrastructure.mappers import SagaResponseMapper
 from app.infrastructure.mappers import UserMapper as AdminUserMapper
 from app.schemas_pydantic.saga import (
     SagaCancellationResponse,
@@ -47,8 +46,7 @@ async def get_saga_status(
     service_user = User.from_response(current_user)
     domain_user = AdminUserMapper.from_pydantic_service_user(service_user)
     saga = await saga_service.get_saga_with_access_check(saga_id, domain_user)
-    mapper = SagaResponseMapper()
-    return mapper.to_response(saga)
+    return SagaStatusResponse.from_domain(saga)
 
 
 @router.get("/execution/{execution_id}", response_model=SagaListResponse)
@@ -79,8 +77,7 @@ async def get_execution_sagas(
     service_user = User.from_response(current_user)
     domain_user = AdminUserMapper.from_pydantic_service_user(service_user)
     sagas = await saga_service.get_execution_sagas(execution_id, domain_user, state)
-    mapper = SagaResponseMapper()
-    saga_responses = mapper.list_to_responses(sagas)
+    saga_responses = [SagaStatusResponse.from_domain(s) for s in sagas]
     return SagaListResponse(sagas=saga_responses, total=len(saga_responses))
 
 
@@ -111,8 +108,7 @@ async def list_sagas(
     service_user = User.from_response(current_user)
     domain_user = AdminUserMapper.from_pydantic_service_user(service_user)
     result = await saga_service.list_user_sagas(domain_user, state, limit, offset)
-    mapper = SagaResponseMapper()
-    saga_responses = mapper.list_to_responses(result.sagas)
+    saga_responses = [SagaStatusResponse.from_domain(s) for s in result.sagas]
     return SagaListResponse(sagas=saga_responses, total=result.total)
 
 

backend/app/domain/events/event_models.py

@@ -1,10 +1,10 @@
 from dataclasses import dataclass, field
 from datetime import datetime
 from typing import Any
-from uuid import uuid4
 
 from app.core.utils import StringEnum
 from app.domain.enums.events import EventType
+from app.infrastructure.kafka.events.metadata import EventMetadata
 
 MongoQueryValue = str | dict[str, str | list[str] | float | datetime]
 MongoQuery = dict[str, MongoQueryValue]
@@ -73,19 +73,6 @@ class CollectionNames(StringEnum):
     DLQ_MESSAGES = "dlq_messages"
 
 
-@dataclass
-class DomainEventMetadata:
-    """Domain model for event metadata."""
-
-    service_name: str
-    service_version: str
-    correlation_id: str = field(default_factory=lambda: str(uuid4()))
-    user_id: str | None = None
-    ip_address: str | None = None
-    user_agent: str | None = None
-    environment: str = "production"
-
-
 @dataclass
 class Event:
     """Domain model for an event."""
@@ -94,7 +81,7 @@ class Event:
     event_type: EventType
     event_version: str
     timestamp: datetime
-    metadata: DomainEventMetadata
+    metadata: EventMetadata
     payload: dict[str, Any]
     aggregate_id: str | None = None
     stored_at: datetime | None = None

backend/app/infrastructure/mappers/__init__.py

@@ -1,21 +1,13 @@
 from .admin_mapper import (
     AuditLogMapper,
     SettingsMapper,
-    UserListResultMapper,
     UserMapper,
 )
-from .admin_overview_api_mapper import AdminOverviewApiMapper
 from .event_mapper import (
     ArchivedEventMapper,
-    EventBrowseResultMapper,
-    EventDetailMapper,
     EventExportRowMapper,
     EventFilterMapper,
-    EventListResultMapper,
     EventMapper,
-    EventProjectionMapper,
-    EventReplayInfoMapper,
-    EventStatisticsMapper,
     EventSummaryMapper,
 )
 from .notification_mapper import NotificationMapper
@@ -38,7 +30,6 @@
     SagaFilterMapper,
     SagaInstanceMapper,
     SagaMapper,
-    SagaResponseMapper,
 )
 from .saved_script_mapper import SavedScriptMapper
 from .sse_mapper import SSEMapper
@@ -47,22 +38,14 @@
 __all__ = [
     # Admin
     "UserMapper",
-    "UserListResultMapper",
     "SettingsMapper",
     "AuditLogMapper",
-    "AdminOverviewApiMapper",
     # Events
     "EventMapper",
     "EventSummaryMapper",
-    "EventDetailMapper",
-    "EventListResultMapper",
-    "EventBrowseResultMapper",
-    "EventStatisticsMapper",
-    "EventProjectionMapper",
     "ArchivedEventMapper",
     "EventExportRowMapper",
     "EventFilterMapper",
-    "EventReplayInfoMapper",
     # Notification
     "NotificationMapper",
     # Rate limit
@@ -86,7 +69,6 @@
     # Saga
     "SagaMapper",
     "SagaFilterMapper",
-    "SagaResponseMapper",
     "SagaEventMapper",
     "SagaInstanceMapper",
 ]

backend/app/infrastructure/mappers/admin_mapper.py

@@ -19,7 +19,6 @@
 from app.domain.user import (
     UserCreation,
     UserFields,
-    UserListResult,
     UserRole,
     UserSearchFilter,
     UserUpdate,
@@ -67,22 +66,6 @@ def from_mongo_document(data: Dict[str, Any]) -> DomainAdminUser:
             updated_at=data.get(UserFields.UPDATED_AT, datetime.now(timezone.utc)),
         )
 
-    @staticmethod
-    def to_response_dict(user: DomainAdminUser) -> Dict[str, Any]:
-        created_at_ts = user.created_at.timestamp() if user.created_at else 0.0
-        updated_at_ts = user.updated_at.timestamp() if user.updated_at else 0.0
-
-        return {
-            "user_id": user.user_id,
-            "username": user.username,
-            "email": user.email,
-            "role": user.role.value,
-            "is_active": user.is_active,
-            "is_superuser": user.is_superuser,
-            "created_at": created_at_ts,
-            "updated_at": updated_at_ts,
-        }
-
     @staticmethod
     def from_pydantic_service_user(user: ServiceUser) -> DomainAdminUser:
         """Convert internal service Pydantic user to domain admin user."""
@@ -140,18 +123,6 @@ def user_creation_to_dict(creation: UserCreation) -> Dict[str, Any]:
         }
 
 
-class UserListResultMapper:
-    @staticmethod
-    def to_dict(result: UserListResult) -> Dict[str, Any]:
-        user_mapper = UserMapper()
-        return {
-            "users": [user_mapper.to_response_dict(user) for user in result.users],
-            "total": result.total,
-            "offset": result.offset,
-            "limit": result.limit,
-        }
-
-
 class SettingsMapper:
     @staticmethod
     def execution_limits_to_dict(limits: ExecutionLimits) -> dict[str, int]:

backend/app/infrastructure/mappers/admin_overview_api_mapper.py

@@ -7,13 +7,11 @@
 from confluent_kafka import Message
 
 from app.dlq.models import (
-    DLQBatchRetryResult,
     DLQFields,
     DLQMessage,
     DLQMessageFilter,
     DLQMessageStatus,
     DLQMessageUpdate,
-    DLQRetryResult,
 )
 from app.events.schema.schema_registry import SchemaRegistryManager
 from app.infrastructure.kafka.events import BaseEvent
@@ -147,44 +145,6 @@ def from_kafka_message(message: Message, schema_registry: SchemaRegistryManager)
             dlq_partition=partition if partition >= 0 else None,
         )
 
-    @staticmethod
-    def to_response_dict(message: DLQMessage) -> dict[str, object]:
-        return {
-            "event_id": message.event_id,
-            "event_type": message.event_type,
-            "event": message.event.to_dict(),
-            "original_topic": message.original_topic,
-            "error": message.error,
-            "retry_count": message.retry_count,
-            "failed_at": message.failed_at,
-            "status": message.status,
-            "age_seconds": message.age_seconds,
-            "producer_id": message.producer_id,
-            "dlq_offset": message.dlq_offset,
-            "dlq_partition": message.dlq_partition,
-            "last_error": message.last_error,
-            "next_retry_at": message.next_retry_at,
-            "retried_at": message.retried_at,
-            "discarded_at": message.discarded_at,
-            "discard_reason": message.discard_reason,
-        }
-
-    @staticmethod
-    def retry_result_to_dict(result: DLQRetryResult) -> dict[str, object]:
-        d: dict[str, object] = {"event_id": result.event_id, "status": result.status}
-        if result.error:
-            d["error"] = result.error
-        return d
-
-    @staticmethod
-    def batch_retry_result_to_dict(result: DLQBatchRetryResult) -> dict[str, object]:
-        return {
-            "total": result.total,
-            "successful": result.successful,
-            "failed": result.failed,
-            "details": [DLQMapper.retry_result_to_dict(d) for d in result.details],
-        }
-
     # Domain construction and updates
     @staticmethod
     def from_failed_event(