openalex

zocrates.discovery.providers.openalex

OpenAlex scholarly literature search provider and identifier resolver.

DEFAULT_POLITE_EMAIL module-attribute

DEFAULT_POLITE_EMAIL = 'zocrates-dev@proton.me'

OPENALEX_SELECT_FIELDS module-attribute

OPENALEX_SELECT_FIELDS = 'id,doi,title,display_name,authorships,publication_year,primary_location,open_access,abstract_inverted_index,cited_by_count,type,ids,primary_topic,keywords,locations'

OPENALEX_WORKS_URL module-attribute

OPENALEX_WORKS_URL = 'https://api.openalex.org/works'

logger module-attribute

logger = logging.getLogger(__name__)

CandidatePaper dataclass

CandidatePaper(candidate_id, title, authors=list(), year=None, venue=None, doi=None, arxiv_id=None, openalex_id='', abstract=None, tldr=None, is_oa=False, oa_pdf_url=None, oa_pdf_urls=list(), citation_count=0, primary_topic=None, keywords=list(), in_library=False)

Represents a discovered scholarly paper candidate.

from_dict classmethod

from_dict(data)

Deserialize CandidatePaper from a dictionary.

Source code in zocrates/discovery/models.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "CandidatePaper":
    """Deserialize CandidatePaper from a dictionary."""
    citation_count_val = data.get("citation_count", 0)
    try:
        citation_count = int(citation_count_val) if citation_count_val is not None else 0
    except (ValueError, TypeError):
        citation_count = 0

    year_val = data.get("year")
    year: int | None = None
    if year_val is not None:
        try:
            year = int(year_val)
        except (ValueError, TypeError):
            year = None

    raw_authors = data.get("authors")
    if isinstance(raw_authors, list):
        authors = [str(a).strip() for a in raw_authors if a and str(a).strip()]
    elif isinstance(raw_authors, str) and raw_authors.strip():
        authors = [raw_authors.strip()]
    else:
        authors = []

    raw_keywords = data.get("keywords")
    if isinstance(raw_keywords, list):
        keywords = [str(k).strip() for k in raw_keywords if k and str(k).strip()]
    elif isinstance(raw_keywords, str) and raw_keywords.strip():
        keywords = [raw_keywords.strip()]
    else:
        keywords = []

    def _clean_opt_str(val: Any) -> str | None:
        """Clean and trim an optional string or scalar value.

        Args:
            val: Input value to convert to a stripped string.

        Returns:
            Stripped string, or None if empty or invalid.
        """
        if val is not None and isinstance(val, (str, int, float)):
            s = str(val).strip()
            return s if s else None
        return None

    raw_urls = data.get("oa_pdf_urls")
    oa_pdf_urls: list[str] = []
    if isinstance(raw_urls, list):
        oa_pdf_urls = [str(u).strip() for u in raw_urls if u and str(u).strip()]
    primary_pdf = _clean_opt_str(data.get("oa_pdf_url"))
    if primary_pdf and primary_pdf not in oa_pdf_urls:
        oa_pdf_urls.insert(0, primary_pdf)

    return cls(
        candidate_id=str(data.get("candidate_id") or ""),
        title=str(data.get("title") or "").strip(),
        authors=authors,
        year=year,
        venue=_clean_opt_str(data.get("venue")),
        doi=_clean_opt_str(data.get("doi")),
        arxiv_id=_clean_opt_str(data.get("arxiv_id")),
        openalex_id=str(data.get("openalex_id") or ""),
        abstract=_clean_opt_str(data.get("abstract")),
        tldr=_clean_opt_str(data.get("tldr")),
        is_oa=bool(data.get("is_oa", False)),
        oa_pdf_url=primary_pdf,
        oa_pdf_urls=oa_pdf_urls,
        citation_count=citation_count,
        primary_topic=_clean_opt_str(data.get("primary_topic")),
        keywords=keywords,
        in_library=bool(data.get("in_library", False)),
    )

to_dict

to_dict()

Serialize CandidatePaper to a plain dictionary.

Source code in zocrates/discovery/models.py

