class_docstr: A cache for requests against Wikidata's SPARQL endpoint.
class_docstr_tok: ["A","cache","for","requests","against","Wikidata","'s","SPARQL","endpoint","."]
docstr: Batched SPARQL query execution for the given IDS.
  
  :param sparql:
      the SPARQL query with a placeholder `ids`
  :param wikidata_ids:
      the Wikidata IDs
  :param batch_size:
      the batch size, i.e., maximum number of IDs per query
  
  :return:
      an iterable over JSON results, where the keys correspond to query variables,
      and the values to the corresponding binding
docstr_tok: ["Batched","SPARQL","query","execution","for","the","given","IDS",".",":","param","sparql",":","the","SPARQL","query","with","a","placeholder","`","ids","`",":","param","wikidata_ids",":","the","Wikidata","IDs",":","param","batch_size",":","the","batch","size",",","i.e.",",","maximum","number","of","IDs","per","query",":","return",":","an","iterable","over","JSON","results",",","where","the","keys","correspond","to","query","variables",",","and","the","values","to","the","corresponding","binding"]
code: def query(
      cls,
      sparql: Union[str, Callable[..., str]],
      wikidata_ids: Sequence[str],
      batch_size: int = 256,
  ) -> Iterable[Mapping[str, Any]]:
      """
      Batched SPARQL query execution for the given IDS.
  
      :param sparql:
          the SPARQL query with a placeholder `ids`
      :param wikidata_ids:
          the Wikidata IDs
      :param batch_size:
          the batch size, i.e., maximum number of IDs per query
  
      :return:
          an iterable over JSON results, where the keys correspond to query variables,
          and the values to the corresponding binding
      """
      if not wikidata_ids:
          return {}
  
      if len(wikidata_ids) > batch_size:
          # break into smaller requests
          return chain.from_iterable(
              cls.query(
                  sparql=sparql,
                  wikidata_ids=id_batch,
                  batch_size=batch_size,
              )
              for id_batch in more_itertools.chunked(
                  wikidata_ids, batch_size
              )
          )
  
      if isinstance(sparql, str):
          sparql = sparql.format
      sparql = sparql(ids=" ".join(f"wd:{i}" for i in wikidata_ids))
      logger.debug("running query: %s", sparql)
      res = requests.get(
          cls.WIKIDATA_ENDPOINT,
          params={"query": sparql, "format": "json"},
          headers=cls.HEADERS,
      )
      res.raise_for_status()
      bindings = res.json()["results"]["bindings"]
      logger.debug(f"Retrieved {len(bindings)} bindings")
      return bindings
  
imports: from __future__ import annotations
  import functools
  import logging
  import pathlib
  import re
  import subprocess
  from abc import ABC, abstractmethod
  from itertools import chain
  from textwrap import dedent
  from typing import Any, Callable, Collection, Dict, Iterable, List, Literal, Mapping, Optional, Sequence, Union, cast
  import more_itertools
  import requests
  import torch
  from tqdm.auto import tqdm
  from ..constants import PYKEEN_MODULE
  from ..typing import OneOrSequence
  from ..utils import nested_get, rate_limited, upgrade_to_sequence
  from ..version import get_version
label_desc: Use image node_id 3 for calling the WikidataCache obj's underlying member method code with example usage: obj.query(cls, sparql, wikidata_ids, batch_size) and returns: bindings, dict, chain