DEVC-601: Add app logic for meaningful error when mismatched app type is used

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 
+## [1.12.1] - 2025-01-27
+### Fixed
+- Add app logic for meaningful error when mismatched app type is used either declared at `manifest.json` or according 
+  to used event payload passed to app
+
+
 ## [1.12.0] - 2024-10-06
 ### Fixed
 - Unset the `CORVA_LOGGER.propagate = False`, so the OTel handler will be able to collect and send those logs as well

docs/antora-playbook.yml

@@ -7,7 +7,7 @@ content:
       start_path: docs
       branches: []
 #      branches: HEAD  # Use this for local development
-      tags: [v1.12.0]
+      tags: [v1.12.1]
 asciidoc:
   attributes:
     page-toclevels: 5

src/corva/handlers.py

@@ -36,11 +36,16 @@
 from corva.service import service
 from corva.service.api_sdk import CachingApiSdk, CorvaApiSdk
 from corva.service.cache_sdk import FakeInternalCacheSdk, InternalRedisSdk, UserRedisSdk
+from corva.validate_app_init import validate_app_type_context
 
 StreamEventT = TypeVar("StreamEventT", bound=StreamEvent)
 ScheduledEventT = TypeVar("ScheduledEventT", bound=ScheduledEvent)
 HANDLERS: Dict[Type[RawBaseEvent], Callable] = {}
-GENERIC_APP_EVENT_TYPES = [RawStreamEvent, RawScheduledEvent, RawTaskEvent]
+GENERIC_APP_EVENT_TYPES = (
+    RawStreamEvent,
+    RawScheduledEvent,
+    RawTaskEvent,
+)
 
 
 def get_cache_key(
@@ -72,11 +77,30 @@ def wrapper(aws_event: Any, aws_context: Any) -> List[Any]:
             user_handler=handler,
             logger=CORVA_LOGGER,
         ) as logging_ctx:
+            # Verify either current call from app_decorator or not
+            # for instance from partial rerun merge
             (
                 raw_custom_event_type,
                 custom_handler,
             ) = _get_custom_event_type_by_raw_aws_event(aws_event)
