RDBC-700 Bulk Insert Options

Author: poissoncorp

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/store/definition.py

@@ -8,7 +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.bulk_insert_operation import BulkInsertOperation, BulkInsertOptions
 from ravendb.documents.operations.executor import MaintenanceOperationExecutor, OperationExecutor
 from ravendb.documents.operations.indexes import PutIndexesOperation
 from ravendb.documents.session.event_args import (
@@ -48,12 +48,12 @@ def __init__(self):
         self.__conventions = None
         self._initialized: bool = False
 
-        self.__certificate_pem_path: Union[None, str] = None
-        self.__trust_store_path: Union[None, str] = None
+        self.__certificate_pem_path: Optional[str] = None
+        self.__trust_store_path: Optional[str] = None
 
         self._urls: List[str] = []
-        self._database: Union[None, str] = None
-        self._disposed: Union[None, bool] = None
+        self._database: Optional[str] = None
+        self._disposed: Optional[bool] = None
 
         self.__before_store: List[Callable[[BeforeStoreEventArgs], None]] = []
         self.__after_save_changes: List[Callable[[AfterSaveChangesEventArgs], None]] = []
@@ -311,10 +311,11 @@ def __init__(self, urls: Union[str, List[str]] = None, database: Optional[str] =
         self.database = database
         self.__request_executors: Dict[str, Lazy[RequestExecutor]] = CaseInsensitiveDict()
         # todo: aggressive cache
-        self.__maintenance_operation_executor: Union[None, MaintenanceOperationExecutor] = None
-        self.__operation_executor: Union[None, OperationExecutor] = None
+        self.__maintenance_operation_executor: Optional[MaintenanceOperationExecutor] = None
+        self.__operation_executor: Optional[OperationExecutor] = None
         # todo: database smuggler
-        self.__identifier: Union[None, str] = None
+        self.__multi_db_hilo: Optional[MultiDatabaseHiLoGenerator] = None
+        self.__identifier: Optional[str] = None
         self.__add_change_lock = threading.Lock()
         self.__database_changes = {}
         self.__after_close: List[Callable[[], None]] = []
@@ -327,15 +328,19 @@ def __exit__(self, exc_type, exc_val, exc_tb):
         self.close()
 
     @property
-    def thread_pool_executor(self):
+    def hilo_id_generator(self) -> Optional[MultiDatabaseHiLoGenerator]:
+        return self.__multi_db_hilo
+
+    @property
+    def thread_pool_executor(self) -> ThreadPoolExecutor:
         return self.__thread_pool_executor
 
     @property
     def subscriptions(self) -> DocumentSubscriptions:
         return self.__subscriptions
 
     @property
-    def identifier(self) -> Union[None, str]:
+    def identifier(self) -> Optional[str]:
         if self.__identifier is not None:
             return self.__identifier
 
@@ -520,9 +525,9 @@ def initialize(self) -> DocumentStore:
 
     # todo: aggressively cache
 
-    def bulk_insert(self, database_name: Optional[str] = None) -> BulkInsertOperation:
+    def bulk_insert(self, database_name: str = None, options: BulkInsertOptions = None) -> BulkInsertOperation:
         self.assert_initialized()
-        return BulkInsertOperation(self.get_effective_database(database_name), self)
+        return BulkInsertOperation(self.get_effective_database(database_name), self, options)
 
     def _assert_valid_configuration(self) -> None:
         if not self.urls: