🦖 Make Library.add fail on duplicate keys by default

bibtexparser/library.py

@@ -17,47 +17,69 @@
 class Library:
     """A collection of parsed bibtex blocks."""
 
-    def __init__(self, blocks: Union[List[Block], None] = None):
+    def __init__(
+        self,
+        blocks: Union[List[Block], None] = None,
+        fail_on_duplicate_key: bool = True,
+    ):
         self._blocks = []
         self._entries_by_key = dict()
         self._strings_by_key = dict()
         if blocks is not None:
-            self.add(blocks)
+            self.add(blocks, fail_on_duplicate_key=fail_on_duplicate_key)
 
-    def add(self, blocks: Union[List[Block], Block], fail_on_duplicate_key: bool = False):
+    def add(self, blocks: Union[List[Block], Block], fail_on_duplicate_key: bool = True):
         """Add blocks to library.
 
-        The adding is key-safe, i.e., it is made sure that no duplicate keys are added.
-        for the same type (i.e., String or Entry). Duplicates are silently replaced with
-        a DuplicateKeyBlock.
+        The adding is key-safe, i.e., it is made sure that no duplicate keys are added
+        for the same type (i.e., String or Entry). Depending on `fail_on_duplicate_key`,
+        duplicates either cause a ValueError (default) or are replaced with
+        a DuplicateBlockKeyBlock.
 
         :param blocks: Block or list of blocks to add.
         :param fail_on_duplicate_key:
-            If True, raises ValueError if a block was replaced with a DuplicateKeyBlock.
+            If True (default), raises ValueError on duplicate keys, leaving the
+            library unchanged. If False, duplicates are silently replaced with
+            DuplicateBlockKeyBlock instances, which can be inspected via
+            `library.failed_blocks`. This is e.g. used when parsing, where a
+            bibtex file with duplicate keys should not raise.
+        :raises ValueError: If fail_on_duplicate_key is True and a duplicate key
+            is found. In this case, no blocks are added to the library.
         """
         if isinstance(blocks, Block):
             blocks = [blocks]
 
-        _added_blocks = []
-        for block in blocks:
-            # This may replace block with a DuplicateEntryKeyBlock
-            block = self._add_to_dicts(block)
-            self._blocks.append(block)
-            _added_blocks.append(block)
-
         if fail_on_duplicate_key:
-            duplicate_keys = []
-            for original, added in zip(blocks, _added_blocks):
-                if original is not added and isinstance(added, DuplicateBlockKeyBlock):
-                    duplicate_keys.append(added.key)
-
+            duplicate_keys = self._find_duplicate_keys(blocks)
             if len(duplicate_keys) > 0:
                 raise ValueError(
                     f"Duplicate keys found: {duplicate_keys}. "
-                    f"Duplicate entries have been added to the library as DuplicateBlockKeyBlock."
-                    f"Use `library.failed_blocks` to access them. "
+                    f"No blocks were added to the library. "
+                    f"To add duplicates as DuplicateBlockKeyBlock instances instead, "
+                    f"use `library.add(blocks, fail_on_duplicate_key=False)`."
                 )
 
