Demonstrate using the Snapquery package to run a query that takes in some parameters

snapquery_demo.py

from typing import Any

from snapquery.snapquery_core import QueryName, NamedQueryManager, QueryBundle

#: See documentation at https://snapquery.bitplan.com/docs
API_ENDPOINT_FMT = "https://snapquery.bitplan.com/api/query/{domain}/{namespace}/{name}"

type Result = dict[str, Any]
type Results = list[Result]


def from_programmatic_api(
    name: str,
    namespace: str,
    domain: str,
    *,
    endpoint_name: str = "wikidata",
    params: dict[str, Any] | None = None,
    limit: int | None = None,
) -> Results:
    if params is not None and {"q"} != set(params):
        raise NotImplementedError("can only support a single parameter, `q`")

    query_manager = NamedQueryManager.from_samples()
    query_name: QueryName = QueryName.from_query_id(query_id=name, namespace=namespace, domain=domain)
    query_bundle: QueryBundle = query_manager.get_query(
        query_name=query_name,
        endpoint_name=endpoint_name,
        limit=limit,
    )
    rv = query_bundle.sparql.queryAsListOfDicts(query_bundle.query.query, param_dict=params)
    # would be nice to extend get_lod to 
    # rv = query_bundle.get_lod(param_dict=params)
    return rv


def main():
    # TODO make sure you load up the queries into the local database
    # snapquery --import /Users/cthoyt/dev/snapquery/snapquery/samples/scholia.json

    payload_1 = dict(
        namespace="snapquery-examples",
        name="cats",
        domain="wikidata.org",
        limit=5,
    )

    # see https://snapquery.bitplan.com/query/scholia.toolforge.org/named_queries/author_list-of-publications
    payload_2 = dict(
        name="author_list-of-publications",
        domain="scholia.toolforge.org",
        namespace="named_queries",
        params={"q": "Q47475003"},
        limit=5,
    )
    for payload in [payload_1, payload_2]:
        print(payload['name'])
        records = from_programmatic_api(**payload)
        for record in records:
            print(record)
        print()


if __name__ == '__main__':
    main()