Split model classes in its own file

Author: LeMyst

wikibaseintegrator/models/__init__.py

@@ -1,13 +1,23 @@
-from .aliases import Alias, Aliases
+from .alias import Alias
+from .aliases import Aliases
 from .basemodel import BaseModel
-from .claims import Claim, Claims
+from .claim import Claim
+from .claims import Claims
 from .descriptions import Descriptions
-from .forms import Form, Forms
+from .form import Form
+from .forms import Forms
+from .glosses import Glosses
 from .labels import Labels
-from .language_values import LanguageValue, LanguageValues
+from .language_value import LanguageValue
+from .language_values import LanguageValues
 from .lemmas import Lemmas
 from .qualifiers import Qualifiers
-from .references import Reference, References
-from .senses import Glosses, Sense, Senses
-from .sitelinks import Sitelink, Sitelinks
-from .snaks import Snak, Snaks
+from .reference import Reference
+from .references import References
+from .representations import Representations
+from .sense import Sense
+from .senses import Senses
+from .sitelink import Sitelink
+from .sitelinks import Sitelinks
+from .snak import Snak
+from .snaks import Snaks

wikibaseintegrator/models/alias.py

@@ -0,0 +1,5 @@
+from wikibaseintegrator.models.language_value import LanguageValue
+
+
+class Alias(LanguageValue):
+    pass

wikibaseintegrator/models/aliases.py

@@ -2,8 +2,8 @@
 
 from typing import Dict, List, Optional, Union
 
+from wikibaseintegrator.models.alias import Alias
 from wikibaseintegrator.models.basemodel import BaseModel
-from wikibaseintegrator.models.language_values import LanguageValue
 from wikibaseintegrator.wbi_config import config
 from wikibaseintegrator.wbi_enums import ActionIfExists
 
@@ -91,7 +91,3 @@ def from_json(self, json_data: Dict[str, List]) -> Aliases:
     # def __contains__(self, item):
     #     all_aliases = [item for sublist in list(self.aliases.values()) for item in sublist]
     #     return item in list(map(lambda x: x.value, all_aliases))
-
-
-class Alias(LanguageValue):
-    pass

wikibaseintegrator/models/claim.py

