Merge pull request #181 from poissoncorp/RDBC-700

RDBC-700 Synchronize v5.2 Python to v5.2 Java

Author: poissoncorp

ravendb/__init__.py

@@ -177,7 +177,9 @@
 from ravendb.documents.session.document_info import DocumentInfo
 from ravendb.documents.session.document_session import DocumentSession
 from ravendb.documents.session.entity_to_json import EntityToJson
-from ravendb.documents.session.in_memory_document_session_operations import InMemoryDocumentSessionOperations
+from ravendb.documents.session.document_session_operations.in_memory_document_session_operations import (
+    InMemoryDocumentSessionOperations,
+)
 from ravendb.documents.session.loaders.include import IncludeBuilder, IncludeBuilderBase, QueryIncludeBuilder
 from ravendb.documents.session.loaders.loaders import (
     LoaderWithInclude,

ravendb/constants.py

@@ -16,6 +16,7 @@ class Metadata:
         LAST_MODIFIED = "@last-modified"
         CHANGE_VECTOR = "@change-vector"
         EXPIRES = "@expires"
+        REFRESH = "@refresh"
         ALL_DOCUMENTS_COLLECTION = "@all_docs"
         EMPTY_COLLECTION = "@empty"
         NESTED_OBJECT_TYPES = "@nested-object-types"

ravendb/documents/bulk_insert_operation.py

@@ -62,17 +62,25 @@ def error_on_request_start(self, exception: Exception):
             self.output_stream_mock.set_exception(exception)
 
     class _BulkInsertCommand(RavenCommand[requests.Response]):
-        def __init__(self, key: int, buffer_exposer: BulkInsertOperation._BufferExposer, node_tag: str):
+        def __init__(
+            self,
+            key: int,
+            buffer_exposer: BulkInsertOperation._BufferExposer,
+            node_tag: str,
+            skip_overwrite_if_unchanged: bool,
+        ):
             super().__init__(requests.Response)
             self._buffer_exposer = buffer_exposer
             self._key = key
             self._selected_node_tag = node_tag
             self.use_compression = False
+            self._skip_overwrite_if_unchanged = skip_overwrite_if_unchanged
 
         def create_request(self, node: ServerNode) -> requests.Request:
             return requests.Request(
                 "POST",
-                f"{node.url}/databases/{node.database}/bulk_insert?id={self._key}",
+                f"{node.url}/databases/{node.database}/bulk_insert?id={self._key}"
+                f"&skipOverwriteIfUnchanged={'true' if self._skip_overwrite_if_unchanged else 'false'}",
                 data=self._buffer_exposer.send_data(),
             )
 
@@ -88,7 +96,7 @@ def send(self, session: requests.Session, request: requests.Request) -> requests
             except Exception as e:
                 self._buffer_exposer.error_on_request_start(e)
 
-    def __init__(self, database: str = None, store: "DocumentStore" = None):
+    def __init__(self, database: str = None, store: "DocumentStore" = None, options: BulkInsertOptions = None):
         self.use_compression = False
 
         self._ongoing_bulk_insert_execute_task: Optional[Future] = None
@@ -103,6 +111,9 @@ def __init__(self, database: str = None, store: "DocumentStore" = None):
         self._conventions = store.conventions
         if not database or database.isspace():
             self._throw_no_database()
+
+        self._use_compression = options.use_compression if options else False
+        self._options = options or BulkInsertOptions()
         self._request_executor = store.get_request_executor(database)
 
         self._enqueue_current_buffer_async = Future()
@@ -232,6 +243,7 @@ def store_as(
 
                 self._write_document(entity, metadata)
                 self._write_string_no_escape("}")
+                # todo: self._flush_if_needed() - causes error - https://issues.hibernatingrhinos.com/issue/RDBC-701
             except Exception as e:
                 self._handle_errors(key, e)
         finally:
@@ -336,7 +348,7 @@ def _get_exception_from_operation(self) -> Optional[BulkInsertAbortedException]:
     def _start_executing_bulk_insert_command(self) -> None:
         try:
             bulk_command = BulkInsertOperation._BulkInsertCommand(
-                self._operation_id, self._buffer_exposer, self._node_tag
+                self._operation_id, self._buffer_exposer, self._node_tag, self._options.skip_overwrite_if_unchanged
             )
             bulk_command.use_compression = self.use_compression
 
@@ -441,3 +453,9 @@ def attachments_for(self, key: str) -> BulkInsertOperation.AttachmentsBulkInsert
             raise ValueError("Document id cannot be None or empty.")
 
         return BulkInsertOperation.AttachmentsBulkInsert(self, key)
+
+
+class BulkInsertOptions:
+    def __init__(self, use_compression: bool = None, skip_overwrite_if_unchanged: bool = None):
+        self.use_compression = use_compression
+        self.skip_overwrite_if_unchanged = skip_overwrite_if_unchanged

ravendb/documents/commands/batches.py

@@ -23,7 +23,9 @@
 if TYPE_CHECKING:
     from ravendb.documents.conventions import DocumentConventions
     from ravendb.documents.operations.patch import PatchRequest
-    from ravendb.documents.session.in_memory_document_session_operations import InMemoryDocumentSessionOperations
+    from ravendb.documents.session.document_session_operations.in_memory_document_session_operations import (
+        InMemoryDocumentSessionOperations,
+    )
 
 
 class CommandType(Enum):
@@ -270,11 +272,15 @@ def serialize(self, conventions: DocumentConventions) -> dict:
 
 
 class DeleteCommandData(CommandData):
-    def __init__(self, key: str, change_vector: str):
-        super(DeleteCommandData, self).__init__(key=key, command_type=CommandType.DELETE)
+    def __init__(self, key: str, change_vector: str, original_change_vector: str = None):
+        super(DeleteCommandData, self).__init__(key=key, command_type=CommandType.DELETE, change_vector=change_vector)
+        self.original_change_vector = original_change_vector
 
     def serialize(self, conventions: DocumentConventions) -> dict:
-        return {"Id": self.key, "ChangeVector": self.change_vector, "Type": CommandType.DELETE}
+        data = {"Id": self.key, "ChangeVector": self.change_vector, "Type": CommandType.DELETE}
+        if self.original_change_vector is not None:
+            data.update({"OriginalChangeVector": self.original_change_vector})
+        return data
 
 
 class PutCommandDataBase(CommandData):

ravendb/documents/commands/crud.py

@@ -90,7 +90,7 @@ def create_request(self, node: ServerNode) -> requests.Request:
         url = f"{node.url}/databases/{node.database}/docs?id={Utils.escape(self.__key, True, False)}"
         request = requests.Request("HEAD", url)
         if self.__change_vector is not None:
-            request.headers["If-None-Match"] = self.__change_vector
+            request.headers[constants.Headers.IF_NONE_MATCH] = self.__change_vector
         return request
 
     def process_response(self, cache: HttpCache, response: requests.Response, url) -> ResponseDisposeHandling:
@@ -387,7 +387,7 @@ def create_request(self, node: ServerNode) -> requests.Request:
             f"{node.url}/databases/{node.database}"
             f"/attachments?id={Utils.quote_key(self.__document_id)}"
             f"&name={Utils.quote_key(self.__name)}",
-            {"If-None-Match": self.__change_vector} if self.__change_vector else None,
+            {constants.Headers.IF_NONE_MATCH: self.__change_vector} if self.__change_vector else None,
         )
 
     def process_response(self, cache: HttpCache, response: requests.Response, url) -> http.ResponseDisposeHandling:
@@ -435,7 +435,7 @@ def __init__(self, key: str, change_vector: str):
     def create_request(self, node: ServerNode) -> requests.Request:
         url = f"{node.url}/databases/{node.database}/docs?id={Utils.quote_key(self.__key)}"
         request = requests.Request("GET", url)
-        request.headers["If-None-Match"] = f'"{self.__change_vector}"'
+        request.headers[constants.Headers.IF_NONE_MATCH] = f'"{self.__change_vector}"'
 
         return request
 

ravendb/documents/commands/multi_get.py

@@ -115,36 +115,6 @@ def __init__(self, request_executor: RequestExecutor, commands: List[GetRequest]
     def create_request(self, node: ServerNode) -> Optional[requests.Request]:
         self.__base_url = f"{node.url}/databases/{node.database}"
         url = self.__base_url + "/multi_get"
-        # todo: aggressive caching
-        # if self.__maybe_read_all_from_cache(self.__request_executor.aggressive_caching):
-        #     self.aggressively_cached = True
-        #     return None
-        #
-        # aggressive_cache_options: AggressiveCacheOptions = self.__request_executor.aggressive_caching
-        # if aggressive_cache_options and aggressive_cache_options.mode == AggressiveCacheMode.TRACK_CHANGES:
-        #     self.result = []
-        #     for command in self.__commands:
-        #         if not command.can_cache_aggressively:
-        #             break
-        #         cache_key = self.__get_cache_key(command)[0]
-        #         cached_item, _, cached_ref = self.__http_cache.get(cache_key, "", "")
-        #         cached_item: ReleaseCacheItem
-        #         if (
-        #             cached_ref is None
-        #             or cached_item.age > aggressive_cache_options.duration
-        #             or cached_item.might_have_been_modified
-        #         ):
-        #             break
-        #         get_response = GetResponse()
-        #         get_response.result = cached_ref
-        #         get_response.status_code = http.HTTPStatus.NOT_MODIFIED
-        #         self.result.append(get_response)
-        #
-        #     if len(self.result) == len(self.__commands):
-        #         return None
-        #
-        #     self.result = None
-
         request = requests.Request("POST", url)
 
         request.data = {
@@ -211,7 +181,6 @@ def __get_cache_key(self, command: GetRequest) -> (str, str):
         req_url = self.__base_url + command.url_and_query
         return command.method + "-" + req_url if command.method else req_url, req_url
 
-    # todo: make sure json parses correctly down there
     def set_response_raw(self, response: requests.Response, stream: bytes) -> None:
         try:
             try:
@@ -224,7 +193,7 @@ def set_response_raw(self, response: requests.Response, stream: bytes) -> None:
 
                 for get_response in self.read_responses(response_temp):
                     command = self.__commands[i]
-                    self.__maybe_set_cache(get_response, command)
+                    self.__maybe_set_cache(get_response, command, i)
 
                     if self.__cached is not None and get_response.status_code == http.HTTPStatus.NOT_MODIFIED:
                         cloned_response = GetResponse()
@@ -241,8 +210,11 @@ def set_response_raw(self, response: requests.Response, stream: bytes) -> None:
         except Exception as e:
             self._throw_invalid_response(e)
 
-    def __maybe_set_cache(self, get_response: GetResponse, command: GetRequest):
+    def __maybe_set_cache(self, get_response: GetResponse, command: GetRequest, cached_index: int):
         if get_response.status_code == http.HTTPStatus.NOT_MODIFIED:
+            # if not modified - update age
+            if self.__cached is not None:
+                self.__cached.values[cached_index][0].not_modified()
             return
 
         cache_key = self.__get_cache_key(command)[0]
@@ -281,7 +253,19 @@ def is_read_request(self) -> bool:
         return False
 
     def close_cache(self):
+        # If _cached is not null - it means that the client approached with this multitask request to node
+        # and the request failed and now client tries to send it to another node.
+
         if self.__cached is not None:
             self.__cached.close()
 
         self.__cached = None
+        # The client sends the commands.
+        # Some of which could be saved in cache with a response
+        # that includes the change vector that received from the old fallen node.
+        # The client can't use those responses because their URLs are different
+        # (include the IP and port of the old node), because of that the client
+        # needs to get those docs again from the new node.
+
+        for command in self.__commands:
+            command.headers.remove(constants.Headers.IF_NONE_MATCH)

ravendb/documents/commands/subscriptions.py

@@ -63,12 +63,14 @@ def __init__(
         certificate: Optional[str] = None,
         urls: Optional[List[str]] = None,
         node_tag: Optional[str] = None,
+        server_id: Optional[str] = None,
     ):
         self.port = port
         self.url = url
         self.certificate = certificate
         self.urls = urls
         self.node_tag = node_tag
+        self.server_id = server_id
 
     @classmethod
     def from_json(cls, json_dict: Dict) -> TcpConnectionInfo:
@@ -78,6 +80,7 @@ def from_json(cls, json_dict: Dict) -> TcpConnectionInfo:
             json_dict.get("Certificate", None),
             json_dict.get("Urls", None),
             json_dict.get("NodeTag", None),
+            json_dict.get("ServerId", None),
         )
 
 

ravendb/documents/conventions.py

@@ -40,6 +40,7 @@ def __init__(self):
         self.throw_if_query_page_size_is_not_set = False
         self._send_application_identifier = True
         self._save_enums_as_integers: Optional[bool] = None
+        self._disable_atomic_document_writes_in_cluster_wide_transaction: Optional[bool] = None
 
         # Configuration
         self.json_default_method = DocumentConventions.json_default
@@ -198,6 +199,15 @@ def read_balance_behavior(self, value: ReadBalanceBehavior):
     def send_application_identifier(self) -> bool:
         return self._send_application_identifier
 
+    @property
+    def disable_atomic_document_writes_in_cluster_wide_transaction(self) -> bool:
+        return self._disable_atomic_document_writes_in_cluster_wide_transaction
+
+    @disable_atomic_document_writes_in_cluster_wide_transaction.setter
+    def disable_atomic_document_writes_in_cluster_wide_transaction(self, value: bool):
+        self.__assert_not_frozen()
+        self._disable_atomic_document_writes_in_cluster_wide_transaction = value
+
     @staticmethod
     def json_default(o):
         if o is None: