API Reference

Browse the programmatic API for pyzot. This SDK provides safe, read-only database connections and query helpers.

Database Connection

pyzot.db.ZoteroDatabase

Context-manager wrapper around a read-only sqlite3 connection.

Source code in src/pyzot/db.py

class ZoteroDatabase:
    """Context-manager wrapper around a read-only sqlite3 connection."""

    def __init__(self, path: Path | str | None = None, warn_if_open: bool = True):
        if path is None:
            path = discover_db()
        self.path = Path(path)

        if warn_if_open and _check_wal_lock(self.path):
            import warnings

            warnings.warn(
                f"Zotero appears to be running (WAL journal found at {self.path}-wal). "
                "Changes made in Zotero may not be visible until it closes.",
                UserWarning,
                stacklevel=2,
            )

        uri = f"file:{self.path}?mode=ro"
        self._conn = sqlite3.connect(uri, uri=True, check_same_thread=False)
        self._conn.row_factory = sqlite3.Row
        self._conn.execute("PRAGMA query_only = ON")

        self.schema_version = _check_schema_version(self._conn)

    # ------------------------------------------------------------------
    # Low-level helpers
    # ------------------------------------------------------------------

    def execute(self, sql: str, params: tuple = ()) -> sqlite3.Cursor:
        return self._conn.execute(sql, params)

    def fetchall(self, sql: str, params: tuple = ()) -> list[sqlite3.Row]:
        return self._conn.execute(sql, params).fetchall()

    def fetchone(self, sql: str, params: tuple = ()) -> sqlite3.Row | None:
        return self._conn.execute(sql, params).fetchone()

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

    # ------------------------------------------------------------------
    # Context manager
    # ------------------------------------------------------------------

    def __enter__(self) -> ZoteroDatabase:
        return self

    def __exit__(self, *_: object) -> None:
        self.close()

    def __repr__(self) -> str:
        return f"ZoteroDatabase(path={self.path!r}, schema_version={self.schema_version})"

Queries: Search

pyzot.queries.search

Search queries — field-level and full-text.

get_item_fulltext(db, item_id_or_key)

Return indexed full-text token content (bag-of-words) with metadata fallback.

Source code in src/pyzot/queries/search.py

def get_item_fulltext(db: ZoteroDatabase, item_id_or_key: int | str) -> str | None:
    """Return indexed full-text token content (bag-of-words) with metadata fallback."""
    text, _ = get_item_fulltext_with_strategy(db, item_id_or_key, prefer_network=False)
    return text

get_item_fulltext_with_strategy(db, item_id_or_key, *, prefer_network=True, auth=None)

Return full-text plus the source used to retrieve it.

Strategy order: 1) .zotero-ft-cache file of any attached PDF (cheapest, most reliable) 2) direct network access (institution/network-location access) 3) config-based auth credentials 4) item metadata fallback

Source code in src/pyzot/queries/search.py

def get_item_fulltext_with_strategy(
    db: ZoteroDatabase,
    item_id_or_key: int | str,
    *,
    prefer_network: bool = True,
    auth: dict[str, str] | None = None,
) -> tuple[str | None, str]:
    """Return full-text plus the source used to retrieve it.

    Strategy order:
    1) ``.zotero-ft-cache`` file of any attached PDF (cheapest, most reliable)
    2) direct network access (institution/network-location access)
    3) config-based auth credentials
    4) item metadata fallback
    """
    item = get_item(db, item_id_or_key)
    if item is None:
        return None, "not_found"

    # 1) Local Zotero full-text cache (.zotero-ft-cache) for any attachment
    cached = _read_local_fulltext_cache(item, db.path.parent)
    if cached:
        return cached, "cache"

    if prefer_network:
        urls = _candidate_fulltext_urls(item)
        for url in urls:
            text = _fetch_url_text(url)
            if text:
                return text, "network"

        auth_cfg = auth or {}
        user = (auth_cfg.get("username") or "").strip()
        password = auth_cfg.get("password") or ""
        if user and password:
            for url in urls:
                text = _fetch_url_text(url, username=user, password=password)
                if text:
                    return text, "config_auth"

    fallback_parts = [item.title]
    abstract = item.fields.get("abstractNote") or item.fields.get("abstract")
    if abstract:
        fallback_parts.append(abstract)
    fallback_parts.extend(note.plain_text for note in item.notes if note.plain_text)
    text = "\n\n".join(part for part in fallback_parts if part).strip()
    if text:
        return text, "metadata"
    return None, "none"

