Wikidata

WikidataClient reads items from the Wikidata API and SPARQL endpoint, with language-aware label selection and database-backed caching of responses.

Wikidata is a useful source of structured data on politicians, companies, and other entities of interest. This module is the low-level client used by the Wikidata enricher and by crawlers that turn Wikidata items into FollowTheMoney entities. It handles the parts that are error-prone to reimplement: request throttling and retries, response caching via a nomenklatura.cache.Cache, and picking a display label from the many languages an item may carry.

The client returns items as Item objects, which expose labels, aliases, descriptions, and claims. A Claim is one property statement on an item — for example P569 (date of birth) — with its qualifiers and references. Text values are wrapped in LangText, which keeps the language tag alongside the string.

Fetching an item requires a Cache, which stores API responses in the same SQL database the rest of nomenklatura uses:

from followthemoney import Dataset
from nomenklatura.cache import Cache
from nomenklatura.db import make_session
from nomenklatura.wikidata import WikidataClient

dataset = Dataset.make({"name": "wikidata_demo", "title": "Wikidata demo"})
with make_session() as session:
    cache = Cache(session, dataset, create=True)
    client = WikidataClient(cache)
    item = client.fetch_item("Q7747")
    if item is not None:
        print(item.id, client.get_label(item.id))

Interface

nomenklatura.wikidata.WikidataClient

Read items and labels from the Wikidata API and SPARQL endpoint.

Responses are cached in a SQL-backed Cache so that crawlers and enrichers can re-run without fetching the same data again, and requests carry a descriptive user agent and retry handling to stay within Wikidata's API etiquette.

Source code in nomenklatura/wikidata/client.py

