RDBC-766 Python client - time series 3 - roll-ups

Author: poissoncorp

ravendb/documents/queries/time_series.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+from typing import Optional, Dict, Any, List, Type, TypeVar, Generic
+
+from ravendb.documents.session.time_series import TimeSeriesEntry, TypedTimeSeriesEntry
+
+_T_TS_Bindable = TypeVar("_T_TS_Bindable")
+
+
+class TimeSeriesQueryResult:
+    def __init__(self, count: Optional[int] = None):
+        self.count = count
+
+    @classmethod
+    def from_json(cls, json_dict: Dict[str, Any]) -> TimeSeriesQueryResult:
+        return cls(json_dict["Count"])
+
+
+class TimeSeriesRawResult(TimeSeriesQueryResult):
+    def __init__(self, count: Optional[int] = None, results: Optional[List[TimeSeriesEntry]] = None):
+        super().__init__(count)
+        self.results = results
+
+    @classmethod
+    def from_json(cls, json_dict: Dict[str, Any]) -> TimeSeriesQueryResult:
+        return cls(
+            json_dict["Count"],
+            [TimeSeriesEntry.from_json(time_series_entry_json) for time_series_entry_json in json_dict["Results"]],
+        )
+
+    def as_typed_result(self, object_type: Type[_T_TS_Bindable]) -> TypedTimeSeriesRawResult[_T_TS_Bindable]:
+        result = TypedTimeSeriesRawResult()
+        result.count = self.count
+        result.results = [time_series_entry.as_typed_entry(object_type) for time_series_entry in self.results]
+        return result
+
+
+class TypedTimeSeriesRawResult(TimeSeriesQueryResult, Generic[_T_TS_Bindable]):
+    def __init__(
+        self, count: Optional[int] = None, results: Optional[List[TypedTimeSeriesEntry[_T_TS_Bindable]]] = None
+    ):
+        super().__init__(count)
+        self.results = results
+
+    @classmethod
+    def from_json(cls, json_dict: Dict[str, Any]) -> TimeSeriesQueryResult:
+        return cls(
+            json_dict["Count"],
+            [TypedTimeSeriesEntry.from_json(typed_ts_entry_json) for typed_ts_entry_json in json_dict["Results"]],
+        )

ravendb/documents/session/document_session.py

