RDBC-766 Python client - time series 3 - stage #1 + test for roll-ups

Author: poissoncorp

ravendb/documents/operations/indexes.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 import enum
 import json
-from typing import List, TYPE_CHECKING
+from typing import List, TYPE_CHECKING, Optional
 
 import requests
 
@@ -556,7 +556,7 @@ def get_raft_unique_request_id(self) -> str:
 
 
 class GetTermsOperation(MaintenanceOperation[List[str]]):
-    def __init__(self, index_name: str, field: str, from_value: str, page_size: int = None):
+    def __init__(self, index_name: str, field: str, from_value: Optional[str], page_size: int = None):
         if index_name is None:
             raise ValueError("Index name cannot be None")
         if field is None:
@@ -584,8 +584,8 @@ def create_request(self, server_node) -> requests.Request:
                 f"{server_node.url}/databases/{server_node.database}"
                 f"/indexes/terms?name={Utils.escape(self.__index_name, False, False)}"
                 f"&field={Utils.escape(self.__field, False, False)}"
-                f"&fromValue={self.__from_value if self.__from_value is not None else ''}"
-                f"&pageSize={self.__page_size if self.__page_size is not None else ''}",
+                f"&fromValue={self.__from_value or ''}"
+                f"&pageSize={self.__page_size or ''}",
             )
 
         def set_response(self, response: str, from_cache: bool) -> None:

ravendb/documents/operations/time_series.py