def to_dict(self) -> dict[str, Any]:
    """Serialize CandidatePaper to a plain dictionary."""
    return {
        "candidate_id": self.candidate_id,
        "title": self.title,
        "authors": list(self.authors),
        "year": self.year,
        "venue": self.venue,
        "doi": self.doi,
        "arxiv_id": self.arxiv_id,
        "openalex_id": self.openalex_id,
        "abstract": self.abstract,
        "tldr": self.tldr,
        "is_oa": self.is_oa,
        "oa_pdf_url": self.oa_pdf_url,
        "oa_pdf_urls": list(self.oa_pdf_urls),
        "citation_count": self.citation_count,
        "primary_topic": self.primary_topic,
        "keywords": list(self.keywords),
        "in_library": self.in_library,
    }

OpenAlexProvider

OpenAlexProvider(email=DEFAULT_POLITE_EMAIL, rate_limiter=None, timeout=15.0, api_key=None)

Bases: SearchProvider

Primary search provider backed by the OpenAlex REST API. Indexes 250M+ scholarly works, arXiv preprints, and open-access metadata.

Initialize the OpenAlex search provider.

Parameters:

Name	Type	Description	Default
email	str	Contact email for polite-pool API requests.	DEFAULT_POLITE_EMAIL
rate_limiter	TokenBucketRateLimiter | None	Token bucket rate limiter instance.	None
timeout	float	HTTP request timeout in seconds.	15.0
api_key	str | None	Optional OpenAlex premium API key.	None

Source code in zocrates/discovery/providers/openalex.py

def __init__(
    self,
    email: str = DEFAULT_POLITE_EMAIL,
    rate_limiter: TokenBucketRateLimiter | None = None,
    timeout: float = 15.0,
    api_key: str | None = None,
) -> None:
    """Initialize the OpenAlex search provider.

    Args:
        email: Contact email for polite-pool API requests.
        rate_limiter: Token bucket rate limiter instance.
        timeout: HTTP request timeout in seconds.
        api_key: Optional OpenAlex premium API key.
    """
    self.email = email
    self.rate_limiter = rate_limiter or TokenBucketRateLimiter(rate=10.0, capacity=10.0)
    self.timeout = timeout
    self.api_key = api_key.strip() if api_key and isinstance(api_key, str) else None

extract_arxiv_id_from_work staticmethod

extract_arxiv_id_from_work(work)

Extract canonical arXiv ID without version suffix from an OpenAlex work record.

Parameters:

Name	Type	Description	Default
work	dict[str, Any]	OpenAlex work dictionary.	required

Returns:

Type	Description
str | None	Canonical lowercase arXiv ID without 'v' suffix (e.g. '2305.12345'), or None.

Source code in zocrates/discovery/providers/openalex.py

@staticmethod
def extract_arxiv_id_from_work(work: dict[str, Any]) -> str | None:
    """
    Extract canonical arXiv ID without version suffix from an OpenAlex work record.

    Args:
        work: OpenAlex work dictionary.

    Returns:
        Canonical lowercase arXiv ID without 'v' suffix (e.g. '2305.12345'), or None.
    """
    # 1. Check locations landing_page_url and pdf_url
    locations = work.get("locations") or []
    primary_loc = work.get("primary_location") or {}
    candidate_locs: list[dict[str, Any]] = [primary_loc]
    if isinstance(locations, list):
        candidate_locs.extend(locations)

    for loc in candidate_locs:
        if not isinstance(loc, dict):
            continue
        for url_key in ("landing_page_url", "pdf_url"):
            val = loc.get(url_key)
            if val and isinstance(val, str):
                cand = normalize_arxiv_id(val)
                if cand:
                    return cand

    # 2. Check DataCite DOIs matching 10.48550/arxiv.(...)
    doi = work.get("doi")
    if doi and isinstance(doi, str):
        cand_doi = normalize_arxiv_id(doi)
        if cand_doi:
            return cand_doi

    # 3. Check ids dictionary if present
    ids = work.get("ids") or {}
    if isinstance(ids, dict):
        arxiv_val = ids.get("arxiv")
        if arxiv_val and isinstance(arxiv_val, str):
            cand_ids = normalize_arxiv_id(arxiv_val)
            if cand_ids:
                return cand_ids

    return None

reconstruct_abstract staticmethod

reconstruct_abstract(inverted_index)

Reconstruct plain-text abstract from an OpenAlex abstract_inverted_index dictionary.

Parameters:

Name	Type	Description	Default
inverted_index	dict[str, list[int]] | None	Mapping of words to their token position indices.	required

Returns:

Type	Description
str | None	Coherent, human-readable abstract string, or None if empty/invalid.

Source code in zocrates/discovery/providers/openalex.py

