Add timeseries operations (Get, Batch, Append, Remove)

IdanHaim · IdanHaim · commit 5f9d1a16a247 · 2020-04-15T18:18:53.000+03:00
diff --git a/pyravendb/connection/requests_executor.py b/pyravendb/connection/requests_executor.py
@@ -41,7 +41,7 @@ def __init__(self, database_name, certificate, conventions=None, **kwargs):
         self._last_known_urls = None
 
         self.headers = {"Accept": "application/json",
-                        "Raven-Client-Version": "4.0.0.0"}
+                        "Raven-Client-Version": "5.0.0.0"}
 
         self.update_topology_lock = Lock()
         self.update_timer_lock = Lock()
diff --git a/pyravendb/raven_operations/timeseries_operations.py b/pyravendb/raven_operations/timeseries_operations.py
@@ -0,0 +1,161 @@
+from typing import Optional, Iterable, List, Generator
+from pyravendb.commands.raven_commands import *
+from pyravendb.tools.utils import Utils
+from .operations import Operation
+from datetime import datetime
+
+import sys
+
+
+class TimeSeriesRange:
+    def __init__(self, name, from_date: Optional[datetime] = None, to_date: Optional[datetime] = None):
+        if not name:
+            raise ValueError("TimeSeriesRange must include name")
+        self.name = name
+        self.from_date = from_date
+        self.to_date = to_date
+
+
+class GetTimeSeriesOperation(Operation):
+    def __init__(self, document_id: str, ranges: Iterable[TimeSeriesRange] or TimeSeriesRange = None,
+                 start: int = 0, page_size: int = sys.maxsize):
+        """
+        Build the time series get operations construct with ranges or from_date and to_date.
+        """
+        super().__init__()
+        if isinstance(ranges, TimeSeriesRange):
+            ranges = [ranges]
+
+        self._ranges = ranges
+        self._document_id = document_id
+        self._start = start
+        self._page_size = page_size
+
+    def get_command(self, store, conventions, cache=None):
+        return self._GetTimeSeriesCommand(self._document_id, self._ranges, self._start, self._page_size)
+
+    class _GetTimeSeriesCommand(RavenCommand):
+        def __init__(self, document_id: str, ranges: Iterable[TimeSeriesRange],
+                     start: int = 0, page_size: int = sys.maxsize):
+            super().__init__(method="GET", is_read_request=True)
+
+            if not document_id:
+                raise ValueError("Invalid document_id")
+
+            self._document_id = document_id
+            self._ranges = ranges
+            self._start = start
+            self._page_size = page_size
+
+        def create_request(self, server_node):
+            self.url = (f"{server_node.url}/databases/{server_node.database}"
+                        f"/timeseries?id={Utils.quote_key(self._document_id)}"
+                        f"{f'&start={self._start}' if self._start > 0 else ''}"
+                        f"{f'&pageSize={self._page_size}' if self._page_size < sys.maxsize else ''}")
+
+            if self._ranges is not None:
+                for range_ in self._ranges:
+                    self.url += (f"&name={Utils.quote_key(range_.name)}"
+                                 f"&from={Utils.datetime_to_string(range_.from_date)}"
+                                 f"&to={Utils.datetime_to_string(range_.to_date)}")
+
+        def set_response(self, response):
+            if response is None:
+                return None
+            try:
+                response = response.json()
+                if "Error" in response:
+                    raise exceptions.ErrorResponseException(response["Error"])
+            except ValueError as e:
+                raise exceptions.ErrorResponseException(e)
+
+            return response
+
+
+class TimeSeriesOperation:
+
+    def __init__(self, name: str, appends: Optional[List["RemoveOperation"]] = None,
+                 removals: [List["RemoveOperation"]] = None):
+        self.name = name
+        self.appends = [] if not appends else Utils.sort_iterable(appends, key=lambda ao: ao.timestamp.timestamp())
+        self.removals = [] if removals is None else removals
+
+    def append(self, timestamp: datetime, values: List[float] or float, tag: Optional[str] = None):
+        self.appends.append(TimeSeriesOperation.AppendOperation(timestamp, values, tag))
+
+    def remove(self, from_date: datetime, to_date: datetime):
+        self.removals.append(TimeSeriesOperation.RemoveOperation(from_date, to_date))
+
+    def to_json(self):
+        if self.appends and isinstance(self.appends, Generator):
+            self.appends = next(self.appends)
+        return {"Name": self.name, "Appends": [a.to_json() for a in self.appends] if self.appends else self.appends,
+                "Removals": [r.to_json() for r in self.removals] if self.removals else self.removals}
+
+    class AppendOperation:
+        def __init__(self, timestamp: datetime, values: List[float] or float, tag: Optional[str] = None):
+            if not timestamp:
+                raise ValueError("Missing timestamp property")
+            self.timestamp = timestamp
+            if values is None:
+                raise ValueError("Missing values property")
+            try:
+                if not isinstance(values, list):
+                    values = [float(values)]
+                else:
+                    values = [float(value) for value in values]
+            except ValueError:
+                raise (f"One or more values from values are not numbers %s" % values)
+            self.values = values
+            self.tag = tag
+
+        def to_json(self):
+            _dict = {"Timestamp": Utils.datetime_to_string(self.timestamp), "Values": self.values}
+            if self.tag:
+                _dict["Tag"] = self.tag
+            return _dict
+
+    class RemoveOperation:
+        def __init__(self, from_date: datetime, to_date: datetime):
+            if not from_date:
+                raise ValueError("Missing from_date property")
+            if not to_date:
+                raise ValueError("Missing to_date property")
+
+            self.from_date = from_date
+            self.to_date = to_date
+
+        def to_json(self):
+            return {"From": self.from_date, "To": self.to_date}
+
+
+class TimeSeriesBatchOperation(Operation):
+    def __init__(self, document_id: str, operation: TimeSeriesOperation):
+
+        super().__init__()
+        self._document_id = document_id
+        self.__operation = operation
+
+    def get_command(self, store, conventions, cache=None):
+        return self._TimeSeriesBatchCommand(self._document_id, self.__operation)
+
+    class _TimeSeriesBatchCommand(RavenCommand):
+        def __init__(self, document_id: str, operation: TimeSeriesOperation):
+            super().__init__(method="POST")
+
+            if not document_id:
+                raise ValueError("Invalid document_id")
+
+            self._document_id = document_id
+            self._operation = operation
+
+        def create_request(self, server_node):
+            self.url = (f"{server_node.url}/databases/{server_node.database}"
+                        f"/timeseries?id={Utils.quote_key(self._document_id)}")
+
+            self.data = self._operation.to_json()
+
+        def set_response(self, response):
+            if response and response.status_code != 204:
+                response = response.json()
+                raise exceptions.ErrorResponseException(response["Error"])
diff --git a/pyravendb/store/document_session.py b/pyravendb/store/document_session.py
@@ -6,6 +6,7 @@
 from pyravendb.commands.commands_data import PutAttachmentCommandData, DeleteAttachmentCommandData, DeleteCommandData
 from pyravendb.tools.utils import Utils
 from pyravendb.data.operation import AttachmentType