+        for block in blocks:
+            # This may replace block with a DuplicateEntryKeyBlock
+            block = self._add_to_dicts(block)
+            self._blocks.append(block)
+
+    def _find_duplicate_keys(self, blocks: List[Block]) -> List[str]:
+        """Keys of blocks that would become duplicates when added to the library."""
+        duplicate_keys = []
+        seen_entry_keys = set(self._entries_by_key)
+        seen_string_keys = set(self._strings_by_key)
+        for block in blocks:
+            if isinstance(block, Entry):
+                if block.key in seen_entry_keys:
+                    duplicate_keys.append(block.key)
+                seen_entry_keys.add(block.key)
+            elif isinstance(block, String):
+                if block.key in seen_string_keys:
+                    duplicate_keys.append(block.key)
+                seen_string_keys.add(block.key)
+        return duplicate_keys
+
     def remove(self, blocks: Union[List[Block], Block]):
         """Remove blocks from library.
 

bibtexparser/middlewares/middleware.py

@@ -97,7 +97,9 @@ def transform(self, library: "Library") -> "Library":
             # Case 4: Something else. Error.
             else:
                 raise TypeError(f"Illegal output type from transform_block: {type(transformed)}")
-        return Library(blocks=blocks)
+        # Duplicate keys (e.g. introduced by a block transformation) must not raise
+        # mid-pipeline; they are exposed as `library.failed_blocks`.
+        return Library(blocks=blocks, fail_on_duplicate_key=False)
 
     def transform_block(
         self, block: Block, library: "Library"

bibtexparser/middlewares/sorting_blocks.py

@@ -128,11 +128,12 @@ def transform(self, library: Library) -> Library:
             block_junks = self._block_junks(blocks)
             block_junks.sort(key=lambda junk: self._key(junk.main_block), reverse=self._reverse)
             return Library(
-                blocks=[block for block_junk in block_junks for block in block_junk.blocks]
+                blocks=[block for block_junk in block_junks for block in block_junk.blocks],
+                fail_on_duplicate_key=False,
             )
         else:
             blocks.sort(key=self._key, reverse=self._reverse)
-            return Library(blocks=blocks)
+            return Library(blocks=blocks, fail_on_duplicate_key=False)
 
 
 class SortBlocksByTypeAndKeyMiddleware(SortBlocksMiddleware):

bibtexparser/splitter.py

@@ -323,9 +323,9 @@ def split(self, library: Optional[Library] = None) -> Library:
                     elif m_val.startswith("@preamble"):
                         library.add(self._handle_preamble())
                     elif m_val.startswith("@string"):
-                        library.add(self._handle_string(m))
+                        library.add(self._handle_string(m), fail_on_duplicate_key=False)
                     else:
-                        library.add(self._handle_entry(m, m_val))
+                        library.add(self._handle_entry(m, m_val), fail_on_duplicate_key=False)
 
                 except BlockAbortedException as e:
                     logger.warning(

tests/test_library.py

@@ -20,7 +20,7 @@ def test_replace_with_duplicates():
     """Test that replace() works when there are duplicate values. See issue 404."""
     library = Library()
     library.add(get_dummy_entry())
-    library.add(get_dummy_entry())
+    library.add(get_dummy_entry(), fail_on_duplicate_key=False)
     # Test precondition
     assert len(library.blocks) == 2
     assert len(library.failed_blocks) == 1
@@ -59,10 +59,38 @@ def test_replace_fail_on_duplicate():
     assert library.entries[1].key == "duplicateKey"
 
 
-def test_fail_on_duplicate_add():
+def test_add_fails_on_duplicate_by_default():
+    library = Library()
+    library.add(get_dummy_entry())
+    with pytest.raises(ValueError):
+        library.add(get_dummy_entry())
+    # The failed add must leave the library unchanged.
+    assert len(library.blocks) == 1
+    assert len(library.failed_blocks) == 0
+
+
+def test_add_with_duplicates_in_same_call_fails_atomically():
+    unique_entry = get_dummy_entry()
+    unique_entry.key = "uniqueKey"
+    library = Library()
+    with pytest.raises(ValueError):
+        library.add([unique_entry, get_dummy_entry(), get_dummy_entry()])
+    # Not even the non-duplicate blocks of the failed call are added.
+    assert len(library.blocks) == 0
+
+
+def test_add_duplicate_does_not_fail_if_disabled():
     library = Library()
     library.add(get_dummy_entry())
+    library.add(get_dummy_entry(), fail_on_duplicate_key=False)
+    assert len(library.blocks) == 2
+    assert len(library.failed_blocks) == 1
+
+
+def test_constructor_fails_on_duplicate_by_default():
     with pytest.raises(ValueError):
-        library.add(get_dummy_entry(), fail_on_duplicate_key=True)
+        Library(blocks=[get_dummy_entry(), get_dummy_entry()])
+
+    library = Library(blocks=[get_dummy_entry(), get_dummy_entry()], fail_on_duplicate_key=False)
     assert len(library.blocks) == 2
     assert len(library.failed_blocks) == 1

tests/test_writer.py

@@ -157,7 +157,8 @@ def test_write_manually_created_duplicate_entries_raises():
     """Reproduces issue #400: duplicate keys on programmatically created entries."""
     library = Library()
     library.add(_dummy_entry())
-    library.add(_dummy_entry())
+    # Silently-added duplicates (no raw bibtex) must still be caught at write time.
+    library.add(_dummy_entry(), fail_on_duplicate_key=False)
     with pytest.raises(ValueError, match="failed_blocks"):
         writer.write(library)