addressing PR comments

Author: corva-alex-s

docs/modules/ROOT/examples/merging/tutorial001.py

@@ -1,9 +1,10 @@
-from corva import Api, StreamTimeEvent, stream
+from corva import Api, StreamTimeEvent, stream, Cache
 
 
 # imagine we actually have 3 incoming events with 3 records each
 @stream(merge_events=True)
-def app(event: StreamTimeEvent, api: Api):
+def app(event: StreamTimeEvent, api: Api, cache: Cache):
     # since we passed merge_events=True all 3 incoming events
     # and their records will be merged into a single event with 9 records
-    assert len(event.records) == 9  # this will not fail
+    assert len(event.records) != 9  # this will not fail
+    return event

docs/modules/ROOT/examples/merging/tutorial002.py

@@ -0,0 +1,6 @@
+from corva import Api, ScheduledNaturalTimeEvent, scheduled, Cache
+
+
+@scheduled(merge_events=True)
+def app(event: ScheduledNaturalTimeEvent, api: Api, cache: Cache):
+    return event

docs/modules/ROOT/pages/index.adoc

@@ -539,6 +539,10 @@ Optionally we can ask to merge them into one event by providing `merge_events=Tr
 ----
 include::example$merging/tutorial001.py[]
 ----
+[source,python]
+----
+include::example$merging/tutorial002.py[]
+----
 Usually this is needed to save some IO operations by processing data in bigger batches.
 Use this parameter with care, in pessimistic scenario you can receive too much data, try to
 process it in "one go" and fail with timeout. In that case your app will be automatically

src/corva/handlers.py

@@ -19,6 +19,7 @@
 
 import pydantic
 import redis
+from typing_extensions import assert_never
 
 from corva.api import Api
 from corva.configuration import SETTINGS
@@ -60,7 +61,7 @@ def base_handler(
     func: Callable,
     raw_event_type: Type[RawBaseEvent],
     handler: Optional[logging.Handler],
-    merge_events: Optional[bool] = False,
+    merge_events: bool = False,
 ) -> Callable[[Any, Any], List[Any]]:
     @functools.wraps(func)
     def wrapper(aws_event: Any, aws_context: Any) -> List[Any]:
@@ -125,14 +126,14 @@ def stream(
     func: Optional[Callable[[StreamEventT, Api, UserRedisSdk], Any]] = None,
     *,
     handler: Optional[logging.Handler] = None,
-    merge_events: Optional[bool] = False,
+    merge_events: bool = False,
 ) -> Callable:
     """Runs stream app.
 
     Arguments:
         handler: logging handler to include in Corva logger.
         merge_events: if True - merge all incoming events into one before
-        passing them to func
+          passing them to func
     """
 
     if func is None:
@@ -234,12 +235,14 @@ def scheduled(
     func: Optional[Callable[[ScheduledEventT, Api, UserRedisSdk], Any]] = None,
     *,
     handler: Optional[logging.Handler] = None,
-    merge_events: Optional[bool] = False,
+    merge_events: bool = False,
 ) -> Callable:
     """Runs scheduled app.
 
     Arguments:
         handler: logging handler to include in Corva logger.
+        merge_events: if True - merge all incoming events into one before
+          passing them to func
     """
 
     if func is None:
@@ -581,20 +584,11 @@ def _merge_events(aws_event: Any, data_transformation_type: Type[RawBaseEvent])
             if is_depth
             else ("schedule_start", "schedule_end")
         )
-        min_event_start, max_event_end = (
-            aws_event[0][event_start],
-            aws_event[0].get(event_end),
+        min_event_start = min(e[event_start] for e in aws_event)
+        max_event_end = max(
+            (e[event_end] for e in aws_event if e.get(event_end) is not None),
+            default=None,
         )
-        for event in aws_event[1:]:
-            if event[event_start] < min_event_start:
-                min_event_start = event[event_start]
-            if (
-                max_event_end
-                and event.get(event_end)
-                and event[event_end] > max_event_end
-            ):
-                max_event_end = event[event_end]
-
         aws_event[0][event_start] = min_event_start
         if max_event_end:
             aws_event[0][event_end] = max_event_end
@@ -608,9 +602,6 @@ def _merge_events(aws_event: Any, data_transformation_type: Type[RawBaseEvent])
         aws_event = [aws_event[0]]
         return aws_event
 
-    # unexpected event type, raise an exception
-    raise RuntimeError(
-        "merge_events parameter was passed to app type other than scheduled "
-        "or stream. Merge strategy is not implemented for "
-        f"{data_transformation_type} event types."
-    )
+    else:
+        # unexpected event type, raise an exception
+        assert_never(data_transformation_type)  # type: ignore

tests/unit/test_docs/test_merging.py

@@ -0,0 +1,91 @@
+from typing import List, Union, Dict
+
+import pytest
+
+from corva import StreamTimeEvent, ScheduledDataTimeEvent
+from corva.configuration import SETTINGS
+from corva.models.scheduled.raw import RawScheduledDataTimeEvent
+from corva.models.scheduled.scheduler_type import SchedulerType
+from corva.models.stream.log_type import LogType
+from corva.models.stream.raw import RawStreamTimeEvent, RawTimeRecord, RawMetadata, RawAppMetadata
+from docs.modules.ROOT.examples.merging import tutorial001, tutorial002
+
+
+def test_tutorial001(context):
+    """
+    merge_events parameter is merging records for "stream" apps
+
+    When 3 events with 2 records each are sent and @stream decorator
+    has optional "merge_events" param set to True - we're supposed
+    to merge incoming events into one event with all records(6 in our
+    case) combined
+    """
+
+    event = []
+    timestamp = 1
+    # generate 3 events with 2 records each
+    for _ in range(3):
+        event.extend(
+            [
+                RawStreamTimeEvent(
+                    records=[
+                        RawTimeRecord(
+                            collection=str(),
+                            timestamp=timestamp,
+                            asset_id=1,
+                            company_id=1,
+                        ),
+                        RawTimeRecord(
+                            collection=str(),
+                            timestamp=timestamp + 1,
+                            asset_id=1,
+                            company_id=1,
+                        ),
+                    ],
+                    metadata=RawMetadata(
+                        app_stream_id=1,
+                        apps={SETTINGS.APP_KEY: RawAppMetadata(app_connection_id=1)},
+                        log_type=LogType.time,
+                    ),
+                ).dict()
+            ]
+        )
+        timestamp += 2
+
+    result_event: StreamTimeEvent = tutorial001.app(event, context)[0]
+    assert len(result_event.records) == 6, "records were not merged into a single event"
+
+
+@pytest.mark.parametrize(
+    "time_ranges, flat",
+    (
+        [((60, 120), (61, None), (62, 122)), True],
+        [((61, None), (60, None), (62, None)), False],
+    ),
+)
+def test_tutorial002(context, time_ranges, flat):
+
+    event: List[Union[List, Dict]] = []
+    for schedule_start, schedule_end in time_ranges:
+        event.append(
+            RawScheduledDataTimeEvent(
+                asset_id=int(),
+                interval=60,
+                schedule=int(),
+                schedule_start=schedule_start,
+                schedule_end=schedule_end,
+                app_connection=int(),
+                app_stream=int(),
+                company=int(),
+                scheduler_type=SchedulerType.data_time,
+            ).dict(by_alias=True, exclude_unset=True)
+        )
+    if not flat:
+        event = [event]
+
+    result_event: ScheduledDataTimeEvent = tutorial002.app(event, context)[0]
+
+    assert result_event.start_time == 1
+    assert result_event.end_time == 60
+    max_schedule_value = time_ranges[-1][-1]
+    assert result_event.schedule_end == max_schedule_value  # type: ignore[attr-defined]

tests/unit/test_merge_events.py

@@ -10,5 +10,5 @@ def test_events_not_merged_on_unexpected_event_type():
     passed - fail with RuntimeError
     """
     aws_event = [{"sample": 1}, {"sample2": 2}]
-    with pytest.raises(RuntimeError):
+    with pytest.raises(AssertionError):
         _merge_events(aws_event, RawTaskEvent)

tests/unit/test_scheduled_app.py

@@ -1,6 +1,5 @@
 import logging
 import re
-from typing import Dict, List, Union
 
 import pytest
 import redis
@@ -533,41 +532,3 @@ def scheduled_app(event, api, cache):
 
     with pytest.raises(redis.exceptions.ConnectionError):
         scheduled_app(event, context)
-
-
-@pytest.mark.parametrize(
-    "time_ranges, flat",
-    (
-        [((60, 120), (61, None), (62, 122)), True],
-        [((61, None), (60, None), (62, None)), False],
-    ),
-)
-def test_merge_events_scheduled_event(context, requests_mock, time_ranges, flat):
-    @scheduled(merge_events=True)
-    def scheduled_app(_event: ScheduledDataTimeEvent, api, state):
-        return _event
-
-    event: List[Union[List, Dict]] = []
-    for schedule_start, schedule_end in time_ranges:
-        event.append(
-            RawScheduledDataTimeEvent(
-                asset_id=int(),
-                interval=60,
-                schedule=int(),
-                schedule_start=schedule_start,
-                schedule_end=schedule_end,
-                app_connection=int(),
-                app_stream=int(),
-                company=int(),
-                scheduler_type=SchedulerType.data_time,
-            ).dict(by_alias=True, exclude_unset=True)
-        )
-    if not flat:
-        event = [event]
-
-    result_event: ScheduledDataTimeEvent = scheduled_app(event, context)[0]
-
-    assert result_event.start_time == 1
-    assert result_event.end_time == 60
-    max_schedule_value = time_ranges[-1][-1]
-    assert result_event.schedule_end == max_schedule_value  # type: ignore[attr-defined]

tests/unit/test_stream_app.py

@@ -615,52 +615,3 @@ def stream_app(event, api, cache):
     result_event: StreamDepthEvent = stream_app(event, context)[0]
 
     assert result_event.log_identifier == 'log_identifier'
-
-
-def test_merge_events_stream_event(context):
-    """
-    merge_events parameter is merging records for "stream" apps
-
-    When 3 events with 2 records each are sent and @stream decorator
-    has optional "merge_events" param set to True - we're supposed
-    to merge incoming events into one event with all records(6 in our
-    case) combined
-    """
-
-    @stream(merge_events=True)
-    def stream_app(event, api, cache):
-        return event
-
-    event = []
-    timestamp = 1
-    # generate 3 events with 2 records each
-    for _ in range(3):
-        event.extend(
-            [
-                RawStreamTimeEvent(
-                    records=[
-                        RawTimeRecord(
-                            collection=str(),
-                            timestamp=timestamp,
-                            asset_id=1,
-                            company_id=1,
-                        ),
-                        RawTimeRecord(
-                            collection=str(),
-                            timestamp=timestamp + 1,
-                            asset_id=1,
-                            company_id=1,
-                        ),
-                    ],
-                    metadata=RawMetadata(
-                        app_stream_id=1,
-                        apps={SETTINGS.APP_KEY: RawAppMetadata(app_connection_id=1)},
-                        log_type=LogType.time,
-                    ),
-                ).dict()
-            ]
-        )
-        timestamp += 2
-
-    result_event: StreamTimeEvent = stream_app(event, context)[0]
-    assert len(result_event.records) == 6, "records were not merged into a single event"