class WikidataClient:
    """Read items and labels from the Wikidata API and SPARQL endpoint.

    Responses are cached in a SQL-backed `Cache` so that crawlers and enrichers
    can re-run without fetching the same data again, and requests carry a
    descriptive user agent and retry handling to stay within Wikidata's API
    etiquette."""

    WD_API = "https://www.wikidata.org/w/api.php"
    QUERY_API = "https://query.wikidata.org/sparql"
    CACHE_SHORT = 1
    CACHE_MEDIUM = CACHE_SHORT * 7
    CACHE_LONG = CACHE_SHORT * 30
    # Retry budget for the transient in-band API errors Wikidata returns as
    # HTTP 200 (see `_fetch_entities`). Class attributes so they stay tunable
    # and test-overridable (set the backoff to 0 to skip real sleeps in tests).
    API_MAX_ATTEMPTS = 3
    API_RETRY_BACKOFF = 2.0  # seconds, exponential base: 2s, 4s between attempts

    LABEL_PREFIX = "wd:lb:"
    LABEL_CACHE_DAYS = 100

    def __init__(
        self,
        cache: Cache,
        session: Session | None = None,
        cache_days: int = 14,
        reference_time: datetime | None = None,
    ) -> None:
        self.cache = cache
        # A bare session gets 403'd (default UA) and throttled by Wikidata, so
        # default to a configured session with a descriptive UA and retries.
        self.session = session or make_session()
        self.cache_days = cache_days
        # The point in time against which claim validity is evaluated (see
        # `Claim.is_ended`). Crawlers pass their pinned run time so a whole
        # run resolves "ended" against one deterministic instant.
        self.reference_time = (
            reference_time if reference_time is not None else utc_now()
        )
        # self.cache.preload(f"{self.LABEL_PREFIX}%")

    @lru_cache(maxsize=MEMO_SMALL)
    def fetch_item(
        self,
        qid: str,
        cache_days: int | None = None,
        modified_at: datetime | None = None,
    ) -> Item | None:
        # https://www.mediawiki.org/wiki/Wikibase/API
        # https://www.wikidata.org/w/api.php?action=help&modules=wbgetentities
        #
        # `modified_at` (typically the item's schema:dateModified) invalidates
        # cache entries stored before that timestamp. This lets callers pin a
        # fixed `cache_days` while still refetching items known to have changed
        # upstream since the cached copy was written.
        params = {
            "format": "json",
            "ids": qid,
            "action": "wbgetentities",
            # Ask for sitelink URLs for proper wikipedia links:
            "props": "info|sitelinks/urls|aliases|labels|descriptions|claims|datatype",
        }
        url = build_url(self.WD_API, params=params)
        cache_days = cache_days if cache_days is not None else self.cache_days
        raw = self.cache.get(url, max_age=cache_days, min_timestamp=modified_at)
        if raw is None:
            log.debug(
                "Cache MISS fetching Wikidata item: %s cache_days=%s modified_at=%s",
                qid,
                cache_days,
                modified_at,
            )
            # `_fetch_entities` retries transient in-band errors; it returns
            # None once they're exhausted (item skipped for this run) and
            # otherwise a cacheable body, so nothing bad reaches the cache.
            raw = self._fetch_entities(url)
            if raw is None:
                return None
            self.cache.set(url, raw)
        else:
            log.debug(
                "Cache HIT fetching Wikidata item: %s cache_days=%s modified_at=%s",
                qid,
                cache_days,
                modified_at,
            )
        data = json.loads(raw)
        entities = data.get("entities")
        if entities is None:
            # Only a cached `no-such-entity` reaches here: a deleted or
            # never-created QID is a permanent fact, read as "no such item".
            return None
        entity = entities.get(qid)
        if entity is None or "missing" in entity:
            return None
        item = Item(self, entity)
        if item.id != qid:
            # Redirected/merged item:
            #
            # The modification date doesn't relate to the new QID, but let's
            # use it as its minimum age nonetheless since there's a good chance
            # the replacement item was edited around the time the original was merged.
            return self.fetch_item(
                item.id, cache_days=cache_days, modified_at=modified_at
            )
        return item

    def _fetch_entities(self, url: str) -> str | None:
        """GET a wbgetentities URL, retrying the transient errors Wikidata hides
        in HTTP 200 bodies (DB lag, rate limits, internal errors).

        Reach for this instead of a bare `session.get` for any wbgetentities
        call: the session's status-code retries can't see an in-band 200 error,
        so a single upstream blip would otherwise abort a whole crawl. Returns
        the raw response text once it is cacheable — a populated `entities` map,
        or a permanent `no-such-entity`. Returns None if a transient error
        outlasts every retry: the caller then reads the item as absent for this
        run, and, because nothing is cached, a later run refetches it.
        """
        error: Any = None
        for attempt in range(self.API_MAX_ATTEMPTS):
            res = self.session.get(url)
            res.raise_for_status()
            raw = res.text
            data = json.loads(raw)
            if data.get("entities") is not None:
                return raw
            error = data.get("error", {})
            if error.get("code") == "no-such-entity":
                return raw
            log.warning(
                "Transient Wikidata API error (attempt %d/%d): %r",
                attempt + 1,
                self.API_MAX_ATTEMPTS,
                error,
            )
            if attempt + 1 < self.API_MAX_ATTEMPTS:
                time.sleep(self.API_RETRY_BACKOFF * (2**attempt))
        log.warning("Skipping %s after %d failed attempts: %r", url, attempt + 1, error)
        return None

    @lru_cache(maxsize=100000)
    def get_label(self, qid: str) -> LangText:
        cache_key = f"{self.LABEL_PREFIX}{qid}"
        cached = self.cache.get_json(cache_key, max_age=self.LABEL_CACHE_DAYS)
        if cached is not None:
            return LangText.parse(cached)
        params = {
            "format": "json",
            "ids": qid,
            "action": "wbgetentities",
            "props": "labels",
        }
        url = build_url(self.WD_API, params=params)
        raw = self._fetch_entities(url)
        if raw is None:
            # Transient error outlasted the retries: treat the label as absent.
            return LangText(None)
        entities = json.loads(raw).get("entities")
        if entities is None:
            # Only a `no-such-entity` body reaches here (see `_fetch_entities`).
            return LangText(None)
        entity = entities.get(qid)
        if entity is None or "missing" in entity:
            return LangText(None)
        labels = LangText.from_dict(entity.get("labels", {}))
        label = LangText.pick(labels)
        if label is None:
            label = LangText(qid)
        label.original = qid
        self.cache.set_json(cache_key, label.pack())
        return label

    def query(self, query_text: str, cache_days: int | None = None) -> SparqlResponse:
        """Query the Wikidata SPARQL endpoint.

        Args:
          cache_days: overrides the client-level default for this call.
        """
        clean_text = squash_spaces(query_text)
        if len(clean_text) == 0:
            raise RuntimeError("Invalid query: %r" % query_text)
        params = {"query": clean_text}
        url = build_url(self.QUERY_API, params=params)
        effective_cache = cache_days if cache_days is not None else self.cache_days
        raw = self.cache.get(url, max_age=effective_cache)
        if raw is None:
            log.debug(
                "Cache MISS fetching Wikidata SPARQL query: %s cache_days=%s",
                clean_text,
                effective_cache,
            )
            headers = {"Accept": "application/sparql-results+json"}
            res = self.session.get(url, headers=headers)
            res.raise_for_status()
            raw = res.text
            self.cache.set(url, raw)
        else:
            log.debug(
                "Cache HIT fetching Wikidata SPARQL query: %s cache_days=%s",
                clean_text,
                effective_cache,
            )
        try:
            data = json.loads(raw)
        except json.JSONDecodeError as err:
            self.cache.delete(url)
            log.exception("Failed to parse JSON: %s", err)
            return SparqlResponse(clean_text, {})
        return SparqlResponse(clean_text, data)

    def search_items(
        self, entity: StatementEntity, aliases: bool = False, limit: int = 7
    ) -> list[str]:
        """Find Wikidata QIDs that might be the same as an OpenSanctions entity.

        Reach for this when reconciling an OS entity against Wikidata: it runs the
        entity's names through the `wbsearchentities` API and returns candidate
        QIDs for a downstream matcher to rank. It returns only QIDs — the caller
        decides which items to fetch and how to project them — so the client stays
        decoupled from the matcher's needs.

        All `name` values are searched. With `aliases`, the search also covers
        aliases (every matchable name-type value), trading more API calls for
        better recall on transliterated or aliased names. `limit` is the per-name
        result cap (the `wbsearchentities` default is 7, max 50); raise it for
        better recall on common names.
        """
        if aliases:
            names = entity.get_type_values(registry.name, matchable=True)
        else:
            names = entity.get("name", quiet=True)
        qids: list[str] = []
        seen: set[str] = set()
        for name in names:
            for qid in self._search_name(name, limit=limit):
                if qid not in seen:
                    seen.add(qid)
                    qids.append(qid)
        return qids

    def _search_name(self, name: str, limit: int = 7) -> list[str]:
        if not name.strip():
            return []
        params = {
            "format": "json",
            "action": "wbsearchentities",
            "type": "item",
            "language": "en",
            "strictlanguage": "false",
            "limit": str(limit),
            "search": name,
        }
        url = build_url(self.WD_API, params=params)
        raw = self.cache.get(url, max_age=self.cache_days)
        if raw is None:
            res = self.session.get(url)
            res.raise_for_status()
            raw = res.text
            self.cache.set(url, raw)
        data = json.loads(raw)
        results = data.get("search")
        if results is None:
            # A response without a `search` key is malformed/transient; don't
            # keep it around to be served from cache.
            self.cache.delete(url)
            log.info("Wikidata search has no results: %s", name)
            return []
        qids: list[str] = []
        for result in results:
            qid = result.get("id")
            if qid is not None and is_qid(qid):
                qids.append(qid)
        return qids

    def __repr__(self) -> str:
        return "<WikidataClient()>"

    def __hash__(self) -> int:
        return 42