@@ -2324,4 +2324,8 @@ def get(
 
     def append_entry(self, entry: TypedTimeSeriesRollupEntry) -> None:
         values = entry.get_values_from_members()
-        self.append(entry.timestamp, values, entry.tag)
+        super().append(entry.timestamp, values, entry.tag)
+
+    def append(self, entry: TypedTimeSeriesRollupEntry[_T_TS_Values_Bindable]) -> None:  # todo: investigate warning
+        values = entry.get_values_from_members()
+        super().append(entry.timestamp, values, entry.tag)

ravendb/documents/session/operations/query.py

@@ -1,5 +1,6 @@
 import datetime
 import enum
+import inspect
 import logging
 from typing import Union, Optional, TypeVar, List, Type, Callable, Dict, TYPE_CHECKING
 
@@ -199,7 +200,12 @@ def deserialize(
             BeforeConversionToEntityEventArgs(session, key, object_type, document)
         )
 
-        result = Utils.initialize_object(document, object_type)
+        # By custom defined 'from_json' serializer class method
+        # todo: make separate interface to do from_json
+        if "from_json" in object_type.__dict__ and inspect.ismethod(object_type.from_json):
+            result = object_type.from_json(document)
+        else:
+            result = Utils.initialize_object(document, object_type)
         session.after_conversion_to_entity_invoke(AfterConversionToEntityEventArgs(session, key, document, result))
 
         return result

ravendb/documents/session/query.py

@@ -1585,6 +1585,12 @@ def get_query_result(self):
 
         return self._query_operation.current_query_results.create_snapshot()
 
+    def single(self) -> _T:
+        result = list(self.__execute_query_operation(2))
+        if len(result) != 1:
+            raise ValueError(f"Expected single result, got: {len(result)} ")
+        return result[0]
+
     def _aggregate_by(self, facet: FacetBase) -> None:
         for token in self._select_tokens:
             if isinstance(token, FacetToken):
@@ -1872,7 +1878,7 @@ def not_(self) -> DocumentQuery[_T]:
         self.negate_next()
         return self
 
-    def take(self, count: int) -> DocumentQuery[_T]:
+    def take(self: DocumentQuery[_T], count: int) -> DocumentQuery[_T]:
         self._take(count)
         return self
 
@@ -1888,12 +1894,6 @@ def count(self) -> int:
         query_result = self.get_query_result()
         return query_result.total_results
 
-    def single(self) -> _T:
-        result = list(self.take(2))
-        if len(result) != 1:
-            raise ValueError(f"Expected signle result, got: {len(result)} ")
-        return result[0]
-
     def where_lucene(self, field_name: str, where_clause: str, exact: bool = False) -> DocumentQuery[_T]:
         self._where_lucene(field_name, where_clause, exact)
         return self

ravendb/documents/session/time_series.py

@@ -64,7 +64,7 @@ def __init__(self, object_type: Type[_T_Values], timestamp: datetime.datetime):
 
     def _create_instance(self) -> _T_Values:
         try:
-            raise NotImplementedError()  # create object type instance
+            return Utils.try_get_new_instance(self._object_type)  # create object type instance
         except Exception as e:
             raise RavenException(f"Unable to create instance of class: {self._object_type.__name__}", e)
 
@@ -183,13 +183,13 @@ def __init__(
         timestamp: datetime.datetime = None,
         tag: str = None,
         values: List[int] = None,
-        rollup: bool = None,
+        is_rollup: bool = None,
         value: _T_TSBindable = None,
     ):
         self.timestamp = timestamp
         self.tag = tag
         self.values = values
-        self.rollup = rollup
+        self.is_rollup = is_rollup
         self.value = value
 
     @classmethod

ravendb/tests/jvm_migrated_tests/client_tests/time_series_tests/test_time_series_typed_session.py

@@ -9,7 +9,12 @@
     ConfigureTimeSeriesOperation,
     RawTimeSeriesPolicy,
 )
-from ravendb.documents.session.time_series import ITimeSeriesValuesBindable, TypedTimeSeriesEntry
+from ravendb.documents.queries.time_series import TimeSeriesRawResult
+from ravendb.documents.session.time_series import (
+    ITimeSeriesValuesBindable,
+    TypedTimeSeriesEntry,
+    TypedTimeSeriesRollupEntry,
+)
 from ravendb.infrastructure.entities import User
 from ravendb.primitives.time_series import TimeValue
 from ravendb.tests.test_base import TestBase
@@ -278,7 +283,7 @@ def test_can_work_with_rollup_time_series_2(self):
 
         config = TimeSeriesConfiguration()
         config.collections = {"users": time_series_collection_configuration}
-        config.policy_check_frequency = timedelta(seconds=1)
+        config.policy_check_frequency = timedelta(milliseconds=100)
 
         self.store.maintenance.send(ConfigureTimeSeriesOperation(config))
         self.store.time_series.register_type(User, StockPrice)
@@ -313,3 +318,92 @@ def test_can_work_with_rollup_time_series_2(self):
             self.assertIsNotNone(r.max)
             self.assertIsNotNone(r.count)
             self.assertIsNotNone(r.average)
