Skip to content

Angelo

openalex

natashahirt/angelo

`zettelkasten.openalex`¶

zettelkasten.openalex ¶

Lightweight OpenAlex client for citation-graph expansion.

OpenAlex (https://openalex.org) is a free, open scholarly catalog with no API key required. We use it to resolve a work by DOI and to walk its citation graph: backward (the works it references) and forward (the works that cite it).

Set OPENALEX_MAILTO to join the polite pool (faster, more reliable). No authentication is needed otherwise.

simplify ¶

simplify(work: dict[str, Any]) -> dict[str, Any]

Reduce a raw OpenAlex work object to the fields we store.

Source code in zettelkasten/openalex.py

def simplify(work: dict[str, Any]) -> dict[str, Any]:
    """Reduce a raw OpenAlex work object to the fields we store."""
    authors = [
        a.get("author", {}).get("display_name", "")
        for a in work.get("authorships", [])
        if a.get("author", {}).get("display_name")
    ]
    doi = (work.get("doi") or "").replace("https://doi.org/", "")
    raw_type = work.get("type", "")
    return {
        "openalex_id": _short_id(work.get("id", "")),
        "title": work.get("display_name", "") or "",
        "authors": authors,
        "year": work.get("publication_year"),
        "doi": doi,
        "doc_type": _TYPE_MAP.get(raw_type, raw_type or ""),
        "cited_by_count": work.get("cited_by_count", 0),
        # Per-year citation history (standard OpenAlex field, no extra request).
        # Each item is {"year": int, "cited_by_count": int}; downstream trend/recency
        # signals consume it. Empty list when absent so callers can store/skip cleanly.
        "counts_by_year": list(work.get("counts_by_year") or []),
        # Plain-text abstract reconstructed from the inverted index, and the
        # venue name. Both degrade to "" when absent so callers can store/skip.
        "abstract": _reconstruct_abstract(work.get("abstract_inverted_index")),
        "venue": _extract_venue(work),
    }

get_work_by_doi ¶

get_work_by_doi(doi: str) -> dict[str, Any] | None

Resolve a single work by DOI. Returns the raw OpenAlex work or None.

Source code in zettelkasten/openalex.py

def get_work_by_doi(doi: str) -> dict[str, Any] | None:
    """Resolve a single work by DOI. Returns the raw OpenAlex work or None."""
    doi = (doi or "").strip().replace("https://doi.org/", "").lower()
    if not doi:
        return None
    resp = httpx.get(f"{_BASE}/doi:{doi}", params=_params({"select": _SELECT}), timeout=_TIMEOUT)
    if resp.status_code != 200:
        return None
    return resp.json()

extract_arxiv_id ¶

extract_arxiv_id(*texts: str) -> str

Pull a bare modern arXiv id (e.g. 1706.03762) from free text, or ''.

Scans each argument in order (a dedicated arxiv field, then a messy venue like "NeurIPS 2017; arXiv:1706.03762") and returns the first match without its version suffix.

Source code in zettelkasten/openalex.py

def extract_arxiv_id(*texts: str) -> str:
    """Pull a bare modern arXiv id (e.g. ``1706.03762``) from free text, or ''.

    Scans each argument in order (a dedicated ``arxiv`` field, then a messy
    ``venue`` like ``"NeurIPS 2017; arXiv:1706.03762"``) and returns the first
    match without its version suffix.
    """
    for t in texts:
        if not t:
            continue
        m = _ARXIV_RE.search(str(t))
        if m:
            return m.group(1)
    return ""

get_work_by_arxiv ¶

get_work_by_arxiv(arxiv_id: str) -> dict[str, Any] | None

Resolve a work by arXiv id via the DataCite DOI arXiv mints (10.48550/arXiv.<id>). Returns the raw OpenAlex work or None.

This is an exact-identifier path (no fuzzy matching), so a hit is safe to trust. Many older preprints are not linked under this DOI in OpenAlex yet, in which case this returns None and the caller falls back to a strict title match.

Source code in zettelkasten/openalex.py

def get_work_by_arxiv(arxiv_id: str) -> dict[str, Any] | None:
    """Resolve a work by arXiv id via the DataCite DOI arXiv mints
    (``10.48550/arXiv.<id>``). Returns the raw OpenAlex work or None.

    This is an *exact-identifier* path (no fuzzy matching), so a hit is safe to
    trust. Many older preprints are not linked under this DOI in OpenAlex yet, in
    which case this returns None and the caller falls back to a strict title
    match.
    """
    aid = (arxiv_id or "").strip().lower().removeprefix("arxiv:")
    if not aid:
        return None
    return get_work_by_doi(f"10.48550/arxiv.{aid}")

find_work_by_title ¶

find_work_by_title(title: str, *, year: int | None = None, first_author: str = '', year_tol: int = 1) -> dict[str, Any] | None

STRICT title resolution for DOI-less works. Returns the raw OpenAlex work only when a candidate's title matches exactly (normalized) AND it agrees on year (within year_tol) AND, when first_author is given, on the first author's surname. Returns None on any near-miss.

The strictness is deliberate: OpenAlex title search readily returns unrelated same-title papers (e.g. a recent paper reusing a famous title), so a loose "top hit" match would assign the wrong paper's citation count. Rejecting a near-miss (leaving the work unenriched / influence "unknown") is the safe failure mode.

Source code in zettelkasten/openalex.py

def find_work_by_title(
    title: str,
    *,
    year: int | None = None,
    first_author: str = "",
    year_tol: int = 1,
) -> dict[str, Any] | None:
    """STRICT title resolution for DOI-less works. Returns the raw OpenAlex work
    only when a candidate's title matches *exactly* (normalized) AND it agrees on
    year (within ``year_tol``) AND, when ``first_author`` is given, on the first
    author's surname. Returns None on any near-miss.

    The strictness is deliberate: OpenAlex title search readily returns unrelated
    same-title papers (e.g. a recent paper reusing a famous title), so a loose
    "top hit" match would assign the wrong paper's citation count. Rejecting a
    near-miss (leaving the work unenriched / influence "unknown") is the safe
    failure mode.
    """
    title = (title or "").strip()
    if not title:
        return None
    resp = httpx.get(
        _BASE,
        params=_params({
            "filter": f"title.search:{title}",
            "select": _SELECT,
            "per-page": 25,
            "sort": "cited_by_count:desc",
        }),
        timeout=_TIMEOUT,
    )
    if resp.status_code != 200:
        return None
    want_title = _norm_title(title)
    want_surname = _surname(first_author)
    for w in resp.json().get("results", []):
        if _norm_title(w.get("display_name", "")) != want_title:
            continue
        wy = w.get("publication_year")
        if year is not None and wy is not None and abs(int(wy) - int(year)) > year_tol:
            continue
        if want_surname:
            surnames = {
                _surname(a.get("author", {}).get("display_name", ""))
                for a in w.get("authorships", [])
            }
            if want_surname not in surnames:
                continue
        return w
    return None

get_citing_works ¶

get_citing_works(openalex_id: str, limit: int = 25) -> list[dict[str, Any]]

Forward edges: works that cite the given work (most-cited first).

Source code in zettelkasten/openalex.py

def get_citing_works(openalex_id: str, limit: int = 25) -> list[dict[str, Any]]:
    """Forward edges: works that cite the given work (most-cited first)."""
    sid = _short_id(openalex_id)
    if not sid:
        return []
    resp = httpx.get(
        _BASE,
        params=_params({
            "filter": f"cites:{sid}",
            "select": _SELECT,
            "per-page": max(1, min(limit, 200)),
            "sort": "cited_by_count:desc",
        }),
        timeout=_TIMEOUT,
    )
    if resp.status_code != 200:
        return []
    return [simplify(w) for w in resp.json().get("results", [])]

get_referenced_works ¶

get_referenced_works(work: dict[str, Any], limit: int = 25) -> list[dict[str, Any]]

Backward edges: works referenced by the given work.

work is a raw OpenAlex object (from get_work_by_doi). The referenced ids are batch-resolved to metadata in a single request.

Source code in zettelkasten/openalex.py

def get_referenced_works(work: dict[str, Any], limit: int = 25) -> list[dict[str, Any]]:
    """Backward edges: works referenced by the given work.

    ``work`` is a raw OpenAlex object (from get_work_by_doi). The referenced
    ids are batch-resolved to metadata in a single request.
    """
    ref_ids = [_short_id(r) for r in work.get("referenced_works", []) if r]
    ref_ids = ref_ids[: max(1, min(limit, 100))]
    if not ref_ids:
        return []
    resp = httpx.get(
        _BASE,
        params=_params({
            "filter": "openalex:" + "|".join(ref_ids),
            "select": _SELECT,
            "per-page": len(ref_ids),
        }),
        timeout=_TIMEOUT,
    )
    if resp.status_code != 200:
        return []
    return [simplify(w) for w in resp.json().get("results", [])]