@@ -32,33 +32,25 @@ def __init__(
         if not name or name.isspace():
             raise ValueError("Name cannot be None or empty")
 
-        if aggregation_time is not None and aggregation_time.compare_to(TimeValue.ZERO()) <= 0:
+        if aggregation_time and aggregation_time.compare_to(TimeValue.ZERO()) <= 0:
             raise ValueError("Aggregation time must be greater than zero")
 
-        if retention_time is not None and retention_time.compare_to(TimeValue.ZERO()) <= 0:
+        if retention_time is None or retention_time.compare_to(TimeValue.ZERO()) <= 0:
             raise ValueError("Retention time must be greater than zero")
 
         self.retention_time = retention_time
         self.aggregation_time = aggregation_time
 
         self.name = name
 
-    @classmethod
-    def from_json(cls, json_dict: Dict[str, Any]) -> TimeSeriesPolicy:
-        return cls(
-            json_dict["Name"],  # todo: Invalid deserialization
-            Utils.string_to_timedelta(json_dict["AggregationTime"]),
-            Utils.string_to_timedelta(json_dict["RetentionTime"]),
-        )
-
     def get_time_series_name(self, raw_name: str) -> str:
         return raw_name + TimeSeriesConfiguration.TIME_SERIES_ROLLUP_SEPARATOR + self.name
 
     def to_json(self) -> Dict[str, Any]:
         return {
-            "Name": self.name,  # todo: Invalid serialization
-            "AggregationTime": Utils.timedelta_to_str(self.aggregation_time),
-            "RetentionTime": Utils.timedelta_to_str(self.retention_time),
+            "Name": self.name,
+            "AggregationTime": self.aggregation_time.to_json() if self.aggregation_time else None,
+            "RetentionTime": self.retention_time.to_json(),
         }
 
 
@@ -72,30 +64,20 @@ def DEFAULT_POLICY(cls) -> RawTimeSeriesPolicy:
     def __init__(self, retention_time: TimeValue = TimeValue.MAX_VALUE()):
         if retention_time.compare_to(TimeValue.ZERO()) <= 0:
             raise ValueError("Retention time must be greater than zero")
-
-        self.name = self.POLICY_STRING
-        self.retention_time = retention_time
+        super().__init__(self.POLICY_STRING, retention_time=retention_time)
 
 
 class TimeSeriesCollectionConfiguration:
     def __init__(
         self,
-        disabled: Optional[bool] = None,
+        disabled: Optional[bool] = False,
         policies: Optional[List[TimeSeriesPolicy]] = None,
         raw_policy: Optional[RawTimeSeriesPolicy] = RawTimeSeriesPolicy.DEFAULT_POLICY(),
     ):
         self.disabled = disabled
         self.policies = policies
         self.raw_policy = raw_policy
 
-    @classmethod
-    def from_json(cls, json_dict: Dict[str, Any]) -> TimeSeriesCollectionConfiguration:
-        return cls(
-            json_dict["Disabled"],
-            [TimeSeriesPolicy.from_json(policy_json) for policy_json in json_dict["Policies"]],
-            RawTimeSeriesPolicy.from_json(json_dict["RawPolicy"]),
-        )
-
     def to_json(self) -> Dict[str, Any]:
         return {
             "Disabled": self.disabled,
@@ -348,12 +330,12 @@ def to_json(self) -> Dict[str, Any]:
 
     def append(self, append_operation: AppendOperation) -> None:
         if self._appends is None:
-            self._appends = []
-        filtered = list(filter(lambda x: x.timestamp == append_operation.timestamp, self._appends))
+            self._appends = [] # todo: perf
+        filtered = self._appends
 
-        if len(filtered) != 0:
-            # element with given timestamp already exists - remove and retry add operation
-            self._appends.remove(filtered.pop())
+        # if len(filtered) != 0:
+        #     # element with given timestamp already exists - remove and retry add operation
+        #     self._appends.remove(filtered.pop())
 
         self._appends.append(append_operation)
 
@@ -699,3 +681,36 @@ def create_request(self, node: ServerNode) -> requests.Request:
 
         def set_response(self, response: Optional[str], from_cache: bool) -> None:
             self.result = TimeSeriesStatistics.from_json(json.loads(response))
+
+
+class ConfigureTimeSeriesOperation(MaintenanceOperation[ConfigureTimeSeriesOperationResult]):
+    def __init__(self, configuration: TimeSeriesConfiguration):
+        if not configuration:
+            raise ValueError("Configuration cannot be None")
+
+        self._configuration = configuration
+
+    def get_command(self, conventions: "DocumentConventions") -> "RavenCommand[ConfigureTimeSeriesOperationResult]":
+        return self.ConfigureTimeSeriesCommand(self._configuration)
+
+    class ConfigureTimeSeriesCommand(RavenCommand[ConfigureTimeSeriesOperationResult], RaftCommand):
+        def __init__(self, configuration: TimeSeriesConfiguration):
+            super().__init__(ConfigureTimeSeriesOperationResult)
+            self._configuration = configuration
+
+        def is_read_request(self) -> bool:
+            return False
+
+        def create_request(self, node: ServerNode) -> requests.Request:
+            request = requests.Request("POST", f"{node.url}/databases/{node.database}/admin/timeseries/config")
+            request.data = self._configuration.to_json()
+            return request
+
+        def set_response(self, response: Optional[str], from_cache: bool) -> None:
+            if not response:
+                self._throw_invalid_response()
+
+            self.result = ConfigureTimeSeriesOperationResult.from_json(json.loads(response))
+
+        def get_raft_unique_request_id(self) -> str:
+            return RaftIdGenerator.new_id()

ravendb/documents/session/document_session.py

@@ -171,7 +171,7 @@ def execute_all_pending_lazy_operations(self) -> ResponseTimeInformation:
         requests = []
         for i in range(len(self._pending_lazy_operations)):
             # todo: pending lazy operation create request - WIP
-            req = self._pending_lazy_operations[i].create_arequest()
+            req = self._pending_lazy_operations[i].create_request()
             if req is None:
                 self._pending_lazy_operations.pop(i)
                 i -= 1
@@ -1757,6 +1757,13 @@ def _throw_document_already_deleted_in_session(document_id: str, time_series: st
         )
 
     def append_single(self, timestamp: datetime, value: float, tag: Optional[str] = None) -> None:
+        if isinstance(value, int):
+            value = float(value)
+        if not isinstance(value, float):
+            raise TypeError(
+                f"Value passed ('{value}') is not a '{float.__name__}'. " f"It is '{value.__class__.__name__}'."
+            )
+
         self.append(timestamp, [value], tag)
 
     def append(self, timestamp: datetime, values: List[float], tag: Optional[str] = None) -> None:
@@ -2267,7 +2274,7 @@ def append(self, timestamp: datetime, entry: _T_TS_Values_Bindable, tag: Optiona
         super().append(timestamp, values, tag)
 
     def append_entry(self, entry: TypedTimeSeriesEntry[_T_TS_Values_Bindable]) -> None:
-        self.append_single(entry.timestamp, entry.value, entry.tag)
+        self.append(entry.timestamp, entry.value, entry.tag)
 
 
 class SessionDocumentRollupTypedTimeSeries(SessionTimeSeriesBase, Generic[_T_TS_Values_Bindable]):

ravendb/documents/store/definition.py

@@ -32,6 +32,7 @@
 )
 from ravendb.documents.session.misc import SessionOptions
 from ravendb.documents.subscriptions.document_subscriptions import DocumentSubscriptions
+from ravendb.documents.time_series import TimeSeriesOperations
 from ravendb.http.request_executor import RequestExecutor
 from ravendb.documents.identity.hilo import MultiDatabaseHiLoGenerator
 from ravendb.http.topology import Topology
@@ -321,6 +322,7 @@ def __init__(self, urls: Union[str, List[str]] = None, database: Optional[str] =
         self.__database_changes = {}
         self.__after_close: List[Callable[[], None]] = []
         self.__before_close: List[Callable[[], None]] = []
+        self.__time_series_operation: Optional[TimeSeriesOperations] = None
 
     def __enter__(self):
         return self
@@ -550,3 +552,10 @@ def operations(self) -> OperationExecutor:
             self.__operation_executor = OperationExecutor(self)
 
         return self.__operation_executor
+
+    @property
+    def time_series(self) -> TimeSeriesOperations:
+        if self.__time_series_operation is None:
+            self.__time_series_operation = TimeSeriesOperations(self)
+
+        return self.__time_series_operation

ravendb/http/request_executor.py

@@ -890,14 +890,14 @@ def __supply_async(
 
         return preferred_task.result().response
 
-    def __create_request(self, node: ServerNode, command: RavenCommand) -> requests.Request:
+    def __create_request(self, node: ServerNode, command: RavenCommand) -> Optional[requests.Request]:
         request = command.create_request(node)
         # todo: optimize that if - look for the way to make less ifs each time
         if request.data and not isinstance(request.data, str) and not inspect.isgenerator(request.data):
             request.data = json.dumps(request.data, default=self.conventions.json_default_method)
 
         # todo: 1117 - 1133
-        return request if request else None
+        return request or None
 
     def should_broadcast(self, command: RavenCommand) -> bool:
         if not isinstance(command, Broadcast):

ravendb/primitives/time_series.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 from enum import Enum
-from typing import List, Tuple
+from typing import List, Tuple, Dict, Any
 
 from ravendb.primitives.constants import int_max, int_min
 
@@ -22,6 +22,9 @@ def __init__(self, value: int, unit: TimeValueUnit):
         self.value = value
         self.unit = unit
 
+    def to_json(self) -> Dict[str, Any]:
+        return {"Value": self.value, "Unit": self.unit}
+
     def __str__(self):
         if self.value == int_max:
             return "MaxValue"

ravendb/misc.py …xing_tests/time_series_tests/__init__.pyravendb/misc.py renamed to ravendb/tests/jvm_migrated_tests/client_tests/indexing_tests/time_series_tests/__init__.py ravendb/misc.py renamed to ravendb/tests/jvm_migrated_tests/client_tests/indexing_tests/time_series_tests/__init__.py

@@ -0,0 +1,60 @@
+from datetime import datetime
+
+from ravendb import GetTermsOperation
+from ravendb.documents.indexes.index_creation import AbstractJavaScriptIndexCreationTask
+from ravendb.infrastructure.orders import Company
+from ravendb.tests.test_base import TestBase
+from ravendb.tools.raven_test_helper import RavenTestHelper
+
+
+class Companies_ByTimeSeriesNames(AbstractJavaScriptIndexCreationTask):
+    def __init__(self):
+        super(Companies_ByTimeSeriesNames, self).__init__()
+        self.maps = ["map('Companies', function (company) {return ({names: timeSeriesNamesFor(company)})})"]
+
+
+company_id = "companies/1"
+base_line = datetime(2023, 8, 20, 21, 30)
+
+
+class TestBasicTimeSeriesIndexesJavaScript(TestBase):
+    def setUp(self):
+        super(TestBasicTimeSeriesIndexesJavaScript, self).setUp()
+
+    def test_time_series_names_for(self):
+        now = RavenTestHelper.utc_today()
+        index = Companies_ByTimeSeriesNames()
+        index.execute(self.store)
+
+        with self.store.open_session() as session:
+            session.store(Company(), company_id)
+            session.save_changes()
+
+        self.wait_for_indexing(self.store)
+        RavenTestHelper.assert_no_index_errors(self.store)
+
+        terms = self.store.maintenance.send(GetTermsOperation(index.index_name, "names", None))
+        self.assertEqual(0, len(terms))
+
+        terms = self.store.maintenance.send(GetTermsOperation(index.index_name, "names_IsArray", None))
+        self.assertEqual(1, len(terms))
+        self.assertIn("true", terms)
+
+        with self.store.open_session() as session:
+            company = session.load(company_id, Company)
+            session.time_series_for_entity(company, "heartRate").append_single(now, 2.5, "tag1")
+            session.time_series_for_entity(company, "heartRate2").append_single(now, 3.5, "tag2")
+            session.save_changes()
+
+        self.wait_for_indexing(self.store)
+
+        RavenTestHelper.assert_no_index_errors(self.store)
+
+        terms = self.store.maintenance.send(GetTermsOperation(index.index_name, "names", None))
+        self.assertEqual(2, len(terms))
+        self.assertIn("heartrate", terms)
+        self.assertIn("heartrate2", terms)
+
+        terms = self.store.maintenance.send(GetTermsOperation(index.index_name, "names_IsArray", None))
+        self.assertEqual(1, len(terms))
+        self.assertIn("true", terms)