search_by_author(db, query)

Search items by creator first or last name (case-insensitive, partial match).

Source code in src/pyzot/queries/search.py

def search_by_author(db: ZoteroDatabase, query: str) -> list[Item]:
    """Search items by creator first or last name (case-insensitive, partial match)."""
    like = f"%{query}%"
    rows = db.fetchall(
        """
        SELECT DISTINCT ic.itemID
        FROM itemCreators ic
        JOIN creators c ON ic.creatorID = c.creatorID
        WHERE c.lastName LIKE ? OR c.firstName LIKE ?
        """,
        (like, like),
    )
    return _build_items(db, [r["itemID"] for r in rows])

search_fulltext(db, query)

Search the full-text index (fulltextWords table).

Source code in src/pyzot/queries/search.py

def search_fulltext(db: ZoteroDatabase, query: str) -> list[Item]:
    """Search the full-text index (fulltextWords table)."""
    words = query.lower().split()
    if not words:
        return []

    # Each word must appear in the same item
    word_subqueries = " AND ".join(
        f"EXISTS (SELECT 1 FROM fulltextWords fw{i} JOIN fulltextItems fi{i} "
        f"ON fw{i}.wordID = fi{i}.wordID WHERE fi{i}.itemID = i.itemID "
        f"AND fw{i}.word LIKE ?)"
        for i in range(len(words))
    )
    params = tuple(f"%{w}%" for w in words)

    sql = f"""
        SELECT DISTINCT i.itemID
        FROM items i
        WHERE i.itemID NOT IN (SELECT itemID FROM itemNotes)
          AND i.itemID NOT IN (SELECT itemID FROM itemAttachments)
          AND {word_subqueries}
        ORDER BY i.dateAdded DESC
    """
    try:
        rows = db.fetchall(sql, params)
        return _build_items(db, [r["itemID"] for r in rows])
    except Exception:
        # fulltextWords may not be populated; fall back to field search
        return search_items(db, query)

search_items(db, query, fields=None, item_type=None)

Search item field values. If fields is None, search all fields.

Source code in src/pyzot/queries/search.py

def search_items(
    db: ZoteroDatabase,
    query: str,
    fields: list[str] | None = None,
    item_type: str | None = None,
) -> list[Item]:
    """Search item field values. If fields is None, search all fields."""
    like = f"%{query}%"

    if fields:
        placeholders = ",".join("?" * len(fields))
        field_clause = f"AND f.fieldName IN ({placeholders})"
        field_params: tuple = tuple(fields)
    else:
        field_clause = ""
        field_params = ()

    type_clause = ""
    type_params: tuple = ()
    if item_type:
        type_clause = "AND it.typeName = ?"
        type_params = (item_type,)

    sql = f"""
        SELECT DISTINCT i.itemID
        FROM items i
        JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
        LEFT JOIN itemData id ON i.itemID = id.itemID
        LEFT JOIN fields f ON id.fieldID = f.fieldID
        LEFT JOIN itemDataValues idv ON id.valueID = idv.valueID
        WHERE i.itemID NOT IN (SELECT itemID FROM itemNotes)
          AND i.itemID NOT IN (SELECT itemID FROM itemAttachments)
          AND idv.value LIKE ?
          {field_clause}
          {type_clause}
        ORDER BY i.dateAdded DESC
    """
    params = (like,) + field_params + type_params
    rows = db.fetchall(sql, params)
    return _build_items(db, [r["itemID"] for r in rows])

Queries: Items

pyzot.queries.items

Item queries — the core denormalised fetch.

get_top_level_items(db, library_id=None)

Items not in any collection (top-level unfiled items).

Source code in src/pyzot/queries/items.py

def get_top_level_items(db: ZoteroDatabase, library_id: int | None = None) -> list[Item]:
    """Items not in any collection (top-level unfiled items)."""
    sql = """
        SELECT i.itemID FROM items i
        WHERE i.itemID NOT IN (SELECT itemID FROM itemNotes)
          AND i.itemID NOT IN (SELECT itemID FROM itemAttachments)
          AND i.itemID NOT IN (SELECT itemID FROM collectionItems)
    """
    params: list = []
    if library_id is not None:
        sql += " AND i.libraryID = ?"
        params.append(library_id)
    rows = db.fetchall(sql, tuple(params))
    return _build_items(db, [r["itemID"] for r in rows])