@staticmethod
def reconstruct_abstract(inverted_index: dict[str, list[int]] | None) -> str | None:
    """
    Reconstruct plain-text abstract from an OpenAlex abstract_inverted_index dictionary.

    Args:
        inverted_index: Mapping of words to their token position indices.

    Returns:
        Coherent, human-readable abstract string, or None if empty/invalid.
    """
    if not inverted_index or not isinstance(inverted_index, dict):
        return None

    positions: list[tuple[int, str]] = []
    for word, pos_list in inverted_index.items():
        if not isinstance(pos_list, list):
            continue
        for pos in pos_list:
            if isinstance(pos, int):
                positions.append((pos, word))

    if not positions:
        return None

    positions.sort(key=lambda x: x[0])
    return " ".join(word for _, word in positions).strip() or None

resolve_identifier async

resolve_identifier(identifier)

Resolve a paper by DOI, arXiv ID, or OpenAlex ID.

Parameters:

Name	Type	Description	Default
identifier	str	DOI, arXiv ID, or OpenAlex ID.	required

Returns:

Type	Description
CandidatePaper | None	Resolved CandidatePaper, or None if not found.

Source code in zocrates/discovery/providers/openalex.py

async def resolve_identifier(self, identifier: str) -> CandidatePaper | None:
    """
    Resolve a paper by DOI, arXiv ID, or OpenAlex ID.

    Args:
        identifier: DOI, arXiv ID, or OpenAlex ID.

    Returns:
        Resolved CandidatePaper, or None if not found.
    """
    ident = identifier.strip() if identifier else ""
    if not ident:
        return None

    # 1. OpenAlex ID (e.g. "https://openalex.org/W..." or "W123456789")
    m_oa = re.search(r"\b(W\d+)\b", ident, re.IGNORECASE)
    if m_oa:
        oa_id = m_oa.group(1).upper()
        url = f"{OPENALEX_WORKS_URL}/{oa_id}"
        params = {"select": OPENALEX_SELECT_FIELDS}
        data = await self._get_with_retry(url, params=params)
        if data and isinstance(data, dict) and "id" in data:
            return self._parse_work(data)

    # 2. DOI resolution
    clean_doi = normalize_doi(ident)
    if clean_doi and clean_doi.startswith("10.") and "/" in clean_doi:
        # If it's a DataCite arXiv DOI, extract the arXiv ID as fallback
        arxiv_from_doi = normalize_arxiv_id(clean_doi)

        # Query OpenAlex with canonical DOI filter
        params = {
            "filter": f"doi:https://doi.org/{clean_doi}",
            "select": OPENALEX_SELECT_FIELDS,
            "per-page": 1,
        }
        data = await self._get_with_retry(OPENALEX_WORKS_URL, params=params)
        if data and isinstance(data, dict):
            results = data.get("results") or []
            if isinstance(results, list) and results:
                return self._parse_work(results[0])

        # If DOI query didn't find it but it's an arXiv DOI, fall through to arXiv lookup
        if arxiv_from_doi:
            ident = arxiv_from_doi

    # 3. arXiv ID resolution
    clean_arxiv = normalize_arxiv_id(ident)
    if clean_arxiv:
        # First try DataCite DOI filter
        datacite_doi = f"https://doi.org/10.48550/arxiv.{clean_arxiv}"
        params = {
            "filter": f"doi:{datacite_doi}",
            "select": OPENALEX_SELECT_FIELDS,
            "per-page": 1,
        }
        data = await self._get_with_retry(OPENALEX_WORKS_URL, params=params)
        if data and isinstance(data, dict):
            results = data.get("results") or []
            if isinstance(results, list) and results:
                return self._parse_work(results[0])

        # Secondary search for landing page or arxiv ID text
        params_search = {
            "search": clean_arxiv,
            "select": OPENALEX_SELECT_FIELDS,
            "per-page": 5,
        }
        data_search = await self._get_with_retry(OPENALEX_WORKS_URL, params=params_search)
        if data_search and isinstance(data_search, dict):
            results = data_search.get("results") or []
            if isinstance(results, list):
                for w in results:
                    if isinstance(w, dict) and self.extract_arxiv_id_from_work(w) == clean_arxiv:
                        return self._parse_work(w)

    return None

search async

search(query, limit=10, year_from=None, year_to=None, is_oa_only=False)

Execute free-text or conceptual search against OpenAlex works.

Parameters:

Name	Type	Description	Default
query	str	User search query string.	required
limit	int	Maximum candidate works to return.	10
year_from	int | None	Optional minimum publication year.	None
year_to	int | None	Optional maximum publication year.	None
is_oa_only	bool	Restrict to open-access papers.	False