query(query_text, cache_days=None)

Query the Wikidata SPARQL endpoint.

Parameters:

Name	Type	Description	Default
cache_days	int | None	overrides the client-level default for this call.	None

Source code in nomenklatura/wikidata/client.py

def query(self, query_text: str, cache_days: int | None = None) -> SparqlResponse:
    """Query the Wikidata SPARQL endpoint.

    Args:
      cache_days: overrides the client-level default for this call.
    """
    clean_text = squash_spaces(query_text)
    if len(clean_text) == 0:
        raise RuntimeError("Invalid query: %r" % query_text)
    params = {"query": clean_text}
    url = build_url(self.QUERY_API, params=params)
    effective_cache = cache_days if cache_days is not None else self.cache_days
    raw = self.cache.get(url, max_age=effective_cache)
    if raw is None:
        log.debug(
            "Cache MISS fetching Wikidata SPARQL query: %s cache_days=%s",
            clean_text,
            effective_cache,
        )
        headers = {"Accept": "application/sparql-results+json"}
        res = self.session.get(url, headers=headers)
        res.raise_for_status()
        raw = res.text
        self.cache.set(url, raw)
    else:
        log.debug(
            "Cache HIT fetching Wikidata SPARQL query: %s cache_days=%s",
            clean_text,
            effective_cache,
        )
    try:
        data = json.loads(raw)
    except json.JSONDecodeError as err:
        self.cache.delete(url)
        log.exception("Failed to parse JSON: %s", err)
        return SparqlResponse(clean_text, {})
    return SparqlResponse(clean_text, data)

