RDBC-700 MultiGetCommand, IndexDefinitionBase, Conventions, Headers

Author: poissoncorp

ravendb/documents/commands/batches.py

@@ -270,11 +270,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/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:

ravendb/documents/indexes/definitions.py

@@ -163,26 +163,49 @@ def to_json(self):
         }
 
 
-class IndexDefinition:
-    def __init__(self, **kwargs):
-        self.name: Union[None, str] = None
-        self.priority: Union[None, IndexPriority] = None
-        self.state: Union[None, IndexState] = None
-        self.lock_mode: Union[None, IndexLockMode] = None
-        self.additional_sources: Dict[str, str] = {}
-        self.additional_assemblies: Set[AdditionalAssembly] = set()
-        self.__maps: Union[None, Set[str]] = None
-        self.fields: Union[None, Dict[str, IndexFieldOptions]] = {}
-        self.reduce: Union[None, str] = None
-        self.configuration: Dict[str, str] = {}
-        self.__index_source_type: Union[None, IndexSourceType] = None
-        self.__index_type: Union[None, IndexType] = None
-        self.output_reduce_to_collection: Union[None, str] = None
-        self.reduce_output_index: Union[None, int] = None
-        self.pattern_for_output_reduce_to_collection_references: Union[None, str] = None
-        self.pattern_references_collection_name: Union[None, str] = None
-        self.deployment_mode: Union[None, IndexDeploymentMode] = None
-        self.__dict__.update(kwargs)
+class IndexDefinitionBase:
+    def __init__(self, name: str = None, priority: IndexPriority = None, state: IndexState = None):
+        self.name = name
+        self.priority = priority
+        self.state = state
+
+
+class IndexDefinition(IndexDefinitionBase):
+    def __init__(
+        self,
+        name: Optional[str] = None,
+        priority: Optional[str] = None,
+        state: Optional[str] = None,
+        lock_mode: Optional[IndexLockMode] = None,
+        additional_sources: Optional[Dict[str, str]] = None,
+        additional_assemblies: Optional[Set[AdditionalAssembly]] = None,
+        maps: Optional[Set[str]] = None,
+        fields: Optional[Dict[str, IndexFieldOptions]] = None,
+        reduce: Optional[str] = None,
+        configuration: Optional[Dict[str, str]] = None,
+        index_source_type: Optional[IndexSourceType] = None,
+        index_type: Optional[IndexType] = None,
+        output_reduce_to_collection: Optional[str] = None,
+        reduce_output_index: Optional[int] = None,
+        pattern_for_output_reduce_to_collection_references: Optional[str] = None,
+        pattern_references_collection_name: Optional[str] = None,
+        deployment_mode: Optional[IndexDeploymentMode] = None,
+    ):
+        super(IndexDefinition, self).__init__(name, priority, state)
+        self.lock_mode = lock_mode
+        self.additional_sources = additional_sources
+        self.additional_assemblies = additional_assemblies
+        self.__maps = maps or set()
+        self.fields = fields or {}
+        self.reduce = reduce
+        self.configuration = configuration or {}
+        self.__index_source_type = index_source_type
+        self.__index_type = index_type
+        self.output_reduce_to_collection = output_reduce_to_collection
+        self.reduce_output_index = reduce_output_index
+        self.pattern_for_output_reduce_to_collection_references = pattern_for_output_reduce_to_collection_references
+        self.pattern_references_collection_name = pattern_references_collection_name
+        self.deployment_mode = deployment_mode
 
     @classmethod
     def from_json(cls, json_dict: dict) -> IndexDefinition:
@@ -297,21 +320,19 @@ def detect_static_index_type(self) -> IndexType:
         return None  # todo: IndexDefinitionHelper.detect_Static_index_type(first_map, self.reduce)
 
 
-class AutoIndexDefinition:
+class AutoIndexDefinition(IndexDefinitionBase):
     def __init__(
         self,
-        type: Optional[IndexType] = None,
+        index_type: Optional[IndexType] = None,
         name: Optional[str] = None,
         priority: Optional[IndexPriority] = None,
         state: Optional[IndexState] = None,
         collection: Optional[str] = None,
         map_fields: Optional[Dict[str, AutoIndexFieldOptions]] = None,
         group_by_fields: Optional[Dict[str, AutoIndexFieldOptions]] = None,
     ):
