better types

Author: HardMax71

backend/app/api/routes/events.py

@@ -325,7 +325,7 @@ async def replay_aggregate_events(
                 user_id=admin.user_id,
             )
             await kafka_event_service.publish_event(
-                event_type=f"replay.{event.event_type}",
+                event_type=event.event_type,
                 payload=event.payload,
                 aggregate_id=aggregate_id,
                 correlation_id=replay_correlation_id,

backend/app/api/routes/execution.py

@@ -1,4 +1,3 @@
-from dataclasses import asdict
 from datetime import datetime, timezone
 from typing import Annotated
 from uuid import uuid4
@@ -11,10 +10,8 @@
 from app.core.exceptions import IntegrationException
 from app.core.tracing import EventAttributes, add_span_attributes
 from app.core.utils import get_client_ip
-from app.domain.enums.common import ErrorType
 from app.domain.enums.events import EventType
 from app.domain.enums.execution import ExecutionStatus
-from app.domain.enums.storage import ExecutionErrorType
 from app.domain.enums.user import UserRole
 from app.infrastructure.kafka.events.base import BaseEvent
 from app.infrastructure.kafka.events.metadata import EventMetadata
@@ -31,7 +28,6 @@
     ExecutionResponse,
     ExecutionResult,
     ResourceLimits,
-    ResourceUsage,
     RetryExecutionRequest,
 )
 from app.schemas_pydantic.user import UserResponse