search_items(entity, aliases=False, limit=7)

Find Wikidata QIDs that might be the same as an OpenSanctions entity.

Reach for this when reconciling an OS entity against Wikidata: it runs the entity's names through the wbsearchentities API and returns candidate QIDs for a downstream matcher to rank. It returns only QIDs — the caller decides which items to fetch and how to project them — so the client stays decoupled from the matcher's needs.

All name values are searched. With aliases, the search also covers aliases (every matchable name-type value), trading more API calls for better recall on transliterated or aliased names. limit is the per-name result cap (the wbsearchentities default is 7, max 50); raise it for better recall on common names.

Source code in nomenklatura/wikidata/client.py

def search_items(
    self, entity: StatementEntity, aliases: bool = False, limit: int = 7
) -> list[str]:
    """Find Wikidata QIDs that might be the same as an OpenSanctions entity.

    Reach for this when reconciling an OS entity against Wikidata: it runs the
    entity's names through the `wbsearchentities` API and returns candidate
    QIDs for a downstream matcher to rank. It returns only QIDs — the caller
    decides which items to fetch and how to project them — so the client stays
    decoupled from the matcher's needs.

    All `name` values are searched. With `aliases`, the search also covers
    aliases (every matchable name-type value), trading more API calls for
    better recall on transliterated or aliased names. `limit` is the per-name
    result cap (the `wbsearchentities` default is 7, max 50); raise it for
    better recall on common names.
    """
    if aliases:
        names = entity.get_type_values(registry.name, matchable=True)
    else:
        names = entity.get("name", quiet=True)
    qids: list[str] = []
    seen: set[str] = set()
    for name in names:
        for qid in self._search_name(name, limit=limit):
            if qid not in seen:
                seen.add(qid)
                qids.append(qid)
    return qids

nomenklatura.wikidata.Item

A wikidata item (or entity).

Source code in nomenklatura/wikidata/model.py