-        self.type = type
-        self.name = name
-        self.priority = priority
-        self.state = state
+        super(AutoIndexDefinition, self).__init__(name, priority, state)
+        self.index_type = index_type
         self.collection = collection
         self.map_fields = map_fields
         self.group_by_fields = group_by_fields
@@ -330,7 +351,7 @@ def from_json(cls, json_dict: Dict) -> AutoIndexDefinition:
 
     def to_json(self) -> Dict:
         return {
-            "Type": self.type.value,
+            "Type": self.index_type.value,
             "Name": self.name,
             "Priority": self.priority.value,
             "State": self.state.value if self.state is not None else None,
@@ -451,9 +472,9 @@ def from_NuGet(
 
 class AbstractCommonApiForIndexes(ABC):
     def __init__(self):
-        self.__additional_sources: Union[None, Dict[str, str]] = None
-        self.__additional_assemblies: Union[None, Set[AdditionalAssembly]] = None
-        self.__configuration: Union[None, Dict[str, str]] = None
+        self.__additional_sources: Optional[Dict[str, str]] = None
+        self.__additional_assemblies: Optional[Set[AdditionalAssembly]] = None
+        self.__configuration: Optional[Dict[str, str]] = None
 
     @property
     def is_map_reduce(self) -> bool:

ravendb/documents/session/in_memory_document_session_operations.py

@@ -784,7 +784,7 @@ def delete(self, key_or_entity: Union[str, object], expected_change_vector: Opti
             if key is None:
                 raise ValueError("Id cannot be None")
             change_vector = None
-            document_info = self._documents_by_id.get(key)
+            document_info = self._documents_by_id.get_value(key)
             if document_info is not None:
                 new_obj = self.entity_to_json.convert_entity_to_json(document_info.entity, document_info)
                 if document_info.entity is not None and self._entity_changed(new_obj, document_info, None):
@@ -801,7 +801,13 @@ def delete(self, key_or_entity: Union[str, object], expected_change_vector: Opti
             change_vector = change_vector if self.__use_optimistic_concurrency else None
             if self._counters_by_doc_id:
                 self._counters_by_doc_id.pop(key, None)
-            self.defer(DeleteCommandData(key, expected_change_vector if expected_change_vector else change_vector))
+            self.defer(
+                DeleteCommandData(
+                    key,
+                    expected_change_vector or change_vector,
+                    expected_change_vector or (document_info.change_vector if document_info is not None else None),
+                )
+            )
             return
 
         entity = key_or_entity
@@ -1030,7 +1036,9 @@ def __prepare_for_entities_deletion(
 
                 change_vector = change_vector if self.__use_optimistic_concurrency else None
                 self.before_delete_invoke(BeforeDeleteEventArgs(self, document_info.key, document_info.entity))
-                result.session_commands.append(DeleteCommandData(document_info.key, document_info.change_vector))
+                result.session_commands.append(
+                    DeleteCommandData(document_info.key, change_vector, document_info.change_vector)
+                )
 
             if changes is None:
                 result.on_success.clear_deleted_entities()

ravendb/documents/store/definition.py

@@ -403,12 +403,13 @@ def close(self):
     def open_session(
         self, database: Optional[str] = None, session_options: Optional[SessionOptions] = None
     ) -> DocumentSession:
-        if not database and not session_options:
+        if not session_options:
             session_options = SessionOptions()
-        if not ((session_options is not None) ^ (database is not None)):
-            raise ValueError("Pass either database str or session_options object")
-        if database:
-            session_options = SessionOptions(database=database)
+            session_options.database = database
+            session_options.disable_atomic_document_writes_in_cluster_wide_transaction = (
+                self.conventions.disable_atomic_document_writes_in_cluster_wide_transaction
+            )
+
         self.assert_initialized()
         self._ensure_not_closed()
 

ravendb/tests/jvm_migrated_tests/server_tests/test_databases.py

@@ -31,7 +31,7 @@ def test_can_get_info_auto_index_info(self):
         auto_index_definition = record.auto_indexes.get("Auto/Genres/Byname")
         self.assertIsNotNone(auto_index_definition)
 
-        self.assertEqual(IndexType.AUTO_MAP, auto_index_definition.type)
+        self.assertEqual(IndexType.AUTO_MAP, auto_index_definition.index_type)
         self.assertEqual("Auto/Genres/Byname", auto_index_definition.name)
         self.assertEqual(IndexPriority.NORMAL, auto_index_definition.priority)
         self.assertEqual("Genres", auto_index_definition.collection)