Queries: Collections

pyzot.queries.collections

Collection queries.

get_collection_tree(db, library_id=None)

Return root-level collections with children nested recursively.

Source code in src/pyzot/queries/collections.py

def get_collection_tree(db: ZoteroDatabase, library_id: int | None = None) -> list[Collection]:
    """Return root-level collections with children nested recursively."""
    all_cols = get_all_collections(db, library_id)
    by_id = {c.collection_id: c for c in all_cols}
    roots: list[Collection] = []
    for c in all_cols:
        if c.parent_collection_id is None:
            roots.append(c)
        elif c.parent_collection_id in by_id:
            by_id[c.parent_collection_id].children.append(c)
    return roots

Queries: Attachments

pyzot.queries.attachments

Attachment queries and path resolution.

enrich_attachment_paths(attachments, data_dir)

Resolve and check existence of attachment paths in place.

Source code in src/pyzot/queries/attachments.py

def enrich_attachment_paths(attachments: list[Attachment], data_dir: Path) -> list[Attachment]:
    """Resolve and check existence of attachment paths in place."""
    for att in attachments:
        p = resolve_attachment_path(att, data_dir)
        att.absolute_path = p
        att.file_exists = p is not None and p.exists()
    return attachments

get_all_pdfs(db, data_dir)

Return all (item, attachment) pairs where the attachment is a PDF.

Source code in src/pyzot/queries/attachments.py

def get_all_pdfs(db: ZoteroDatabase, data_dir: Path) -> list[tuple[Item, Attachment]]:
    """Return all (item, attachment) pairs where the attachment is a PDF."""
    rows = db.fetchall(
        """
        SELECT ia.parentItemID, i.itemID, i.key, ia.linkMode,
               ia.contentType, ia.path
        FROM itemAttachments ia
        JOIN items i ON ia.itemID = i.itemID
        WHERE LOWER(ia.contentType) = 'application/pdf'
          AND ia.parentItemID IS NOT NULL
        """
    )
    # Group by parent item
    from collections import defaultdict

    parent_ids: list[int] = []
    att_by_parent: dict[int, list[Attachment]] = defaultdict(list)
    for r in rows:
        parent_id = r["parentItemID"]
        att = Attachment(
            item_id=r["itemID"],
            key=r["key"],
            parent_item_id=parent_id,
            link_mode=r["linkMode"] if r["linkMode"] is not None else 0,
            content_type=r["contentType"] or "",
            path=r["path"],
        )
        p = resolve_attachment_path(att, data_dir)
        att.absolute_path = p
        att.file_exists = p is not None and p.exists()
        att_by_parent[parent_id].append(att)
        if parent_id not in parent_ids:
            parent_ids.append(parent_id)

    items = {item.item_id: item for item in _build_items(db, parent_ids)}
    result: list[tuple[Item, Attachment]] = []
    for pid in parent_ids:
        item = items.get(pid)
        if item:
            for att in att_by_parent[pid]:
                result.append((item, att))
    return result

resolve_attachment_path(attachment, data_dir)

Resolve the absolute path of an attachment file.

Source code in src/pyzot/queries/attachments.py

def resolve_attachment_path(attachment: Attachment, data_dir: Path) -> Path | None:
    """Resolve the absolute path of an attachment file."""
    if attachment.link_mode == 0:  # imported_file: storage/{key}/{filename}
        if attachment.path:
            # Zotero stores "storage:filename" for imported files
            filename = attachment.path.removeprefix("storage:")
            p = data_dir / "storage" / attachment.key / filename
        else:
            # Try to find any file under the storage key directory
            storage_dir = data_dir / "storage" / attachment.key
            if storage_dir.exists():
                files = list(storage_dir.iterdir())
                return files[0] if files else None
            return None
        return p
    elif attachment.link_mode == 2:  # linked_file: absolute path
        if attachment.path:
            return Path(attachment.path)
    elif attachment.link_mode == 1:  # imported_url: same as imported_file
        if attachment.path:
            filename = attachment.path.removeprefix("storage:")
            return data_dir / "storage" / attachment.key / filename
    # linked_url (3) has no local file
    return None