class Item:
    """A wikidata item (or entity)."""

    def __init__(self, client: "WikidataClient", data: dict[str, Any]) -> None:
        self.client = client
        self.id: str = data.pop("id")
        self.modified: str | None = data.pop("modified", None)

        self.labels: set[LangText] = LangText.from_dict(data.pop("labels", {}))
        self.aliases: set[LangText] = LangText.from_dict(data.pop("aliases", {}))

        descriptions = LangText.from_dict(data.pop("descriptions", {}))
        self.description = LangText.pick(descriptions)

        self.claims: list[Claim] = []
        self.deprecated: list[Claim] = []
        claims: dict[str, list[dict[str, Any]]] = data.pop("claims", {})
        for prop, values in claims.items():
            for value in values:
                claim = Claim(client, value, prop)
                if claim.deprecated:
                    self.deprecated.append(claim)
                else:
                    self.claims.append(claim)

        # Merged pages handling:
        redirects = data.pop("redirects", {})
        self.redirect_id = redirects.get("to", None)
        if self.redirect_id is not None:
            self.id = self.redirect_id

        self.sitelinks: list[SiteLink] = []
        for data in data.pop("sitelinks", {}).values():
            self.sitelinks.append(SiteLink(self.id, data))

    @property
    def label(self) -> LangText | None:
        label = LangText.pick(self.labels)
        if label is not None:
            return label
        return LangText.pick(self.aliases)

    @property
    def sorted_labels(self) -> list[LangText]:
        return LangText.sorted(self.labels)

    @property
    def sorted_aliases(self) -> list[LangText]:
        return LangText.sorted(self.aliases)

    @property
    def wikilinks(self) -> list[SiteLink]:
        wikilinks = [s for s in self.sitelinks if s.is_wiki]
        # Skip commonswiki since it doesn't offer much more than wikidata as a wiki website.
        return [s for s in wikilinks if s.site != "commonswiki"]

    def is_instance(self, qid: str) -> bool:
        for claim in self.claims:
            if claim.property == "P31" and claim.qid == qid:
                return True
        return False

    def _types(self, path: list[str]) -> set[str]:
        qid = path[-1]
        types = set([qid])
        if len(path) > 6:
            return types

        item = self if qid == self.id else self.client.fetch_item(qid)
        if item is None:
            # A deleted P31/P279 ancestor shouldn't break type expansion:
            log.warning("Missing type ancestor item: %s (path: %r)", qid, path)
            return types
        for type_ in _type_props(item):
            if type_ not in path:
                types.update(self._types(path + [type_]))
        return types

    @property
    def types(self) -> set[str]:
        """Get all the `instance of` and `subclass of` types for an item."""
        return self._types([self.id])

    def __repr__(self) -> str:
        return f"<Item({self.id})>"

    def __hash__(self) -> int:
        return hash(self.id)

types property

Get all the instance of and subclass of types for an item.

nomenklatura.wikidata.Claim

Bases: Snak

One property statement on a Wikidata item — e.g. P569 (date of birth) on a person — including its qualifiers, references, and rank.

Source code in nomenklatura/wikidata/model.py

class Claim(Snak):
    """One property statement on a Wikidata item — e.g. `P569` (date of birth)
    on a person — including its qualifiers, references, and rank."""

    def __init__(
        self, client: "WikidataClient", data: dict[str, Any], prop: str
    ) -> None:
        self.id = data.pop("id")
        self.rank = data.pop("rank")
        super().__init__(client, data.pop("mainsnak"))
        self.qualifiers: dict[str, list[Snak]] = {}
        for prop, snaks in data.pop("qualifiers", {}).items():
            self.qualifiers[prop] = [Snak(client, s) for s in snaks]

        self.references = [Reference(client, r) for r in data.pop("references", [])]
        self.property = self.property or prop
        self.deprecated = bool(self.rank == "deprecated")

    def get_qualifier(self, prop: str) -> list[Snak]:
        return self.qualifiers.get(prop, [])

    def is_ended(self, reference_time: datetime | None = None) -> bool:
        """Whether the claim's validity period (P582) ended before
        `reference_time`, defaulting to the client's run reference time.

        A "no value" end time is Wikidata's way of asserting a claim is
        current, and an imprecise end date only counts once its whole span
        has elapsed — so a year-precision end in the current year is not
        yet ended."""
        if reference_time is None:
            reference_time = self.client.reference_time
        for snak in self.qualifiers.get("P582", []):
            if snak.snaktype == "novalue":
                continue
            if snak.snaktype == "somevalue":
                # Ended at an unknown date:
                return True
            text = snak.text.text
            if text is None:
                # An end is asserted, but the date didn't convert:
                return True
            try:
                if ended_before(text, reference_time):
                    return True
            except ValueError:
                return True
        return False

    def __repr__(self) -> str:
        return f"<Claim({self.qid}, {self.property}, {self.value_type})>"

    def __hash__(self) -> int:
        return hash((self.qid, self.property, self.id))

is_ended(reference_time=None)

Whether the claim's validity period (P582) ended before reference_time, defaulting to the client's run reference time.

A "no value" end time is Wikidata's way of asserting a claim is current, and an imprecise end date only counts once its whole span has elapsed — so a year-precision end in the current year is not yet ended.

Source code in nomenklatura/wikidata/model.py

