RDBC-644 Bulk Insert Operation miscellaneous

Author: poissoncorp

ravendb/documents/commands/bulkinsert.py

@@ -0,0 +1,34 @@
+import json
+from typing import Optional
+
+import requests
+
+from ravendb.http.server_node import ServerNode
+from ravendb.http.raven_command import RavenCommand, VoidRavenCommand
+
+
+class GetNextOperationIdCommand(RavenCommand[int]):
+    def __init__(self):
+        super(GetNextOperationIdCommand, self).__init__(int)
+        self._node_tag = 0
+
+    def is_read_request(self) -> bool:
+        return False  # disable caching
+
+    def create_request(self, node: ServerNode) -> requests.Request:
+        return requests.Request("GET", f"{node.url}/databases/{node.database}/operations/next-operation-id")
+
+    def set_response(self, response: Optional[str], from_cache: bool) -> None:
+        json_node = json.loads(response)
+        self.result = json_node.get("Id", None)
+        self._node_tag = json_node.get("NodeTag", None)
+
+
+class KillOperationCommand(VoidRavenCommand):
+    def __init__(self, operation_id: int, node_tag: Optional[str] = None):
+        super(KillOperationCommand, self).__init__()
+        self._id = operation_id
+        self._selected_node_tag = node_tag
+
+    def create_request(self, node: ServerNode) -> requests.Request:
+        return requests.Request("POST", f"{node.url}/databases/{node.database}/operations/kill?id={self._id}")

ravendb/documents/conventions.py

@@ -219,7 +219,10 @@ def json_default(o):
         elif isinstance(o, (int, float)):
             return str(o)
         else:
-            raise TypeError(repr(o) + " is not JSON serializable (Try add a json default method to convention)")
+            raise TypeError(
+                repr(o) + " is not JSON serializable (Try add a json default method to convention"
+                " or try to add methods - to_json & classmethod from_json - to object class)"
+            )
 
     @staticmethod
     def default_transform_plural(name):

ravendb/documents/store/definition.py

@@ -8,6 +8,7 @@
 
 from ravendb import constants, exceptions
 from ravendb.changes.database_changes import DatabaseChanges
+from ravendb.documents.bulk_insert_operation import BulkInsertOperation
 from ravendb.documents.operations.executor import MaintenanceOperationExecutor, OperationExecutor
 from ravendb.documents.operations.indexes import PutIndexesOperation
 from ravendb.documents.session.event_args import (
@@ -163,14 +164,10 @@ def maintenance(self) -> MaintenanceOperationExecutor:
     def operations(self) -> OperationExecutor:
         pass
 
-    # todo: changes
-
     # todo: aggressive_caching
 
     # todo: time_series
 
-    # todo: bulk_insert
-
     @abstractmethod
     def open_session(self, database: Optional[str] = None, session_options: Optional = None):
         pass
@@ -313,7 +310,6 @@ def __init__(self, urls: Optional[Union[str, List[str]]] = None, database: Optio
         self.urls = [urls] if isinstance(urls, str) else urls
         self.database = database
         self.__request_executors: Dict[str, Lazy[RequestExecutor]] = CaseInsensitiveDict()
-        # todo: database changes
         # todo: aggressive cache
         self.__maintenance_operation_executor: Union[None, MaintenanceOperationExecutor] = None
         self.__operation_executor: Union[None, OperationExecutor] = None
@@ -519,6 +515,10 @@ def initialize(self) -> DocumentStore:
 
     # todo: aggressively cache
 
+    def bulk_insert(self, database_name: Optional[str] = None) -> BulkInsertOperation:
+        self.assert_initialized()
+        return BulkInsertOperation(self.get_effective_database(database_name), self)
+
     def _assert_valid_configuration(self) -> None:
         if not self.urls:
             raise ValueError("Document URLs cannot be empty.")

ravendb/exceptions/documents/bulkinsert.py

@@ -0,0 +1,13 @@
+from typing import Optional
+
+from ravendb.exceptions.raven_exceptions import RavenException
+
+
+class BulkInsertAbortedException(RavenException):
+    def __init__(self, message: str, cause: Optional[Exception] = None):
+        super(BulkInsertAbortedException, self).__init__(message, cause)
+
+
+class BulkInsertProtocolViolationException(RavenException):
+    def __init__(self, message: str, cause: Optional[Exception] = None):
+        super(BulkInsertProtocolViolationException, self).__init__(message, cause)

ravendb/http/request_executor.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import datetime
+import inspect
 import json
 import logging
 import os
@@ -579,33 +580,21 @@ def __send_request_to_server(
         session_info: SessionInfo,
         request: requests.Request,
         url: str,
-    ) -> requests.Response:
+    ) -> Optional[requests.Response]:
         try:
             self.number_of_server_requests += 1
             timeout = command.timeout if command.timeout else self.__default_timeout
-            if timeout:
+
+            if not timeout:
+                return self.__send(chosen_node, command, session_info, request)
+
+            else:
                 try:
-                    # todo: create Task from lines below and call it
-                    # AggressiveCacheOptions callingTheadAggressiveCaching = aggressiveCaching.get();
-                    # CompletableFuture<CloseableHttpResponse> sendTask = CompletableFuture.supplyAsync(() ->
-                    # AggressiveCacheOptions aggressiveCacheOptionsToRestore = aggressiveCaching.get();
-                    try:
-                        return self.__send(chosen_node, command, session_info, request)
-                    except IOError:
-                        # throw ExceptionsUtils.unwrapException(e);
-                        raise
-                    # finally aggressiveCaching.set(aggressiveCacheOptionsToRestore);
+                    return self.__send(chosen_node, command, session_info, request)
                 except requests.Timeout as t:
-                    # request.abort()
-                    # net.ravendb.client.exceptions.TimeoutException timeoutException =
-                    # new net.ravendb.client.exceptions.TimeoutException(
-                    # "The request for " + request.getURI() + " failed with timeout after " +
-                    # TimeUtils.durationToTimeSpan(timeout), e);
-
                     if not should_retry:
                         if command.failed_nodes is None:
                             command.failed_nodes = {}
-
                         command.failed_nodes[chosen_node] = t
                         raise t
 
@@ -615,10 +604,6 @@ def __send_request_to_server(
                         self.__throw_failed_to_contact_all_nodes(command, request)
 
                     return None
-                except IOError as e:
-                    raise e
-            else:
-                return self.__send(chosen_node, command, session_info, request)
         except IOError as e:
             if not should_retry:
                 raise
@@ -633,7 +618,7 @@ def __send_request_to_server(
     def __send(
         self, chosen_node: ServerNode, command: RavenCommand, session_info: SessionInfo, request: requests.Request
     ) -> requests.Response:
-        response: requests.Response = None
+        response: Optional[requests.Response] = None
 
         if self.should_execute_on_all(chosen_node, command):
             response = self.__execute_on_all_to_figure_out_the_fastest(chosen_node, command)
@@ -891,7 +876,8 @@ def __supply_async(
 
     def __create_request(self, node: ServerNode, command: RavenCommand) -> requests.Request:
         request = command.create_request(node)
-        if request.data and not isinstance(request.data, str):
+        # 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
@@ -1125,10 +1111,10 @@ def __handle_server_down(
         self,
         url: str,
         chosen_node: ServerNode,
-        node_index: int,
+        node_index: Optional[int],
         command: RavenCommand,
         request: requests.Request,
-        response: requests.Response,
+        response: Optional[requests.Response],
         e: Exception,
         session_info: SessionInfo,
         should_retry: bool,