API Reference

Auto-generated documentation from the source code.

SQLite database layer for the OpenAlex Research Manager.

Database

Thin wrapper around an SQLite connection for the library.

Source code in openalex_pygui/db.py

class Database:
    """Thin wrapper around an SQLite connection for the library."""

    def __init__(self, path: Path | str | None = None):
        self.path = Path(path) if path else _DEFAULT_DB
        self.conn = sqlite3.connect(str(self.path), check_same_thread=False)
        self.conn.execute("PRAGMA journal_mode=WAL")
        self.conn.execute("PRAGMA foreign_keys=ON")
        self.conn.executescript(_SCHEMA)
        self._run_migrations()

    def _run_migrations(self):
        for sql in _MIGRATIONS:
            try:
                self.conn.execute(sql)
            except sqlite3.OperationalError:
                pass

    # ── Works ──────────────────────────────────────────────────────────

    def add_work(
        self,
        openalex_id: str,
        title: str,
        *,
        doi: str | None = None,
        publication_year: int | None = None,
        work_type: str | None = None,
        cited_by_count: int = 0,
        abstract: str | None = None,
        journal: str | None = None,
        authors: list[dict] | None = None,
        keywords: list[str] | None = None,
        relationships: list[dict] | None = None,
    ) -> None:
        with self.conn:
            self.conn.execute(
                """INSERT OR IGNORE INTO works
                   (id, doi, title, publication_year, type, cited_by_count, abstract, journal)
                   VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
                (openalex_id, doi, title, publication_year, work_type,
                 cited_by_count, abstract, journal),
            )

            if authors:
                for author in authors:
                    aid = author.get("id")
                    name = author.get("name", "")
                    orcid = author.get("orcid")
                    pos = author.get("position")
                    self.conn.execute(
                        "INSERT OR IGNORE INTO authors (id, name, orcid) VALUES (?, ?, ?)",
                        (aid, name, orcid),
                    )
                    self.conn.execute(
                        "INSERT OR IGNORE INTO work_authors (work_id, author_id, position) VALUES (?, ?, ?)",
                        (openalex_id, aid, pos),
                    )

            if keywords:
                for kw in keywords:
                    self.conn.execute(
                        "INSERT OR IGNORE INTO work_keywords (work_id, keyword) VALUES (?, ?)",
                        (openalex_id, kw),
                    )

            if relationships:
                for rel in relationships:
                    self.conn.execute(
                        "INSERT OR IGNORE INTO work_relationships (work_id, related_id, relationship) VALUES (?, ?, ?)",
                        (openalex_id, rel["id"], rel["type"]),
                    )

    def remove_work(self, openalex_id: str) -> None:
        with self.conn:
            self.conn.execute("DELETE FROM works WHERE id = ?", (openalex_id,))

    def remove_works(self, openalex_ids: list[str]) -> None:
        with self.conn:
            self.conn.executemany(
                "DELETE FROM works WHERE id = ?", [(wid,) for wid in openalex_ids]
            )

    def get_work(self, openalex_id: str) -> dict | None:
        row = self.conn.execute("SELECT * FROM works WHERE id = ?", (openalex_id,)).fetchone()
        if not row:
            return None
        cols = [d[0] for d in self.conn.execute("SELECT * FROM works LIMIT 0").description]
        return dict(zip(cols, row))

    def list_works(
        self,
        *,
        search: str | None = None,
        sort_by: str = "date_added",
        keyword: str | None = None,
        tag: str | None = None,
    ) -> list[dict]:
        order_cols = {"title", "publication_year", "cited_by_count", "date_added"}
        if sort_by not in order_cols:
            sort_by = "date_added"

        sql = f"SELECT DISTINCT w.* FROM works w"
        params: list = []
        joins: list[str] = []
        conditions: list[str] = []

        if keyword:
            joins.append("JOIN work_keywords wk ON w.id = wk.work_id")
            conditions.append("wk.keyword = ?")
            params.append(keyword)

        if tag:
            joins.append("JOIN work_tags wt ON w.id = wt.work_id")
            conditions.append("wt.tag = ?")
            params.append(tag)

        if search:
            conditions.append("(w.title LIKE ? OR w.abstract LIKE ?)")
            params.extend((f"%{search}%", f"%{search}%"))

        if joins:
            sql += " " + " ".join(joins)
        if conditions:
            sql += " WHERE " + " AND ".join(conditions)
        sql += f" ORDER BY w.{sort_by} DESC"

        rows = self.conn.execute(sql, params).fetchall()
        cols = [d[0] for d in self.conn.execute("SELECT * FROM works LIMIT 0").description]
        return [dict(zip(cols, r)) for r in rows]

    def work_exists(self, openalex_id: str) -> bool:
        return (
            self.conn.execute("SELECT 1 FROM works WHERE id = ?", (openalex_id,)).fetchone()
            is not None
        )

    def set_bibtex(self, openalex_id: str, bibtex: str) -> None:
        with self.conn:
            self.conn.execute(
                "UPDATE works SET bibtex = ? WHERE id = ?", (bibtex, openalex_id)
            )

    def set_notes(self, openalex_id: str, notes: str) -> None:
        with self.conn:
            self.conn.execute(
                "UPDATE works SET notes = ? WHERE id = ?", (notes, openalex_id)
            )

    def set_abstract(self, openalex_id: str, abstract: str) -> None:
        with self.conn:
            self.conn.execute(
                "UPDATE works SET abstract = ? WHERE id = ?", (abstract, openalex_id)
            )

    # ── Authors ────────────────────────────────────────────────────────

    def get_authors_for_work(self, openalex_id: str) -> list[dict]:
        rows = self.conn.execute(
            """SELECT a.id, a.name, a.orcid, wa.position
               FROM authors a JOIN work_authors wa ON a.id = wa.author_id
               WHERE wa.work_id = ?
               ORDER BY wa.position""",
            (openalex_id,),
        ).fetchall()
        cols = ["id", "name", "orcid", "position"]
        return [dict(zip(cols, r)) for r in rows]

    # ── Relationships ──────────────────────────────────────────────────

    def get_relationships(self, openalex_id: str, rel_type: str | None = None) -> list[dict]:
        if rel_type:
            rows = self.conn.execute(
                "SELECT related_id, relationship FROM work_relationships WHERE work_id = ? AND relationship = ?",
                (openalex_id, rel_type),
            ).fetchall()
        else:
            rows = self.conn.execute(
                "SELECT related_id, relationship FROM work_relationships WHERE work_id = ?",
                (openalex_id,),
            ).fetchall()
        return [{"id": r[0], "type": r[1]} for r in rows]

    # ── Keywords ───────────────────────────────────────────────────────

    def get_keywords_for_work(self, openalex_id: str) -> list[str]:
        rows = self.conn.execute(
            "SELECT keyword FROM work_keywords WHERE work_id = ?", (openalex_id,)
        ).fetchall()
        return [r[0] for r in rows]

    def list_all_keywords(self) -> list[str]:
        rows = self.conn.execute(
            "SELECT DISTINCT keyword FROM work_keywords ORDER BY keyword"
        ).fetchall()
        return [r[0] for r in rows]

    # ── Tags ───────────────────────────────────────────────────────────

    def get_tags_for_work(self, openalex_id: str) -> list[str]:
        rows = self.conn.execute(
            "SELECT tag FROM work_tags WHERE work_id = ?", (openalex_id,)
        ).fetchall()
        return [r[0] for r in rows]

    def set_tags(self, openalex_id: str, tags: list[str]) -> None:
        with self.conn:
            self.conn.execute("DELETE FROM work_tags WHERE work_id = ?", (openalex_id,))
            for tag in tags:
                tag = tag.strip()
                if tag:
                    self.conn.execute(
                        "INSERT OR IGNORE INTO work_tags (work_id, tag) VALUES (?, ?)",
                        (openalex_id, tag),
                    )

    def list_all_tags(self) -> list[str]:
        rows = self.conn.execute("SELECT DISTINCT tag FROM work_tags ORDER BY tag").fetchall()
        return [r[0] for r in rows]

    # ── Settings ───────────────────────────────────────────────────────

    def get_setting(self, key: str, default: str | None = None) -> str | None:
        row = self.conn.execute("SELECT value FROM settings WHERE key = ?", (key,)).fetchone()
        return row[0] if row else default

    def set_setting(self, key: str, value: str) -> None:
        with self.conn:
            self.conn.execute(
                "INSERT OR REPLACE INTO settings (key, value) VALUES (?, ?)", (key, value)
            )

    # ── Export ─────────────────────────────────────────────────────────

    def export_bibtex(self, *, ids: list[str] | None = None) -> str:
        """Return BibTeX string. If *ids* given, export only those; otherwise all."""
        if ids:
            placeholders = ",".join("?" for _ in ids)
            rows = self.conn.execute(
                f"SELECT bibtex FROM works WHERE id IN ({placeholders}) AND bibtex IS NOT NULL",
                ids,
            ).fetchall()
        else:
            rows = self.conn.execute(
                "SELECT bibtex FROM works WHERE bibtex IS NOT NULL ORDER BY date_added DESC"
            ).fetchall()
        return "\n\n".join(r[0] for r in rows)

    # ── Lifecycle ──────────────────────────────────────────────────────

    def close(self) -> None:
        self.conn.close()

export_bibtex(*, ids=None)

Return BibTeX string. If ids given, export only those; otherwise all.

Source code in openalex_pygui/db.py

def export_bibtex(self, *, ids: list[str] | None = None) -> str:
    """Return BibTeX string. If *ids* given, export only those; otherwise all."""
    if ids:
        placeholders = ",".join("?" for _ in ids)
        rows = self.conn.execute(
            f"SELECT bibtex FROM works WHERE id IN ({placeholders}) AND bibtex IS NOT NULL",
            ids,
        ).fetchall()
    else:
        rows = self.conn.execute(
            "SELECT bibtex FROM works WHERE bibtex IS NOT NULL ORDER BY date_added DESC"
        ).fetchall()
    return "\n\n".join(r[0] for r in rows)

API layer: OpenAlex search and BibTeX retrieval.

OpenAlexSearcher

Searches OpenAlex and converts results to plain dicts ready for the DB.

Source code in openalex_pygui/api.py

class OpenAlexSearcher:
    """Searches OpenAlex and converts results to plain dicts ready for the DB."""

    def __init__(self, api_key: str | None = None, email: str | None = None):
        if api_key:
            config.api_key = api_key
        self._email = email

    def search(
        self,
        query: str,
        *,
        limit: int = 25,
        sort: str = "relevance_score:desc",
        semantic: bool = False,
        scope: str = "default",
    ) -> list[dict]:
        ws = WorksSync()
        if semantic:
            ws = ws.similar(query)
            if sort != "relevance_score:desc":
                q = ws.sort(**_parse_sort(sort))
                ws._params = q.params
        elif scope == "title":
            q = ws.search_filter(title=query).sort(**_parse_sort(sort))
            ws._params = q.params
        elif scope == "fulltext":
            q = ws.search_filter(fulltext=query).sort(**_parse_sort(sort))
            ws._params = q.params
        else:
            q = ws.search(query).sort(**_parse_sort(sort))
            ws._params = q.params
        results: list[Work] = ws.get(per_page=limit)
        return [work_to_dict(w) for w in results]

    @retry(stop=stop_after_attempt(3), wait=wait_exponential_jitter(initial=0.5, max=8, jitter=1))
    def _fetch_single_work(self, openalex_id: str) -> dict:
        w = WorksSync().get_by_id(openalex_id)
        return work_to_dict(w)

    def fetch_work_by_id(self, openalex_id: str) -> dict | None:
        try:
            return self._fetch_single_work(openalex_id)
        except NotFoundError:
            return None
        except Exception:
            return None

    def fetch_bibtex(self, doi: str) -> str | None:
        headers: dict[str, str] = {"Accept": "application/x-bibtex"}
        if self._email:
            headers["User-Agent"] = (
                f"OpenAlex-PyGUI/0.1 (mailto:{self._email})"
            )

        try:
            r = httpx.get(
                f"https://doi.org/{doi}",
                headers=headers,
                follow_redirects=True,
                timeout=15,
            )
            if r.status_code == 200 and "@" in r.text:
                return r.text.strip()
        except httpx.HTTPError:
            pass

        try:
            r = httpx.get(
                f"https://api.crossref.org/works/{doi}/transform/application/x-bibtex",
                headers=headers,
                timeout=15,
            )
            if r.status_code == 200 and "@" in r.text:
                return r.text.strip()
        except httpx.HTTPError:
            pass

        return None

work_to_dict(w)

Convert a Work entity to a plain dict for the DB and UI.

Source code in openalex_pygui/api.py

def work_to_dict(w: Work) -> dict:
    """Convert a Work entity to a plain dict for the DB and UI."""
    authors = []
    if hasattr(w, "authorships") and w.authorships:
        for i, a in enumerate(w.authorships):
            authors.append(
                {
                    "id": a.author.id or "",
                    "name": a.author.display_name or "",
                    "orcid": a.author.orcid,
                    "position": i,
                }
            )

    keywords = []
    if hasattr(w, "keywords") and w.keywords:
        keywords = [kw.display_name for kw in w.keywords if kw.display_name]

    relationships = []
    if hasattr(w, "related_works") and w.related_works:
        relationships = [{"id": rid, "type": "related"} for rid in w.related_works]
    if hasattr(w, "referenced_works") and w.referenced_works:
        relationships.extend(
            {"id": rid, "type": "references"} for rid in w.referenced_works
        )

    abstract = ""
    if hasattr(w, "abstract"):
        abstract = w.abstract or ""

    journal = ""
    if hasattr(w, "primary_location") and w.primary_location:
        loc = w.primary_location
        if hasattr(loc, "source") and loc.source:
            journal = loc.source.display_name or ""

    return {
        "id": w.id if hasattr(w, "id") else "",
        "doi": w.doi if hasattr(w, "doi") else None,
        "title": w.title if hasattr(w, "title") else "",
        "publication_year": w.publication_year if hasattr(w, "publication_year") else None,
        "type": w.type if hasattr(w, "type") else None,
        "cited_by_count": w.cited_by_count if hasattr(w, "cited_by_count") else 0,
        "relevance_score": w.relevance_score if hasattr(w, "relevance_score") else None,
        "abstract": abstract,
        "journal": journal,
        "authors": authors,
        "keywords": keywords,
        "relationships": relationships,
    }