def is_ended(self, reference_time: datetime | None = None) -> bool:
    """Whether the claim's validity period (P582) ended before
    `reference_time`, defaulting to the client's run reference time.

    A "no value" end time is Wikidata's way of asserting a claim is
    current, and an imprecise end date only counts once its whole span
    has elapsed — so a year-precision end in the current year is not
    yet ended."""
    if reference_time is None:
        reference_time = self.client.reference_time
    for snak in self.qualifiers.get("P582", []):
        if snak.snaktype == "novalue":
            continue
        if snak.snaktype == "somevalue":
            # Ended at an unknown date:
            return True
        text = snak.text.text
        if text is None:
            # An end is asserted, but the date didn't convert:
            return True
        try:
            if ended_before(text, reference_time):
                return True
        except ValueError:
            return True
    return False

nomenklatura.wikidata.LangText

A text value together with the language it is expressed in.

Wikidata labels and descriptions exist in many languages. Keeping the language tag with the string lets apply() write the value to an entity property with the language attached, and lets callers pick a preferred display language.

Source code in nomenklatura/wikidata/lang.py

class LangText:
    """A text value together with the language it is expressed in.

    Wikidata labels and descriptions exist in many languages. Keeping the
    language tag with the string lets `apply()` write the value to an entity
    property with the language attached, and lets callers pick a preferred
    display language."""

    __slots__ = ["lang", "original", "text"]

    def __init__(
        self,
        text: str | None,
        lang: str | None = None,
        original: str | None = None,
    ) -> None:
        if text is None or len(text.strip()) == 0:
            text = None
        if text is not None:
            text = remove_unsafe_chars(text)
        self.text = text
        self.lang: str | None = None
        if lang is not None:
            if lang == MULTI_LANG:
                self.lang = MULTI_LANG
            else:
                self.lang = registry.language.clean_text(lang)
        if lang is not None and self.lang is None:
            # Language is given, but it is not one supported by the FtM ecosystem:
            self.text = None
        self.original = original or self.text

    def apply(
        self,
        entity: StatementEntity,
        prop: str,
        clean: Callable[[str], str | None] | None = None,
    ) -> None:
        if self.text is None:
            return
        clean_text = self.text if clean is None else clean(self.text)
        if clean_text is None or clean_text.strip() == "":
            return
        lang = None if self.lang == MULTI_LANG else self.lang
        entity.add(prop, clean_text, lang=lang, original_value=self.original)

    def pack(self) -> dict[str, str | None]:
        data = {"t": self.text, "l": self.lang}
        if self.original is not None and self.original != self.text:
            data["o"] = self.original
        return data

    @classmethod
    def parse(cls, data: dict[str, str | None]) -> "LangText":
        return LangText(data["t"], data["l"], original=data.get("o"))

    @classmethod
    def pick(cls, texts: Iterable["LangText"]) -> Optional["LangText"]:
        for lang in PREFERRED_WD_LANGS:
            for lt in texts:
                if lt.lang == lang:
                    return lt
        for lt in texts:
            return lt
        return None

    @classmethod
    def sorted(cls, texts: Iterable["LangText"]) -> list["LangText"]:
        def sort_key(lt: LangText) -> Any:
            if lt.lang is None or lt.lang not in PREFERRED_WD_LANGS:
                index = len(PREFERRED_WD_LANGS)
            else:
                index = PREFERRED_WD_LANGS.index(lt.lang) + 1
            return (index, lt.text or "")

        return sorted(texts, key=sort_key)

    @classmethod
    def from_dict(cls, data: dict[str, list[dict[str, str]]]) -> set["LangText"]:
        langs: set[LangText] = set()
        for objs in data.values():
            if not isinstance(objs, list):
                objs = [objs]
            for obj in objs:
                value = obj["value"]
                if value is None:
                    continue
                lang = obj["language"]
                lt = LangText(value, lang, original=value)
                if lt.text is None:
                    continue
                langs.add(lt)
        return langs

    def __str__(self) -> str:
        if self.text is None:
            return ""
        return self.text

    def __hash__(self) -> int:
        return hash((self.text, self.lang, self.original))

    def __eq__(self, other: object) -> bool:
        return hash(self) == hash(other)

    def __repr__(self) -> str:
        return f"<LangText({self.text!r}, {self.lang!r}, {self.original!r})>"