+from .session_timeseries import TimeSeries
 from copy import deepcopy
 
 
@@ -32,6 +33,7 @@ def __init__(self, database, document_store, requests_executor, session_id, **kw
         self._included_documents_by_id = {}
         self._deleted_entities = set()
         self._documents_by_entity = {}
+        self._time_series_by_document_id = {}
         self._known_missing_ids = set()
         self.id_value = None
         self._defer_commands = set()
@@ -453,6 +455,14 @@ def clear(self):
         self.session.included_documents_by_id.clear()
         self.session.known_missing_ids.clear()
 
+    # TODO: complete time_series in the session
+    def time_series_for(self, entity_or_document_id, name):
+        """
+        Get A time series object associated with the document
+        """
+        raise NotImplemented("Time-series from the session is not implemented")
+        # return TimeSeries(self, entity_or_document_id, name)
+
 
 class _Attachment:
     def __init__(self, session):
@@ -471,7 +481,7 @@ def _command_exists(self, document_id):
     def throw_not_in_session(self, entity):
         raise ValueError(
             repr(entity) + " is not associated with the session, cannot add attachment to it. "
-                           "Use documentId instead or track the entity in the session.")
+                           "Use document Id instead or track the entity in the session.")
 
     def store(self, entity_or_document_id, name, stream, content_type=None, change_vector=None):
         if not isinstance(entity_or_document_id, str):
