Add check_constraints() helper for constraint checking

test/conftest.py

@@ -79,6 +79,7 @@ def __init__(self, mocker: requests_mock_lib.Mocker, base_url: str = 'https://wi
         self.search_results: list[dict] = []  # wbsearchentities results
         self.fulltext_results: list[dict] = []  # list=search results
         self.sparql_bindings: list[dict] = []  # bindings returned by the SPARQL endpoint
+        self.constraint_results: dict[str, list[dict]] = {}  # wbcheckconstraints results, keyed by entity id
         self.valid_credentials: dict[str, str] = {}  # user -> password accepted by (client)login
         self.login_token = 'aabbccddeeff+\\'
         self.csrf_token = '0123456789abcdef+\\'
@@ -368,6 +369,11 @@ def _action_wbremoveclaims(self, params: dict[str, str]) -> dict:
     def _action_delete(self, params: dict[str, str]) -> dict:
         return {'delete': {'title': params.get('title', ''), 'reason': params.get('reason', 'mock deletion'), 'logid': 1}}
 
+    def _action_wbcheckconstraints(self, params: dict[str, str]) -> dict:
+        requested_ids = params['id'].split('|') if 'id' in params else list(self.constraint_results.keys())
+        claims = {entity_id: deepcopy(self.constraint_results[entity_id]) for entity_id in requested_ids if entity_id in self.constraint_results}
+        return {'claims': claims, 'success': 1}
+
 
 # ---------------------------------------------------------------------- #
 # SPARQL helpers usable from the tests

test/test_wbi_helpers.py

@@ -9,9 +9,9 @@
 
 from wikibaseintegrator.wbi_config import config as wbi_config
 from wikibaseintegrator.wbi_exceptions import AnonymousEditNotAllowedError, MaxRetriesReachedException, ModificationFailed, MWApiError, NonExistentEntityError, SaveFailed
-from wikibaseintegrator.wbi_helpers import (download_entity_ttl, execute_sparql_query, format2wbi, format_amount, fulltext_search, generate_entity_instances, get_user_agent,
-                                            lexeme_edit_sense, lexeme_remove_form, lexeme_remove_sense, mediawiki_api_call, mediawiki_api_call_helper, merge_items, remove_claims,
-                                            search_entities)
+from wikibaseintegrator.wbi_helpers import (check_constraints, download_entity_ttl, execute_sparql_query, format2wbi, format_amount, fulltext_search, generate_entity_instances,
+                                            get_user_agent, lexeme_edit_sense, lexeme_remove_form, lexeme_remove_sense, mediawiki_api_call, mediawiki_api_call_helper, merge_items,
+                                            remove_claims, search_entities)
 
 
 class FakeLogin:
@@ -254,6 +254,39 @@ def test_search_strict_language_and_limit_parameters(self, wikibase):
         assert len(results) == 10
 
 
+class TestCheckConstraints:
+    def test_requires_entity_id_or_claim_id(self):
+        with pytest.raises(ValueError):
+            check_constraints()
+
+    def test_check_by_entity_id(self, wikibase):
+        wikibase.constraint_results = {
+            'Q582': [{'id': 'Q582$1', 'results': [{'status': 'violation', 'property': 'P17'}]}]
+        }
+
+        result = check_constraints(entity_id='Q582')
+        assert wikibase.last_request['action'] == 'wbcheckconstraints'
+        assert wikibase.last_request['id'] == 'Q582'
+        assert result['claims']['Q582'][0]['results'][0]['status'] == 'violation'
+
+    def test_check_by_multiple_entity_ids(self, wikibase):
+        wikibase.constraint_results = {
+            'Q582': [{'id': 'Q582$1', 'results': []}],
+            'Q1': [{'id': 'Q1$1', 'results': []}],
+        }
+
+        check_constraints(entity_id=['Q582', 'Q1'])
+        assert wikibase.last_request['id'] == 'Q582|Q1'
+
+    def test_check_by_claim_id(self, wikibase):
+        check_constraints(claim_id='Q582$1d2e3f4a-5b6c-7d8e-9f0a-1b2c3d4e5f6a')
+        assert wikibase.last_request['claimid'] == 'Q582$1d2e3f4a-5b6c-7d8e-9f0a-1b2c3d4e5f6a'
+
+    def test_check_with_status_filter(self, wikibase):
+        check_constraints(entity_id='Q582', status=['violation', 'warning'])
+        assert wikibase.last_request['status'] == 'violation|warning'
+
+
 class TestFulltextSearch:
     def test_fulltext_search(self, wikibase):
         wikibase.fulltext_results = [{'ns': 0, 'title': 'Q582', 'pageid': 892, 'snippet': 'Villeurbanne'}]

wikibaseintegrator/wbi_helpers.py

@@ -433,6 +433,44 @@ def remove_claims(claim_id: str, summary: str | None = None, baserevid: int | No
     return mediawiki_api_call_helper(data=params, is_bot=is_bot, **kwargs)
 
 
+def check_constraints(entity_id: str | list[str] | None = None, claim_id: str | list[str] | None = None, status: list[str] | None = None, allow_anonymous: bool = True,
+                      **kwargs: Any) -> dict:
+    """
+    Check the constraint violations of one or more entities or claims, using the target Wikibase instance's own constraint checker
+    (the ``wbcheckconstraints`` API module provided by the `WikibaseQualityConstraints
+    <https://www.mediawiki.org/wiki/Extension:WikibaseQualityConstraints>`_ extension).
+
+    This relies entirely on the constraints configured on the Wikibase instance being queried (e.g. on Wikidata itself), rather than
+    on a hardcoded or Wikidata-specific set of rules, so it works the same way against any Wikibase instance that has the extension
+    installed. If the extension isn't installed, the underlying API call will fail with a :class:`~wikibaseintegrator.wbi_exceptions.MWApiError`.
+
+    :param entity_id: One or more entity IDs (item, property, lexeme, etc.) whose claims should be checked.
+    :param claim_id: One or more claim GUIDs to check, instead of checking whole entities.
+    :param status: Only return results with one of these statuses, e.g. ``['violation']``. By default, the API returns every status.
+    :param allow_anonymous: Allow anonymous interaction with the MediaWiki API. 'True' by default since this is a read-only action.
+    :param kwargs: Extra parameters for mediawiki_api_call_helper()
+    :return: The data returned by the API as a dictionary
+    """
+    if not entity_id and not claim_id:
+        raise ValueError("You must provide either 'entity_id' or 'claim_id'.")
+
+    params: dict[str, Any] = {
+        'action': 'wbcheckconstraints',
+        'format': 'json'
+    }
+
+    if entity_id:
+        params.update({'id': '|'.join(entity_id) if isinstance(entity_id, list) else entity_id})
+
+    if claim_id:
+        params.update({'claimid': '|'.join(claim_id) if isinstance(claim_id, list) else claim_id})
+
+    if status:
+        params.update({'status': '|'.join(status)})
+
+    return mediawiki_api_call_helper(data=params, allow_anonymous=allow_anonymous, **kwargs)
+
+
 def search_entities(search_string: str, language: str | None = None, strict_language: bool = False, search_type: str = 'item', max_results: int = 50, dict_result: bool = False,
                     allow_anonymous: bool = True, **kwargs: Any) -> list[dict[str, Any]]:
     """