Returns:

Type	Description
list[CandidatePaper]	List of parsed CandidatePaper instances.

Source code in zocrates/discovery/providers/openalex.py

async def search(
    self,
    query: str,
    limit: int = 10,
    year_from: int | None = None,
    year_to: int | None = None,
    is_oa_only: bool = False,
) -> list[CandidatePaper]:
    """
    Execute free-text or conceptual search against OpenAlex works.

    Args:
        query: User search query string.
        limit: Maximum candidate works to return.
        year_from: Optional minimum publication year.
        year_to: Optional maximum publication year.
        is_oa_only: Restrict to open-access papers.

    Returns:
        List of parsed CandidatePaper instances.
    """
    q = query.strip() if query else ""
    if not q:
        return []

    # If query looks like an exact DOI or arXiv ID, delegate directly to resolve_identifier
    clean_doi = normalize_doi(q)
    if clean_doi and ("/" in clean_doi) and clean_doi.startswith("10."):
        single = await self.resolve_identifier(clean_doi)
        return [single] if single else []

    clean_arxiv = normalize_arxiv_id(q)
    if clean_arxiv and re.match(r"^(\d{4}\.\d{4,5}|[a-zA-Z\-]+(?:\.[a-zA-Z\-]+)?\/\d{7})$", clean_arxiv):
        single = await self.resolve_identifier(clean_arxiv)
        return [single] if single else []

    per_page = min(max(1, limit), 100)
    params: dict[str, Any] = {
        "search": q,
        "per-page": per_page,
        "select": OPENALEX_SELECT_FIELDS,
    }

    # Build filter expressions
    filters: list[str] = []
    if is_oa_only:
        filters.append("is_oa:true")

    if year_from is not None and year_to is not None:
        if year_from == year_to:
            filters.append(f"publication_year:{year_from}")
        else:
            filters.append(f"publication_year:{year_from}-{year_to}")
    elif year_from is not None:
        filters.append(f"publication_year:>{year_from - 1}")
    elif year_to is not None:
        filters.append(f"publication_year:<{year_to + 1}")

    if filters:
        params["filter"] = ",".join(filters)

    data = await self._get_with_retry(OPENALEX_WORKS_URL, params=params)
    if not data:
        return []

    results = data.get("results") or []
    candidates: list[CandidatePaper] = []
    if isinstance(results, list):
        for work in results:
            if isinstance(work, dict):
                candidates.append(self._parse_work(work))

    return candidates[:limit]

SearchProvider

Bases: ABC

Abstract base class for scholarly literature search providers.

resolve_identifier abstractmethod async

resolve_identifier(identifier)

Resolve a single scholarly paper by DOI, arXiv ID, or provider-specific ID.

Parameters:

Name	Type	Description	Default
identifier	str	DOI, arXiv ID, or provider ID string.	required

Returns:

Type	Description
CandidatePaper | None	CandidatePaper if resolved, or None if not found.

Source code in zocrates/discovery/providers/base.py

@abstractmethod
async def resolve_identifier(self, identifier: str) -> CandidatePaper | None:
    """
    Resolve a single scholarly paper by DOI, arXiv ID, or provider-specific ID.

    Args:
        identifier: DOI, arXiv ID, or provider ID string.

    Returns:
        CandidatePaper if resolved, or None if not found.
    """
    ...

search abstractmethod async

search(query, limit=10, year_from=None, year_to=None, is_oa_only=False)

Search for scholarly literature matching the query and filters.

Parameters:

Name	Type	Description	Default
query	str	Free-text search terms or conceptual query.	required
limit	int	Maximum number of candidate papers to return.	10
year_from	int | None	Optional minimum publication year (inclusive).	None
year_to	int | None	Optional maximum publication year (inclusive).	None
is_oa_only	bool	When True, restrict results to open-access works.	False

Returns:

Type	Description
list[CandidatePaper]	List of CandidatePaper instances.

Source code in zocrates/discovery/providers/base.py

@abstractmethod
async def search(
    self,
    query: str,
    limit: int = 10,
    year_from: int | None = None,
    year_to: int | None = None,
    is_oa_only: bool = False,
) -> list[CandidatePaper]:
    """
    Search for scholarly literature matching the query and filters.

    Args:
        query: Free-text search terms or conceptual query.
        limit: Maximum number of candidate papers to return.
        year_from: Optional minimum publication year (inclusive).
        year_to: Optional maximum publication year (inclusive).
        is_oa_only: When True, restrict results to open-access works.

    Returns:
        List of CandidatePaper instances.
    """
    ...