diff --git a/pyravendb/store/session_timeseries.py b/pyravendb/store/session_timeseries.py
@@ -0,0 +1,27 @@
+from datetime import datetime
+import sys
+
+
+class TimeSeries:
+
+    @staticmethod
+    def throw_not_in_session(entity):
+        raise ValueError(
+            repr(entity) + " is not associated with the session, cannot add time-series to it. "
+                           "Use document Id instead or track the entity in the session.")
+
+    def __init__(self, session, entity_or_document_id, name):
+        self._session = session
+        if not isinstance(entity_or_document_id, str):
+            entity = self._session.documents_by_entity.get(entity_or_document_id, None)
+            if not entity:
+                self.throw_not_in_session(entity_or_document_id)
+            entity_or_document_id = entity["metadata"]["@id"]
+
+        if not entity_or_document_id:
+            raise ValueError(entity_or_document_id)
+        if not name:
+            raise ValueError(name)
+
+    def get(self, from_date: datetime, to_date: datetime, start: int = 0, page_size: int = sys.maxsize):
+        raise NotImplemented("Time-series from the session is not implemented")
diff --git a/pyravendb/tools/utils.py b/pyravendb/tools/utils.py
@@ -199,7 +199,7 @@ def dict_to_string(dictionary):
 
     @staticmethod
     def datetime_to_string(datetime_obj):
-        return datetime_obj.strftime("%Y-%m-%dT%H:%M:%S.%f0")
+        return datetime_obj.strftime("%Y-%m-%dT%H:%M:%S.%f0") if datetime_obj else ''
 
     @staticmethod
     def start_a_timer(interval, function, args=None, name=None, daemon=False):
@@ -359,3 +359,14 @@ def contains_any(text, any_of):
             if c in text:
                 return True
         return False
+
+    @staticmethod
+    def sort_iterable(iterable, key=None):
+        """
+        This function will take an iterable and try to sort it by the key
+        if the key is not given will use the sorted default one.
+        return a generator
+        """
+        if not iterable:
+            return iterable
+        yield sorted(iterable, key=key)
diff --git a/pyravendb/tryouts/program.py b/pyravendb/tryouts/program.py
@@ -1,4 +1,31 @@
 from pyravendb.store.document_store import DocumentStore
+from pyravendb.raven_operations.timeseries_operations import \
+    GetTimeSeriesOperation, TimeSeriesRange, TimeSeriesBatchOperation, TimeSeriesOperation
+from datetime import datetime, timedelta
 
 if __name__ == "__main__":
-    pass
+    with DocumentStore(urls=["http://localhost:8080"], database="NorthWind") as store:
+        store.initialize()
+        # Add time_series to user/1-A document
+        time_series_operation = TimeSeriesOperation(name="Heartrate")
+        time_series_operation.append(datetime.now(), 73, tag="heart/rates")
+        time_series_operation.append(datetime.now() + timedelta(minutes=5), 78, tag="heart/rates")
+
+        time_series_batch_operation = TimeSeriesBatchOperation(document_id="user/1-A", operation=time_series_operation)
+        store.operations.send(time_series_batch_operation)
+
+        # Fetch all time_series from the document
+        tsr = TimeSeriesRange("Heartrate", datetime.now() - timedelta(days=2), datetime.now() + timedelta(days=2))
+        time_series = store.operations.send(GetTimeSeriesOperation("user/1-A", ranges=tsr))
+        print(len(time_series['Values']["Heartrate"][0]['Entries']))
+
+        # Remove all time_series from the document
+        time_series_operation = TimeSeriesOperation(name="Heartrate")
+        time_series_operation.remove(datetime.now() - timedelta(days=2), datetime.now() + timedelta(days=2))
+
+        time_series_batch_operation = TimeSeriesBatchOperation(document_id="user/1-A", operation=time_series_operation)
+        store.operations.send(time_series_batch_operation)
+
+        tsr = TimeSeriesRange("Heartrate", datetime.now() - timedelta(days=2), datetime.now() + timedelta(days=2))
+        time_series = store.operations.send(GetTimeSeriesOperation("user/1-A", ranges=tsr))
+        print(len(time_series['Values']["Heartrate"][0]['Entries']))
diff --git a/setup.py b/setup.py
@@ -3,14 +3,15 @@
 setup(
     name='pyravendb',
     packages=find_packages(),
-    version='4.0.6.2',
+    version='5.0.0.0',
     long_description=open("README.rst").read(),
-    description='This is the official python client for RavenDB v4.0 document database',
+    description='This is the official python client for RavenDB v5.0 document database',
     author='RavenDB',
     author_email='support@ravendb.net',
     url='https://github.com/ravendb/RavenDB-Python-Client',
     license='MIT',
     keywords='pyravendb',
+    python_requires='~=3.5',
     install_requires=
     [
         'requests >= 2.18.4',