@@ -55,35 +51,7 @@ async def get_execution_with_access(
     if domain_exec.user_id and domain_exec.user_id != current_user.user_id and current_user.role != UserRole.ADMIN:
         raise HTTPException(status_code=403, detail="Access denied")
 
-    # Map domain to Pydantic for dependency consumer
-    ru = None
-    if domain_exec.resource_usage is not None:
-        ru = ResourceUsage(**vars(domain_exec.resource_usage))
-    # Map error_type to public ErrorType in API model via mapper rules
-    error_type = (
-        (
-            ErrorType.SCRIPT_ERROR
-            if domain_exec.error_type == ExecutionErrorType.SCRIPT_ERROR
-            else ErrorType.SYSTEM_ERROR
-        )
-        if domain_exec.error_type is not None
-        else None
-    )
-    return ExecutionInDB(
-        execution_id=domain_exec.execution_id,
-        script=domain_exec.script,
-        status=domain_exec.status,
-        stdout=domain_exec.stdout,
-        stderr=domain_exec.stderr,
-        lang=domain_exec.lang,
-        lang_version=domain_exec.lang_version,
-        resource_usage=ru,
-        user_id=domain_exec.user_id,
-        exit_code=domain_exec.exit_code,
-        error_type=error_type,
-        created_at=domain_exec.created_at,
-        updated_at=domain_exec.updated_at,
-    )
+    return ExecutionInDB.model_validate(domain_exec)
 
 
 @router.post("/execute", response_model=ExecutionResponse)
@@ -269,24 +237,14 @@ async def retry_execution(
 async def get_execution_events(
     execution: Annotated[ExecutionInDB, Depends(get_execution_with_access)],
     event_service: FromDishka[EventService],
-    event_types: str | None = Query(None, description="Comma-separated event types to filter"),
+    event_types: list[EventType] | None = Query(None, description="Event types to filter"),
     limit: int = Query(100, ge=1, le=1000),
 ) -> list[ExecutionEventResponse]:
     """Get all events for an execution."""
-    event_type_list = None
-    if event_types:
-        event_type_list = [t.strip() for t in event_types.split(",")]
-
     events = await event_service.get_events_by_aggregate(
-        aggregate_id=execution.execution_id, event_types=event_type_list, limit=limit
+        aggregate_id=execution.execution_id, event_types=event_types, limit=limit
     )
-
-    return [
-        ExecutionEventResponse(
-            event_id=event.event_id, event_type=event.event_type, timestamp=event.timestamp, payload=event.payload
-        )
-        for event in events
-    ]
+    return [ExecutionEventResponse.model_validate(e) for e in events]
 
 
 @router.get("/user/executions", response_model=ExecutionListResponse)
@@ -337,7 +295,7 @@ async def get_k8s_resource_limits(
 ) -> ResourceLimits:
     try:
         limits = await execution_service.get_k8s_resource_limits()
-        return ResourceLimits(**asdict(limits))
+        return ResourceLimits.model_validate(limits)
     except Exception as e:
         raise HTTPException(status_code=500, detail="Failed to retrieve resource limits") from e
 

backend/app/db/repositories/event_repository.py

@@ -9,6 +9,7 @@
 from app.core.logging import logger
 from app.core.tracing import EventAttributes
 from app.core.tracing.utils import add_span_attributes
+from app.domain.enums.events import EventType
 from app.domain.enums.user import UserRole
 from app.domain.events import (
     ArchivedEvent,
@@ -114,11 +115,11 @@ async def get_events_by_type(
         return [self.mapper.from_mongo_document(doc) for doc in docs]
 
     async def get_events_by_aggregate(
-        self, aggregate_id: str, event_types: list[str] | None = None, limit: int = 100
+        self, aggregate_id: str, event_types: list[EventType] | None = None, limit: int = 100
     ) -> list[Event]:
         query: dict[str, Any] = {EventFields.AGGREGATE_ID: aggregate_id}
         if event_types:
-            query[EventFields.EVENT_TYPE] = {"$in": event_types}
+            query[EventFields.EVENT_TYPE] = {"$in": [t.value for t in event_types]}
 
         cursor = self._collection.find(query).sort(EventFields.TIMESTAMP, ASCENDING).limit(limit)
         docs = await cursor.to_list(length=limit)

backend/app/db/repositories/sse_repository.py

@@ -1,41 +1,30 @@
+from datetime import datetime, timezone
+
 from app.core.database_context import Collection, Database
+from app.domain.enums.execution import ExecutionStatus
 from app.domain.events.event_models import CollectionNames
 from app.domain.execution import DomainExecution
-from app.domain.sse import SSEEventDomain, SSEExecutionStatusDomain
+from app.domain.sse import SSEExecutionStatusDomain
 from app.infrastructure.mappers import SSEMapper
 
 
 class SSERepository:
-    def __init__(self, database: Database):
+    def __init__(self, database: Database) -> None:
         self.db = database
         self.executions_collection: Collection = self.db.get_collection(CollectionNames.EXECUTIONS)
-        self.events_collection: Collection = self.db.get_collection(CollectionNames.EVENTS)
         self.mapper = SSEMapper()
 
     async def get_execution_status(self, execution_id: str) -> SSEExecutionStatusDomain | None:
-        execution = await self.executions_collection.find_one(
-            {"execution_id": execution_id}, {"status": 1, "execution_id": 1, "_id": 0}
-        )
-
-        if execution:
-            return self.mapper.to_execution_status(execution_id, execution.get("status", "unknown"))
-        return None
-
-    async def get_execution_events(self, execution_id: str, limit: int = 100, skip: int = 0) -> list[SSEEventDomain]:
-        cursor = (
-            self.events_collection.find({"aggregate_id": execution_id}).sort("timestamp", 1).skip(skip).limit(limit)
+        doc = await self.executions_collection.find_one(
+            {"execution_id": execution_id}, {"status": 1, "_id": 0}
         )
-
-        events: list[SSEEventDomain] = []
-        async for event in cursor:
-            events.append(self.mapper.event_from_mongo_document(event))
-        return events
-
-    async def get_execution_for_user(self, execution_id: str, user_id: str) -> DomainExecution | None:
-        doc = await self.executions_collection.find_one({"execution_id": execution_id, "user_id": user_id})
         if not doc:
             return None
-        return self.mapper.execution_from_mongo_document(doc)
+        return SSEExecutionStatusDomain(
+            execution_id=execution_id,
+            status=ExecutionStatus(doc["status"]),
+            timestamp=datetime.now(timezone.utc).isoformat(),
+        )
 
     async def get_execution(self, execution_id: str) -> DomainExecution | None:
         doc = await self.executions_collection.find_one({"execution_id": execution_id})

backend/app/dlq/models.py

@@ -3,6 +3,7 @@
 from typing import Any
 
 from app.core.utils import StringEnum
+from app.domain.enums.events import EventType
 from app.infrastructure.kafka.events import BaseEvent
 
 
@@ -89,9 +90,9 @@ def age_seconds(self) -> float:
         return (datetime.now(timezone.utc) - self.failed_at).total_seconds()
 
     @property
-    def event_type(self) -> str:
+    def event_type(self) -> EventType:
         """Get event type from the event."""
-        return str(self.event.event_type)
+        return self.event.event_type
 
 
 @dataclass

backend/app/domain/events/event_models.py

@@ -3,6 +3,7 @@
 from typing import Any
 
 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]
@@ -77,7 +78,7 @@ class Event:
     """Domain model for an event."""
 
     event_id: str
-    event_type: str
+    event_type: EventType
     event_version: str
     timestamp: datetime
     metadata: EventMetadata

backend/app/domain/sse/models.py

@@ -3,6 +3,8 @@
 from dataclasses import dataclass
 from datetime import datetime
 
+from app.domain.enums.execution import ExecutionStatus
+
 
 @dataclass
 class ShutdownStatus:
@@ -31,7 +33,7 @@ class SSEHealthDomain:
 @dataclass
 class SSEExecutionStatusDomain:
     execution_id: str
-    status: str
+    status: ExecutionStatus
     timestamp: str
 
 

backend/app/domain/user/settings_models.py

@@ -26,8 +26,6 @@ class DomainEditorSettings:
     use_tabs: bool = False
     word_wrap: bool = True
     show_line_numbers: bool = True
-    # fixed, non-configurable editor attributes are omitted from domain
-    # as they are UI concerns
 
 
 @dataclass
@@ -55,24 +53,6 @@ class DomainUserSettingsUpdate:
     editor: Optional[DomainEditorSettings] = None
     custom_settings: Optional[Dict[str, Any]] = None
 
-    def to_update_dict(self) -> Dict[str, Any]:
-        out: Dict[str, Any] = {}
-        if self.theme is not None:
-            out["theme"] = self.theme
-        if self.timezone is not None:
-            out["timezone"] = self.timezone
-        if self.date_format is not None:
-            out["date_format"] = self.date_format
-        if self.time_format is not None:
-            out["time_format"] = self.time_format
-        if self.notifications is not None:
-            out["notifications"] = self.notifications
-        if self.editor is not None:
-            out["editor"] = self.editor
-        if self.custom_settings is not None:
-            out["custom_settings"] = self.custom_settings
-        return out
-
 
 @dataclass
 class DomainSettingChange:
@@ -94,7 +74,7 @@ class DomainSettingsEvent:
 @dataclass
 class DomainSettingsHistoryEntry:
     timestamp: datetime
-    event_type: str
+    event_type: EventType
     field: str
     old_value: Any
     new_value: Any

backend/app/infrastructure/mappers/execution_api_mapper.py

@@ -1,10 +1,6 @@
 from __future__ import annotations
 
-from typing import Optional
-
-from app.domain.enums.common import ErrorType
-from app.domain.enums.storage import ExecutionErrorType
-from app.domain.execution import DomainExecution, ResourceUsageDomain
+from app.domain.execution import DomainExecution
 from app.schemas_pydantic.execution import ExecutionResponse, ExecutionResult
 from app.schemas_pydantic.execution import ResourceUsage as ResourceUsageSchema
 
@@ -20,18 +16,9 @@ def to_response(e: DomainExecution) -> ExecutionResponse:
     @staticmethod
     def to_result(e: DomainExecution) -> ExecutionResult:
         ru = None
-        if isinstance(e.resource_usage, ResourceUsageDomain):
+        if e.resource_usage is not None:
             ru = ResourceUsageSchema(**e.resource_usage.to_dict())
 
-        # Map domain ExecutionErrorType -> public ErrorType
-        def _map_error(t: Optional[ExecutionErrorType]) -> Optional[ErrorType]:
-            if t is None:
-                return None
-            if t == ExecutionErrorType.SCRIPT_ERROR:
-                return ErrorType.SCRIPT_ERROR
-            # TIMEOUT, RESOURCE_LIMIT, SYSTEM_ERROR, PERMISSION_DENIED -> SYSTEM_ERROR class
-            return ErrorType.SYSTEM_ERROR
-
         return ExecutionResult(
             execution_id=e.execution_id,
             status=e.status,
@@ -41,5 +28,5 @@ def _map_error(t: Optional[ExecutionErrorType]) -> Optional[ErrorType]:
             lang_version=e.lang_version,
             resource_usage=ru,
             exit_code=e.exit_code,
-            error_type=_map_error(e.error_type),
+            error_type=e.error_type,
         )

backend/app/infrastructure/mappers/sse_mapper.py

@@ -1,46 +1,27 @@
 from datetime import datetime, timezone
-from typing import Any, Dict
+from typing import Any
 
 from app.domain.enums.execution import ExecutionStatus
 from app.domain.execution import DomainExecution, ResourceUsageDomain
-from app.domain.sse import SSEEventDomain, SSEExecutionStatusDomain
 
 
 class SSEMapper:
-    """Mapper for SSE-related domain models and MongoDB documents."""
+    """Mapper for SSE-related domain models from MongoDB documents."""
 
-    # Execution status (lightweight)
     @staticmethod
-    def to_execution_status(execution_id: str, status: str) -> SSEExecutionStatusDomain:
-        return SSEExecutionStatusDomain(
-            execution_id=execution_id,
-            status=status,
-            timestamp=datetime.now(timezone.utc).isoformat(),
-        )
-
-    # Execution events
-    @staticmethod
-    def event_from_mongo_document(doc: Dict[str, Any]) -> SSEEventDomain:
-        return SSEEventDomain(
-            aggregate_id=str(doc.get("aggregate_id", "")),
-            timestamp=doc["timestamp"],
-        )
-
-    # Executions
-    @staticmethod
-    def execution_from_mongo_document(doc: Dict[str, Any]) -> DomainExecution:
-        sv = doc.get("status")
+    def execution_from_mongo_document(doc: dict[str, Any]) -> DomainExecution:
+        resource_usage_data = doc.get("resource_usage")
         return DomainExecution(
             execution_id=str(doc.get("execution_id")),
             script=str(doc.get("script", "")),
-            status=ExecutionStatus(str(sv)),
+            status=ExecutionStatus(str(doc.get("status"))),
             stdout=doc.get("stdout"),
             stderr=doc.get("stderr"),
             lang=str(doc.get("lang", "python")),
             lang_version=str(doc.get("lang_version", "3.11")),
             created_at=doc.get("created_at", datetime.now(timezone.utc)),
             updated_at=doc.get("updated_at", datetime.now(timezone.utc)),
-            resource_usage=ResourceUsageDomain.from_dict(doc.get("resource_usage") or {}),
+            resource_usage=ResourceUsageDomain.from_dict(resource_usage_data) if resource_usage_data else None,
             user_id=doc.get("user_id"),
             exit_code=doc.get("exit_code"),
             error_type=doc.get("error_type"),