@@ -0,0 +1,249 @@
+from __future__ import annotations
+
+import copy
+from typing import Any, Callable, Dict, List, Optional, Union
+
+from wikibaseintegrator.models.basemodel import BaseModel
+from wikibaseintegrator.models.qualifiers import Qualifiers
+from wikibaseintegrator.models.references import Reference, References
+from wikibaseintegrator.models.snaks import Snak, Snaks
+from wikibaseintegrator.wbi_enums import WikibaseRank
+
+
+class Claim(BaseModel):
+    DTYPE = 'claim'
+
+    def __init__(self, qualifiers: Qualifiers = None, rank: WikibaseRank = None, references: Union[References, List[Union[Claim, List[Claim]]]] = None) -> None:
+        """
+
+        :param qualifiers:
+        :param rank:
+        :param references: A References object, a list of Claim object or a list of list of Claim object
+        """
+        self.mainsnak = Snak(datatype=self.DTYPE)
+        self.type = 'statement'
+        self.qualifiers = qualifiers or Qualifiers()
+        self.qualifiers_order = []
+        self.id = None
+        self.rank = rank or WikibaseRank.NORMAL
+        self.removed = False
+
+        self.references = References()
+
+        if isinstance(references, References):
+            self.references = references
+        elif isinstance(references, list):
+            for ref_list in references:
+                ref = Reference()
+                if isinstance(ref_list, list):
+                    snaks = Snaks()
+                    for ref_claim in ref_list:
+                        if isinstance(ref_claim, Claim):
+                            snaks.add(Snak().from_json(ref_claim.get_json()['mainsnak']))
+                        else:
+                            raise ValueError("The references must be a References object or a list of Claim object")
+                    ref.snaks = snaks
+                elif isinstance(ref_list, Claim):
+                    ref.snaks = Snaks().add(Snak().from_json(ref_list.get_json()['mainsnak']))
+                elif isinstance(ref_list, Reference):
+                    ref = ref_list
+                self.references.add(reference=ref)
+        elif references is not None:
+            raise ValueError("The references must be a References object or a list of Claim object")
+
+    @property
+    def mainsnak(self) -> Snak:
+        return self.__mainsnak
+
+    @mainsnak.setter
+    def mainsnak(self, value: Snak):
+        self.__mainsnak = value
+
+    @property
+    def type(self) -> Union[str, Dict]:
+        return self.__type
+
+    @type.setter
+    def type(self, value: Union[str, Dict]):
+        self.__type = value
+
+    @property
+    def qualifiers(self) -> Qualifiers:
+        return self.__qualifiers
+
+    @qualifiers.setter
+    def qualifiers(self, value: Qualifiers) -> None:
+        assert isinstance(value, (Qualifiers, list))
+        self.__qualifiers: Qualifiers = Qualifiers().set(value) if isinstance(value, list) else value
+
+    @property
+    def qualifiers_order(self) -> List[str]:
+        return self.__qualifiers_order
+
+    @qualifiers_order.setter
+    def qualifiers_order(self, value: List[str]):
+        self.__qualifiers_order = value
+
+    @property
+    def id(self) -> Optional[str]:
+        return self.__id
+
+    @id.setter
+    def id(self, value: Optional[str]):
+        self.__id = value
+
+    @property
+    def rank(self) -> WikibaseRank:
+        return self.__rank
+
+    @rank.setter
+    def rank(self, value: WikibaseRank):
+        """Parse the rank. The enum thows an error if it is not one of the recognized values"""
+        self.__rank = WikibaseRank(value)
+
+    @property
+    def references(self) -> References:
+        return self.__references
+
+    @references.setter
+    def references(self, value: References):
+        self.__references = value
+
+    @property
+    def removed(self) -> bool:
+        return self.__removed
+
+    @removed.setter
+    def removed(self, value: bool):
+        self.__removed = value
+
+    def remove(self, remove=True) -> None:
+        self.removed = remove
+
+    def update(self, claim: Claim) -> None:
+        self.mainsnak = claim.mainsnak
+        self.qualifiers = claim.qualifiers
+        self.qualifiers_order = claim.qualifiers_order
+        self.rank = claim.rank
+        self.references = claim.references
+
+    def from_json(self, json_data: Dict[str, Any]) -> Claim:
+        """
+
+        :param json_data: a JSON representation of a Claim
+        """
+        self.mainsnak = Snak().from_json(json_data['mainsnak'])
+        self.type = str(json_data['type'])
+        if 'qualifiers' in json_data:
+            self.qualifiers = Qualifiers().from_json(json_data['qualifiers'])
+        if 'qualifiers-order' in json_data:
+            self.qualifiers_order = list(json_data['qualifiers-order'])
+        self.id = str(json_data['id'])
+        self.rank: WikibaseRank = WikibaseRank(json_data['rank'])
+        if 'references' in json_data:
+            self.references = References().from_json(json_data['references'])
+
+        return self
+
+    def get_json(self) -> Dict[str, Any]:
+        json_data: Dict[str, Union[str, List[Dict], List[str], Dict[str, str], Dict[str, List], None]] = {
+            'mainsnak': self.mainsnak.get_json(),
+            'type': self.type,
+            'id': self.id,
+            'rank': self.rank.value
+        }
+        # Remove id if it's a temporary one
+        if not self.id:
+            del json_data['id']
+        if len(self.qualifiers) > 0:
+            json_data['qualifiers'] = self.qualifiers.get_json()
+            json_data['qualifiers-order'] = list(self.qualifiers_order)
+        if len(self.references) > 0:
+            json_data['references'] = self.references.get_json()
+        if self.removed:
+            if self.id:
+                json_data['remove'] = ''
+        return json_data
+
+    def has_equal_qualifiers(self, other: Claim) -> bool:
+        # check if the qualifiers are equal with the 'other' object
+        self_qualifiers = copy.deepcopy(self.qualifiers)
+        other_qualifiers = copy.deepcopy(other.qualifiers)
+
+        if len(self_qualifiers) != len(other_qualifiers):
+            return False
+
+        for property_number in self_qualifiers.qualifiers:
+            if property_number not in other_qualifiers.qualifiers:
+                return False
+
+            if len(self_qualifiers.qualifiers[property_number]) != len(other_qualifiers.qualifiers[property_number]):
+                return False
+
+            flg = [False for _ in range(len(self_qualifiers.qualifiers[property_number]))]
+            for count, i in enumerate(self_qualifiers.qualifiers[property_number]):
+                for q in other_qualifiers:
+                    if i == q:
+                        flg[count] = True
+            if not all(flg):
+                return False
+
+        return True
+
+    # TODO: rewrite this?
+    def __contains__(self, item):
+        if isinstance(item, Claim):
+            return self == item
+
+        if isinstance(item, str):
+            return self.mainsnak.datavalue == item
+
+        return super().__contains__(item)
+
+    def __eq__(self, other):
+        if isinstance(other, Claim):
+            return self.mainsnak.datavalue == other.mainsnak.datavalue and self.mainsnak.property_number == other.mainsnak.property_number and self.has_equal_qualifiers(other)
+
+        if isinstance(other, str):
+            return self.mainsnak.property_number == other
+
+        raise super().__eq__(other)
+
+    def equals(self, that: Claim, include_ref: bool = False, fref: Callable = None) -> bool:
+        """
+        Tests for equality of two statements.
+        If comparing references, the order of the arguments matters!!!
+        self is the current statement, the next argument is the new statement.
+        Allows passing in a function to use to compare the references 'fref'. Default is equality.
+        fref accepts two arguments 'oldrefs' and 'newrefs', each of which are a list of references,
+        where each reference is a list of statements
+        """
+
+        if not include_ref:
+            # return the result of BaseDataType.__eq__, which is testing for equality of value and qualifiers
+            return self == that
+
+        if self != that:
+            return False
+
+        if fref is None:
+            return Claim.refs_equal(self, that)
+
+        return fref(self, that)
+
+    @staticmethod
+    def refs_equal(olditem: Claim, newitem: Claim) -> bool:
+        """
+        tests for exactly identical references
+        """
+
+        oldrefs = olditem.references
+        newrefs = newitem.references
+
+        def ref_equal(oldref: References, newref: References) -> bool:
+            return (len(oldref) == len(newref)) and all(x in oldref for x in newref)
+
+        return len(oldrefs) == len(newrefs) and all(any(ref_equal(oldref, newref) for oldref in oldrefs) for newref in newrefs)
+
+    def get_sparql_value(self) -> str:
+        pass