+
+    def test_can_work_with_rollup_time_series(self):
+        raw_hours = 24
+        raw = RawTimeSeriesPolicy(TimeValue.of_hours(raw_hours))
+
+        p1 = TimeSeriesPolicy("By6Hours", TimeValue.of_hours(6), TimeValue.of_hours(raw_hours * 4))
+        p2 = TimeSeriesPolicy("By1Day", TimeValue.of_days(1), TimeValue.of_hours(raw_hours * 5))
+        p3 = TimeSeriesPolicy("By30Minutes", TimeValue.of_minutes(30), TimeValue.of_hours(raw_hours * 2))
+        p4 = TimeSeriesPolicy("By1Hour", TimeValue.of_hours(1), TimeValue.of_hours(raw_hours * 3))
+
+        users_config = TimeSeriesCollectionConfiguration()
+        users_config.raw_policy = raw
+        users_config.policies = [p1, p2, p3, p4]
+
+        config = TimeSeriesConfiguration()
+        config.collections = {"Users": users_config}
+        config.policy_check_frequency = timedelta(milliseconds=100)
+
+        self.store.maintenance.send(ConfigureTimeSeriesOperation(config))
+        self.store.time_series.register_type(User, StockPrice)
+
+        # please notice we don't modify server time here!
+
+        now = datetime.utcnow()
+        base_line = RavenTestHelper.utc_today() - timedelta(days=12)
+
+        total = TimeValue.of_days(12).value // 60
+
+        with self.store.open_session() as session:
+            session.store(User(name="Karmel"), "users/karmel")
+
+            ts = session.typed_time_series_for(StockPrice, "users/karmel")
+            for i in range(total + 1):
+                open = i
+                close = i + 100_000
+                high = i + 200_000
+                low = i + 300_000
+                volume = i + 400_000
+                ts.append(
+                    base_line + timedelta(minutes=i), StockPrice(open, close, high, low, volume), "watches/fitbit"
+                )
+
+            session.save_changes()
+
+        time.sleep(1.5)  # wait for rollups
+
+        with self.store.open_session() as session:
+            query = (
+                session.advanced.raw_query(
+                    "declare timeseries out()\n"
+                    "{\n"
+                    "    from StockPrices\n"
+                    "    between $start and $end\n"
+                    "}\n"
+                    "from Users as u\n"
+                    "select out()",
+                    TimeSeriesRawResult,
+                )
+                .add_parameter("start", base_line - timedelta(days=1))
+                .add_parameter("end", now + timedelta(days=1))
+            )
+            result_raw = query.single()
+            result = result_raw.as_typed_result(StockPrice)
+
+            self.assertGreater(len(result.results), 0)
+
+            for res in result.results:
+                if res.is_rollup:
+                    self.assertGreater(len(res.values), 0)
+                    self.assertGreater(len(res.value.low), 0)
+                    self.assertGreater(len(res.value.high), 0)
+                else:
+                    self.assertEqual(5, len(res.values))
+
+        now = datetime.utcnow()
+
+        with self.store.open_session() as session:
+            ts = session.time_series_rollup_for(StockPrice, "users/karmel", p1.name)
+            a = TypedTimeSeriesRollupEntry(StockPrice, datetime.utcnow())
+            a.max.close = 1
+            ts.append(a)
+            session.save_changes()
+
+        with self.store.open_session() as session:
+            ts = session.time_series_rollup_for(StockPrice, "users/karmel", p1.name)
+
+            res = ts.get(now - timedelta(milliseconds=1), now + timedelta(days=1))
+            self.assertEqual(1, len(res))
+            self.assertEqual(1, res[0].max.close)

ravendb/tests/jvm_migrated_tests/spatial_tests/test_spatial.py

@@ -45,7 +45,7 @@ def __init__(self, Id: str = None, date: datetime.datetime = None, latitude: flo
     def from_json(cls, json_dict: Dict) -> MyProjection:
         return cls(
             json_dict["Id"],
-            datetime.datetime.fromisoformat(json_dict["date"]),
+            datetime.datetime.fromisoformat(json_dict["date"]) if "date" in json_dict else None,
             json_dict["latitude"],
             json_dict["longitude"],
         )