Add Item.merge() with tests

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: ghuron

src/wdpy/item.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 import json
 import logging
+import uuid
 from typing import Any, Dict, List, Optional
 
 from wdpy import Statement, SourceItem, get_entities, api_write
@@ -76,6 +77,23 @@ def compress(statement: Statement) -> None:
         for prop, items in grouped.items():
             self.claims[prop] = items
 
+    def merge(self, statement: Statement) -> Statement:
+        """Upsert a statement into this item's claims.
+
+        If an existing statement with the same mainsnak value and compatible
+        qualifiers is found, references are merged into it and it is returned.
+        Otherwise the statement is appended as a new claim (with an id assigned
+        when the item already has a qid) and returned.
+        """
+        self._ensure_loaded()
+        bucket = self.claims.setdefault(statement.mainsnak.property, [])
+        if (existing := statement.upsert(bucket)) is None:
+            if self.qid:
+                statement.id = f'{self.qid}${uuid.uuid4()}'
+            bucket.append(statement)
+            return statement
+        return existing
+
     def json(self) -> str:
         data: Dict[str, Any] = {}
         if self.qid:

tests/test_item.py

@@ -80,6 +80,73 @@ def test_write_creates_item(self, api_write_mock, *_):
         self.assertEqual('item', kwargs.get('new'))
         self.assertIn('data', kwargs)
 
+class Merge(unittest.TestCase):
+
+    def setUp(self):
+        self.item = wdpy.Item('Q1')
+        self.item._loaded = True
+
+    def _s(self, val='Q5', prop='P31'):
+        return wdpy.Statement(wdpy.Snak(prop, (val,)))
+
+    # ── insert ────────────────────────────────────────────────────────────────
+
+    def test_new_property_bucket_created(self):
+        s = self._s()
+        self.item.merge(s)
+        self.assertIn('P31', self.item.claims)
+
+    def test_new_statement_appended(self):
+        s = self._s()
+        result = self.item.merge(s)
+        self.assertIn(s, self.item.claims['P31'])
+        self.assertIs(result, s)
+
+    def test_id_assigned_when_qid_known(self):
+        self.item.merge(self._s())
+        self.assertTrue(self.item.claims['P31'][0].id.startswith('Q1$'))
+
+    def test_no_id_assigned_without_qid(self):
+        self.item.qid = None
+        s = self._s()
+        self.item.merge(s)
+        self.assertIsNone(s.id)
+
+    def test_second_distinct_value_also_inserted(self):
+        self.item.merge(self._s('Q5'))
+        self.item.merge(self._s('Q6'))
+        self.assertEqual(len(self.item.claims['P31']), 2)
+
+    def test_multiple_properties_independent(self):
+        self.item.merge(self._s('Q5', 'P31'))
+        self.item.merge(self._s('Q5', 'P21'))
+        self.assertEqual(len(self.item.claims['P31']), 1)
+        self.assertEqual(len(self.item.claims['P21']), 1)
+
+    # ── upsert ────────────────────────────────────────────────────────────────
+
+    def test_duplicate_returns_existing(self):
+        existing = self._s()
+        self.item.claims['P31'] = [existing]
+        result = self.item.merge(self._s())
+        self.assertIs(result, existing)
+        self.assertEqual(len(self.item.claims['P31']), 1)
+
+    def test_duplicate_does_not_overwrite_existing_id(self):
+        existing = self._s()
+        existing.id = 'Q1$original'
+        self.item.claims['P31'] = [existing]
+        self.item.merge(self._s())
+        self.assertEqual(self.item.claims['P31'][0].id, 'Q1$original')
+
+    # ── _ensure_loaded called ─────────────────────────────────────────────────
+
+    @patch('wdpy.item.get_entities', return_value={})
+    def test_triggers_load_if_not_loaded(self, mock_get):
+        item = wdpy.Item('Q99')
+        item.merge(self._s())
+        mock_get.assert_called_once()
+
     @patch('wdpy.item.api_write', return_value={'entity': {'id': 'Q7'}})
     def test_write_updates_existing_item(self, api_write_mock, *_):
         item = wdpy.Item('Q7')