-            specific_callable = custom_handler or func
+            is_direct_app_call: bool = not custom_handler
+            data_transformation_type = raw_custom_event_type or raw_event_type
+            if merge_events:
+                aws_event = _merge_events(aws_event, data_transformation_type)
+
+            if (
+                is_direct_app_call
+                and data_transformation_type not in GENERIC_APP_EVENT_TYPES
+            ):
+                CORVA_LOGGER.warning(
+                    f"Handler for {data_transformation_type.__name__!r} "
+                    f"event not found. Skipping..."
+                )
+                return []
+
+            if is_direct_app_call:
+                # Means current app call is not RawPartialRerunMergeEvent or similar
+                validate_app_type_context(aws_event, raw_event_type)
 
             try:
                 context = CorvaContext.from_aws(
@@ -86,20 +110,8 @@ def wrapper(aws_event: Any, aws_context: Any) -> List[Any]:
                 redis_client = redis.Redis.from_url(
                     url=SETTINGS.CACHE_URL, decode_responses=True, max_connections=1
                 )
-                data_transformation_type = raw_custom_event_type or raw_event_type
-                if merge_events:
-                    aws_event = _merge_events(aws_event, data_transformation_type)
                 raw_events = data_transformation_type.from_raw_event(event=aws_event)
-
-                if (
-                    custom_handler is None
-                    and data_transformation_type not in GENERIC_APP_EVENT_TYPES
-                ):
-                    CORVA_LOGGER.warning(
-                        f"Handler for {data_transformation_type.__name__!r} "
-                        f"event not found. Skipping..."
-                    )
-                    return []
+                specific_callable = custom_handler or func
 
                 results = [
                     specific_callable(
@@ -354,7 +366,11 @@ def task(
         return functools.partial(task, handler=handler)
 
     @functools.wraps(func)
-    @functools.partial(base_handler, raw_event_type=RawTaskEvent, handler=handler)
+    @functools.partial(
+        base_handler,
+        raw_event_type=RawTaskEvent,
+        handler=handler,
+    )
     def wrapper(
         event: RawTaskEvent,
         api_key: str,

src/corva/models/base.py

@@ -1,11 +1,18 @@
 from __future__ import annotations
 
 import abc
+from enum import Enum
 from typing import Any, Sequence
 
 import pydantic
 
 
+class AppType(str, Enum):
+    STREAM = "stream"
+    TASK = "task"
+    SCHEDULER = "scheduler"
+
+
 class CorvaBaseEvent(pydantic.BaseModel):
     class Config:
         extra = pydantic.Extra.allow

src/corva/validate_app_init.py

@@ -0,0 +1,83 @@
+import contextlib
+import functools
+import json
+import os
+from typing import Any, Dict, Optional, Type
+
+import pydantic
+
+from corva.models.base import AppType, RawBaseEvent
+from corva.models.scheduled.raw import RawScheduledEvent
+from corva.models.stream.raw import RawStreamEvent
+from corva.models.task import RawTaskEvent
+
+MANIFESTED_APP_TYPE_TO_RAW_BASE_EVENT = {
+    AppType.TASK: RawTaskEvent,
+    AppType.STREAM: RawStreamEvent,
+    AppType.SCHEDULER: RawScheduledEvent,
+}
+
+
+def find_leaf_subclasses(base_class):
+    leaf_classes = []
+    for subclass in base_class.__subclasses__():
+        if not subclass.__subclasses__():
+            leaf_classes.append(subclass)
+        else:
+            leaf_classes.extend(find_leaf_subclasses(subclass))
+    return leaf_classes
+
+
+def get_event_type(aws_event: Any) -> Optional[Type[RawBaseEvent]]:
+    all_children = find_leaf_subclasses(RawBaseEvent)
+    for child_cls in all_children:
+        with contextlib.suppress(pydantic.ValidationError):
+            child_cls.from_raw_event(aws_event)
+            return child_cls
+    return None
+
+
+def validate_manifested_type(
+    manifest_app_type: 'AppType', raw_event_type: Type[RawBaseEvent]
+) -> None:
+    expected_base_event_cls = MANIFESTED_APP_TYPE_TO_RAW_BASE_EVENT[manifest_app_type]
+    if expected_base_event_cls != raw_event_type:
+        raise RuntimeError(
+            f'Application with type "{manifest_app_type.value}" '
+            f'can\'t invoke a "{raw_event_type}" handler'
+        )
+
+
+def validate_event_payload(aws_event: Any, raw_event_type: Type[RawBaseEvent]) -> None:
+
+    if event_cls := get_event_type(aws_event):
+        if not issubclass(event_cls, raw_event_type):
+            raise RuntimeError(
+                f'Application with type "{raw_event_type}" '
+                f'was invoked with "{event_cls}" event type'
+            )
+
+
+@functools.lru_cache(maxsize=1)
+def read_manifest() -> Optional[Dict[str, Any]]:
+    manifest_json_path = os.path.join(os.getcwd(), "manifest.json")
+    if os.path.exists(manifest_json_path):
+        with open(manifest_json_path, "r") as manifest_json_file:
+            return json.load(manifest_json_file)
+    return None
+
+
+def get_manifested_type() -> Optional['AppType']:
+    if manifest := read_manifest():
+        application_type = manifest.get("application", {}).get("type")
+        return AppType(application_type) if application_type else None
+    return None
+
+
+def validate_app_type_context(
+    aws_event: Any, raw_event_type: Type[RawBaseEvent]
+) -> None:
+    if manifested_type := get_manifested_type():
+        validate_manifested_type(manifested_type, raw_event_type)
+    else:
+        validate_event_payload(aws_event, raw_event_type)

src/version.py

@@ -1 +1 @@
-VERSION = "1.12.0"
+VERSION = "1.12.1"

tests/unit/conftest.py

@@ -3,6 +3,7 @@
 
 from corva.configuration import SETTINGS
 from corva.testing import TestClient
+from corva.validate_app_init import read_manifest
 
 
 @pytest.fixture(scope='function', autouse=True)
@@ -16,6 +17,13 @@ def clean_redis():
     redis_client.flushall()
 
 
+@pytest.fixture(scope='function', autouse=True)
+def clean_read_manifest_lru_cache():
+    read_manifest.cache_clear()
+    yield
+    read_manifest.cache_clear()
+
+
 @pytest.fixture(scope='function')
 def context():
     return TestClient._context