TokenBucketRateLimiter

TokenBucketRateLimiter(rate=10.0, capacity=10.0)

Token-bucket rate limiter for respectful, throttled API consumption. Thread/asyncio safe.

Parameters:

Name	Type	Description	Default
rate	float	Refill rate in tokens per second.	10.0
capacity	float	Maximum burst capacity in tokens.	10.0

Source code in zocrates/discovery/providers/base.py

def __init__(self, rate: float = 10.0, capacity: float = 10.0) -> None:
    """
    Args:
        rate: Refill rate in tokens per second.
        capacity: Maximum burst capacity in tokens.
    """
    self.rate = max(0.1, float(rate))
    self.capacity = max(1.0, float(capacity))
    self.tokens = self.capacity
    self.last_update = time.monotonic()
    self._lock = asyncio.Lock()

acquire async

acquire(tokens=1.0)

Acquire tokens from the bucket, sleeping if necessary until replenished.

Source code in zocrates/discovery/providers/base.py

async def acquire(self, tokens: float = 1.0) -> None:
    """Acquire `tokens` from the bucket, sleeping if necessary until replenished."""
    async with self._lock:
        now = time.monotonic()
        elapsed = now - self.last_update
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.last_update = now

        if self.tokens < tokens:
            needed = tokens - self.tokens
            wait_time = needed / self.rate
            await asyncio.sleep(wait_time)
            self.tokens = 0.0
            self.last_update = time.monotonic()
        else:
            self.tokens -= tokens

make_user_agent

make_user_agent(email=DEFAULT_POLITE_EMAIL)

Build a respectful User-Agent header string adhering to API etiquette.

Source code in zocrates/discovery/providers/base.py

def make_user_agent(email: str = DEFAULT_POLITE_EMAIL) -> str:
    """Build a respectful User-Agent header string adhering to API etiquette."""
    polite_email = email.strip() if email and email.strip() else DEFAULT_POLITE_EMAIL
    return f"Zocrates/0.2.0 (mailto:{polite_email})"

normalize_arxiv_id

normalize_arxiv_id(arxiv_id)

Normalize an arXiv identifier by stripping URL prefixes, 'arxiv:' tags, and version suffixes.

Source code in zocrates/rag/zotero_db.py

def normalize_arxiv_id(arxiv_id: str | None) -> str | None:
    """Normalize an arXiv identifier by stripping URL prefixes, 'arxiv:' tags, and version suffixes."""
    if not arxiv_id:
        return None
    s = arxiv_id.strip()
    url_m = re.search(
        r"(?i)arxiv\.org\/(?:abs|pdf)\/([a-zA-Z\-]+(?:\.[a-zA-Z\-]+)?\/\d{7}|\d{4}\.\d{4,5}(?:v\d+)?)",
        s,
    )
    if url_m:
        s = url_m.group(1)
    else:
        doi_m = re.search(
            r"(?i)10\.48550\/arxiv\.([a-zA-Z\-]+(?:\.[a-zA-Z\-]+)?\/\d{7}|\d{4}\.\d{4,5}(?:v\d+)?)",
            s,
        )
        if doi_m:
            s = doi_m.group(1)
        else:
            if s.lower().startswith("arxiv:"):
                s = s[6:].strip()

    s = re.sub(r"(?i)v\d+$", "", s).strip()
    s = re.sub(r"(?i)\.pdf$", "", s).strip().rstrip(".")
    s = s.lower()
    if ARXIV_ID_PATTERN.match(s):
        return s
    return None

normalize_doi

normalize_doi(doi)

Normalize DOI by stripping standard URI and resolver prefixes and lowercasing.

Source code in zocrates/rag/zotero_db.py

def normalize_doi(doi: str | None) -> str | None:
    """Normalize DOI by stripping standard URI and resolver prefixes and lowercasing."""
    if not doi:
        return None
    d = doi.strip()
    prefixes = (
        "https://doi.org/",
        "http://doi.org/",
        "https://dx.doi.org/",
        "http://dx.doi.org/",
        "doi.org/",
        "doi:",
    )
    for prefix in prefixes:
        if d.lower().startswith(prefix):
            d = d[len(prefix) :].strip()
    d = d.strip().rstrip("/")
    return d.lower() if d else None