"""Core module providing a Python client for querying the NAKB Solr API."""
import pandas as pd
import requests
BASE_URL = "https://nakb.org/node/solr/nakb/select"
[docs]
class NAKBClient:
def __init__(self, base_url: str = BASE_URL):
"""
Client for interacting with the NAKB Solr search API.
Parameters
----------
base_url : str, optional
Base endpoint for the NAKB Solr server.
Defaults to ``https://nakb.org/node/solr/nakb/select``.
"""
self.base_url = base_url
[docs]
def query_raw(
self,
q: str = "*:*",
fl: str | None = None,
fq: list[str] | str | None = None,
rows: int = 10,
as_dataframe: bool = True,
):
"""
Send a raw Solr query to NAKB.
This method sends a direct query to the Solr backend, exposing all
parameters such as ``q`` (query string), ``fl`` (field list), and
``fq`` (filter queries). It is a low-level interface; for a more
ergonomic wrapper, use :meth:`query`.
Parameters
----------
q : str, optional
Main Solr query string. Defaults to ``"*:*"`` (match all entries).
fl : str or None, optional
Comma-separated list of fields to return.
If ``None``, Solr returns all fields.
fq : list of str, str, or None, optional
Filter query/queries passed directly to Solr.
May be a single string or a list of strings.
If ``None``, Solr applies no filters.
rows : int, optional
Maximum number of rows to return. Default is ``10``.
as_dataframe : bool, optional
If ``True``, results are returned as a :class:`pandas.DataFrame`.
Otherwise, returns the raw list of document dictionaries.
Returns
-------
pandas.DataFrame or list[dict]
Query results, either as a DataFrame (default) or as a list of
dicts.
Raises
------
requests.HTTPError
If the Solr request returns a non-successful status code.
"""
params = {
"wt": "json",
"q": q,
"rows": rows,
}
if fl:
params["fl"] = fl
if fq:
params["fq"] = fq
r = requests.get(self.base_url, params=params)
r.raise_for_status()
docs = r.json()["response"]["docs"]
return pd.DataFrame(docs) if as_dataframe else docs
[docs]
def query(self, fields: list[str] | None = None, rows: int = 10, **kwargs):
"""
High-level query interface for NAKB.
This method provides a Pythonic interface over Solr's query system.
It automatically builds Solr filter queries (``fq``) based on keyword
arguments, supporting operators such as:
- ``field__lt=value`` → ``field:[* TO value]``
- ``field__gt=value`` → ``field:[value TO *]``
- ``field__range=(a, b)`` → ``field:[a TO b]``
- ``field__contains=substr`` → ``field:*substr*``
- ``field="value"`` → exact match
- ``field=[v1, v2, ...]`` → OR query (``field:( "v1" OR "v2" )``)
Parameters
----------
fields : list[str] or None, optional
List of fields to return. If ``None``, Solr returns all fields.
rows : int, optional
Number of rows to retrieve. Default is ``10``.
**kwargs
Field filters encoded using either direct matches or extended
operators.
Returns
-------
pandas.DataFrame
A DataFrame containing the matching rows.
Raises
------
ValueError
If an unsupported operator is used in a keyword argument.
"""
q = "*:*"
if fields:
fl = ",".join(fields)
else:
fl = None
fq = []
for key, value in kwargs.items():
if "__" in key:
field, op = key.split("__", 1)
if op == "lt": # less than
fq.append(f"{field}:[* TO {value}]")
elif op == "gt": # greater than
fq.append(f"{field}:[{value} TO *]")
elif op == "range":
a, b = value
fq.append(f"{field}:[{a} TO {b}]")
elif op == "contains":
fq.append(f"{field}:*{value}*")
else:
raise ValueError(f"Operator not supported: {op}")
else:
# string → direct match
if isinstance(value, str):
fq.append(f'{key}:"{value}"')
# list → OR
elif isinstance(value, (list, tuple)):
ors = " OR ".join(f'"{v}"' for v in value)
fq.append(f"{key}:({ors})")
# int → direct match
else:
fq.append(f"{key}:{value}")
return self.query_raw(
q=q,
fq=fq,
fl=fl,
rows=rows,
as_dataframe=True,
)
[docs]
def get_fields(self):
"""
Retrieve the list of available fields in NAKB.
This method performs a minimal query and returns the column names
of the resulting DataFrame, which correspond to metadata fields
stored in the Solr index.
Returns
-------
list[str]
List of available field names in NAKB.
"""
df = self.query(pdbid="1A1H")
return df.columns.tolist()