zotero_db

zocrates.rag.zotero_db

Read-only access layer for the Zotero SQLite database.

All queries use a URI connection opened in read-only mode to guarantee we never accidentally write to the user's library.

ARXIV_ID_PATTERN module-attribute

ARXIV_ID_PATTERN = re.compile('^(?:\\d{4}\\.\\d{4,5}|[a-zA-Z\\-]+(?:\\.[a-zA-Z\\-]+)?\\/\\d{7})$')

_PUBLICATION_FIELDS module-attribute

_PUBLICATION_FIELDS = ('publicationTitle', 'proceedingsTitle', 'bookTitle', 'repository', 'websiteTitle', 'blogTitle', 'publisher', 'institution', 'university', 'encyclopediaTitle', 'dictionaryTitle')

_SQL_BATCH_SIZE module-attribute

_SQL_BATCH_SIZE = 500

_sqlite_snapshot_lock module-attribute

_sqlite_snapshot_lock = threading.Lock()

_sqlite_snapshots module-attribute

_sqlite_snapshots = {}

_swept_stale_snapshots module-attribute

_swept_stale_snapshots = False

logger module-attribute

logger = logging.getLogger(__name__)

ZoteroAttachment dataclass

ZoteroAttachment(item_id, item_key, parent_item_id, parent_item_key, content_type, path, title='')

An attachment (PDF, HTML, etc.) linked to a Zotero item.

ZoteroCollection dataclass

ZoteroCollection(name, key)

Representation of a Zotero collection.

ZoteroDB

ZoteroDB(sqlite_path, storage_dir)

Read-only façade over zotero.sqlite.

Initialize ZoteroDB.

Parameters:

Name	Type	Description	Default
sqlite_path	Path	Absolute path to the zotero.sqlite database file.	required
storage_dir	Path	Absolute path to the Zotero storage/ directory, used to resolve relative attachment paths.	required

Source code in zocrates/rag/zotero_db.py

def __init__(self, sqlite_path: Path, storage_dir: Path) -> None:
    """Initialize ZoteroDB.

    Args:
        sqlite_path: Absolute path to the ``zotero.sqlite`` database file.
        storage_dir: Absolute path to the Zotero ``storage/`` directory,
            used to resolve relative attachment paths.
    """
    self._sqlite_path = sqlite_path
    self._storage_dir = storage_dir  # <ZoteroDataDir>/storage/
    self._table_exists_cache: dict[str, bool] = {}

get_active_item_keys

get_active_item_keys(item_keys)

Return a set of item keys that exist in the library and are not deleted or parent-deleted.

Parameters:

Name	Type	Description	Default
item_keys	list[str]	List of Zotero item keys (can include regular items, attachments, notes).	required

Returns:

Type	Description
set[str]	Set of active item keys.

Source code in zocrates/rag/zotero_db.py

def get_active_item_keys(self, item_keys: list[str]) -> set[str]:
    """Return a set of item keys that exist in the library and are not deleted or parent-deleted.

    Args:
        item_keys: List of Zotero item keys (can include regular items, attachments, notes).

    Returns:
        Set of active item keys.
    """
    if not item_keys:
        return set()

    unique_keys = list(dict.fromkeys(k for k in item_keys if k))
    if not unique_keys:
        return set()

    active: set[str] = set()
    batch_size = _SQL_BATCH_SIZE

    with _open_db(self._sqlite_path) as conn:
        has_deleted_items = self._table_exists(conn, "deletedItems")
        has_item_attachments = self._table_exists(conn, "itemAttachments")
        has_item_notes = self._table_exists(conn, "itemNotes")

        del_clause = ""
        if has_deleted_items:
            del_clause = "AND i.itemID NOT IN (SELECT itemID FROM deletedItems)"
            if has_item_attachments:
                del_clause += " AND (ia.parentItemID IS NULL OR ia.parentItemID NOT IN (SELECT itemID FROM deletedItems))"
            if has_item_notes:
                del_clause += " AND (n.parentItemID IS NULL OR n.parentItemID NOT IN (SELECT itemID FROM deletedItems))"

        join_attachments = (
            "LEFT JOIN itemAttachments ia ON i.itemID = ia.itemID"
            if has_item_attachments
            else ""
        )
        join_notes = (
            "LEFT JOIN itemNotes n ON i.itemID = n.itemID"
            if has_item_notes
            else ""
        )

        for i in range(0, len(unique_keys), batch_size):
            batch = unique_keys[i : i + batch_size]
            placeholders = ",".join("?" for _ in batch)
            sql = f"""
                SELECT i.key
                FROM items i
                {join_attachments}
                {join_notes}
                WHERE i.key IN ({placeholders})
                  {del_clause}
            """
            rows = conn.execute(sql, batch).fetchall()
            for row in rows:
                if row["key"]:
                    active.add(row["key"])

    return active

get_all_arxiv_ids

get_all_arxiv_ids()

Return a set of all normalized arXiv IDs across active non-deleted items in the library.

Source code in zocrates/rag/zotero_db.py

def get_all_arxiv_ids(self) -> set[str]:
    """Return a set of all normalized arXiv IDs across active non-deleted items in the library."""
    with _open_db(self._sqlite_path) as conn:
        has_deleted_items = self._table_exists(conn, "deletedItems")
        del_clause = "AND i.itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted_items else ""
        sql = f"""
            SELECT f.fieldName, idv.value
            FROM items i
            JOIN itemData id ON i.itemID = id.itemID
            JOIN fields f ON id.fieldID = f.fieldID
            JOIN itemDataValues idv ON id.valueID = idv.valueID
            WHERE f.fieldName IN ('extra', 'url', 'DOI')
              {del_clause}
        """
        rows = conn.execute(sql).fetchall()
        arxiv_ids: set[str] = set()
        for r in rows:
            field_name = r["fieldName"]
            val = r["value"]
            cand: str | None = None
            if field_name == "extra":
                cand = extract_arxiv_id(extra=val)
            elif field_name == "url":
                cand = extract_arxiv_id(url=val)
            elif field_name == "DOI":
                cand = extract_arxiv_id(doi=val)
            if cand:
                arxiv_ids.add(cand)
        return arxiv_ids

get_all_attachments

get_all_attachments(content_types=('application/pdf', 'text/html', 'text/markdown', 'text/plain'))

Return all file attachments with a parent item across the library.

Parameters:

Name	Type	Description	Default
content_types	tuple[str, ...]	Tuple of MIME types to include.	('application/pdf', 'text/html', 'text/markdown', 'text/plain')

Returns:

Type	Description
list[ZoteroAttachment]	List of ZoteroAttachment objects with resolved filesystem paths.

Source code in zocrates/rag/zotero_db.py

def get_all_attachments(
    self,
    content_types: tuple[str, ...] = ("application/pdf", "text/html", "text/markdown", "text/plain"),
) -> list[ZoteroAttachment]:
    """Return all file attachments with a parent item across the library.

    Args:
        content_types: Tuple of MIME types to include.

    Returns:
        List of ``ZoteroAttachment`` objects with resolved filesystem paths.
    """
    sql = """
        SELECT
            i.itemID,
            i.key,
            ia.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(ia.contentType, '') AS contentType,
            COALESCE(ia.path, '') AS path
        FROM items i
        JOIN itemAttachments ia ON i.itemID = ia.itemID
        LEFT JOIN items parentItem ON ia.parentItemID = parentItem.itemID
        WHERE i.itemID NOT IN (SELECT itemID FROM deletedItems)
          AND ia.parentItemID IS NOT NULL
    """
    with _open_db(self._sqlite_path) as conn:
        rows = conn.execute(sql).fetchall()

    attachments: list[ZoteroAttachment] = []
    for row in rows:
        if row["contentType"] not in content_types:
            continue
        resolved = self._resolve_attachment_path(row["key"], row["path"])
        attachments.append(
            ZoteroAttachment(
                item_id=row["itemID"],
                item_key=row["key"],
                parent_item_id=row["parentItemID"],
                parent_item_key=row["parentKey"],
                content_type=row["contentType"],
                path=resolved,
            )
        )
    return attachments

get_all_convertible_attachments

get_all_convertible_attachments()

Return all valid document attachments in the library in a single fast query.

Source code in zocrates/rag/zotero_db.py

def get_all_convertible_attachments(self) -> list[dict[str, Any]]:
    """Return all valid document attachments in the library in a single fast query."""
    sql = """
        SELECT
            i.itemID AS attachment_id,
            i.key AS attachment_key,
            ia.parentItemID AS parent_item_id,
            parentItem.key AS parent_key,
            COALESCE(idvTitle.value, '') AS parent_title,
            COALESCE(ia.contentType, '') AS content_type,
            COALESCE(ia.path, '') AS raw_path
        FROM items i
        JOIN itemAttachments ia ON i.itemID = ia.itemID
        LEFT JOIN items parentItem ON ia.parentItemID = parentItem.itemID
        LEFT JOIN itemData idTitle ON parentItem.itemID = idTitle.itemID AND idTitle.fieldID = (
            SELECT fieldID FROM fields WHERE fieldName = 'title'
        )
        LEFT JOIN itemDataValues idvTitle ON idTitle.valueID = idvTitle.valueID
        WHERE i.itemID NOT IN (SELECT itemID FROM deletedItems)
          AND (parentItem.itemID IS NULL OR parentItem.itemID NOT IN (SELECT itemID FROM deletedItems))
        ORDER BY i.itemID
    """
    with _open_db(self._sqlite_path) as conn:
        rows = conn.execute(sql).fetchall()

    results: list[dict[str, Any]] = []
    for row in rows:
        resolved_path = self._resolve_attachment_path(row["attachment_key"], row["raw_path"])
        if resolved_path and resolved_path.exists():
            results.append({
                "attachment_id": row["attachment_id"],
                "attachment_key": row["attachment_key"],
                "parent_item_id": row["parent_item_id"],
                "parent_key": row["parent_key"] or "",
                "parent_title": row["parent_title"] or "",
                "content_type": row["content_type"],
                "path": resolved_path,
                "filename": resolved_path.name,
            })
    return results

get_all_dedup_records

get_all_dedup_records()

Return lightweight records (title, year, creators, doi, arxiv_id) for active items for fuzzy dedup.

Source code in zocrates/rag/zotero_db.py

def get_all_dedup_records(self) -> list[dict[str, Any]]:
    """Return lightweight records (title, year, creators, doi, arxiv_id) for active items for fuzzy dedup."""
    with _open_db(self._sqlite_path) as conn:
        has_deleted_items = self._table_exists(conn, "deletedItems")
        del_clause = "AND i.itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted_items else ""
        sql = f"""
            SELECT i.itemID, f.fieldName, idv.value
            FROM items i
            JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
            JOIN itemData id ON i.itemID = id.itemID
            JOIN fields f ON id.fieldID = f.fieldID
            JOIN itemDataValues idv ON id.valueID = idv.valueID
            WHERE it.typeName NOT IN ('attachment', 'note')
              AND f.fieldName IN ('title', 'date', 'DOI', 'url', 'extra')
              {del_clause}
            ORDER BY i.itemID
        """
        rows = conn.execute(sql).fetchall()

        records_map: dict[int, dict[str, Any]] = {}
        for r in rows:
            item_id = r["itemID"]
            if item_id not in records_map:
                records_map[item_id] = {
                    "item_id": item_id,
                    "title": None,
                    "year": None,
                    "doi": None,
                    "url": None,
                    "extra": None,
                }
            field_name = r["fieldName"]
            val = r["value"]
            if field_name == "title":
                records_map[item_id]["title"] = val
            elif field_name == "date":
                y_str = extract_year(val)
                if y_str:
                    try:
                        records_map[item_id]["year"] = int(y_str)
                    except ValueError:
                        pass
            elif field_name == "DOI":
                records_map[item_id]["doi"] = normalize_doi(val)
            elif field_name == "url":
                records_map[item_id]["url"] = val
            elif field_name == "extra":
                records_map[item_id]["extra"] = val

        sql_creators = f"""
            SELECT ic.itemID, c.lastName
            FROM itemCreators ic
            JOIN creators c ON ic.creatorID = c.creatorID
            JOIN items i ON ic.itemID = i.itemID
            WHERE 1=1 {del_clause}
            ORDER BY ic.itemID, ic.orderIndex
        """
        c_rows = conn.execute(sql_creators).fetchall()
        creators_map: dict[int, list[str]] = {}
        for cr in c_rows:
            i_id = cr["itemID"]
            lname = (cr["lastName"] or "").strip()
            if lname:
                creators_map.setdefault(i_id, []).append(lname)

        results: list[dict[str, Any]] = []
        for item_id, rec in records_map.items():
            title = rec["title"]
            if not title:
                continue
            arxiv_id = extract_arxiv_id(extra=rec["extra"], url=rec["url"], doi=rec["doi"])
            results.append({
                "item_id": item_id,
                "title": title,
                "year": rec["year"],
                "doi": rec["doi"],
                "arxiv_id": arxiv_id,
                "creators": creators_map.get(item_id, []),
            })
        return results

get_all_dois

get_all_dois()

Return a set of all normalized DOIs across active non-deleted items in the library.

Source code in zocrates/rag/zotero_db.py

def get_all_dois(self) -> set[str]:
    """Return a set of all normalized DOIs across active non-deleted items in the library."""
    with _open_db(self._sqlite_path) as conn:
        has_deleted_items = self._table_exists(conn, "deletedItems")
        del_clause = "AND i.itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted_items else ""
        sql = f"""
            SELECT idv.value
            FROM items i
            JOIN itemData id ON i.itemID = id.itemID
            JOIN fields f ON id.fieldID = f.fieldID
            JOIN itemDataValues idv ON id.valueID = idv.valueID
            WHERE f.fieldName = 'DOI'
              {del_clause}
        """
        rows = conn.execute(sql).fetchall()
        dois: set[str] = set()
        for r in rows:
            norm = normalize_doi(r["value"])
            if norm:
                dois.add(norm)
        return dois

get_all_item_relations

get_all_item_relations()

Return all explicit item relations as (source_key, target_key) pairs.

Source code in zocrates/rag/zotero_db.py

def get_all_item_relations(self) -> list[tuple[str, str]]:
    """Return all explicit item relations as (source_key, target_key) pairs."""
    with _open_db(self._sqlite_path) as conn:
        if not self._table_exists(conn, "itemRelations"):
            return []

        has_deleted_items = self._table_exists(conn, "deletedItems")
        del_clause = "WHERE i.itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted_items else ""
        sql = f"""
            SELECT i.key AS source_key, ir.object AS target_uri
            FROM itemRelations ir
            JOIN items i ON ir.itemID = i.itemID
            {del_clause}
        """
        rows = conn.execute(sql).fetchall()

        del_keys: set[str] = set()
        if has_deleted_items and self._table_exists(conn, "items"):
            del_rows = conn.execute(
                "SELECT key FROM items WHERE itemID IN (SELECT itemID FROM deletedItems)"
            ).fetchall()
            del_keys = {r["key"] for r in del_rows if r["key"]}

        relations: list[tuple[str, str]] = []
        seen: set[tuple[str, str]] = set()
        for r in rows:
            src = str(r["source_key"] or "").strip()
            target_uri = str(r["target_uri"] or "").strip()
            if not src or "/items/" not in target_uri:
                continue
            target_key = target_uri.split("/items/")[-1].strip().split("/")[0].split("?")[0].split("#")[0]
            if target_key and target_key not in del_keys and target_key != src:
                pair = (src, target_key)
                if pair not in seen:
                    seen.add(pair)
                    relations.append(pair)

        return relations

get_all_items

get_all_items()

Return all non-attachment, non-deleted items in the library.

Returns:

Type	Description
list[ZoteroItem]	List of fully populated ZoteroItem objects.

Source code in zocrates/rag/zotero_db.py

def get_all_items(self) -> list[ZoteroItem]:
    """Return all non-attachment, non-deleted items in the library.

    Returns:
        List of fully populated ``ZoteroItem`` objects.
    """
    with _open_db(self._sqlite_path) as conn:
        has_deleted = self._table_exists(conn, "deletedItems")
        del_clause = "AND i.itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted else ""
        sql = f"""
            SELECT
                i.itemID,
                i.key,
                it.typeName
            FROM items i
            JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
            WHERE it.typeName != 'attachment'
              AND it.typeName != 'note'
              {del_clause}
            ORDER BY i.itemID
        """
        rows = conn.execute(sql).fetchall()
        return self._build_items_batched(conn, rows)

get_all_notes

get_all_notes()

Return all notes (standalone and child notes) in the library.

Returns:

Type	Description
list[ZoteroNote]	List of ZoteroNote objects ordered by note_id.

Source code in zocrates/rag/zotero_db.py

def get_all_notes(self) -> list[ZoteroNote]:
    """Return all notes (standalone and child notes) in the library.

    Returns:
        List of ``ZoteroNote`` objects ordered by ``note_id``.
    """
    sql = """
        SELECT
            i.itemID AS noteID,
            i.key AS noteKey,
            n.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(n.note, '') AS noteHtml,
            COALESCE(n.title, '') AS noteTitle
        FROM items i
        JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
        JOIN itemNotes n ON i.itemID = n.itemID
        LEFT JOIN items parentItem ON n.parentItemID = parentItem.itemID
        WHERE it.typeName = 'note'
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
        ORDER BY i.itemID
    """
    with _open_db(self._sqlite_path) as conn:
        rows = conn.execute(sql).fetchall()

    notes: list[ZoteroNote] = []
    for row in rows:
        notes.append(
            ZoteroNote(
                note_id=row["noteID"],
                note_key=row["noteKey"],
                parent_item_id=row["parentItemID"],
                parent_item_key=row["parentKey"],
                title=row["noteTitle"],
                note_html=row["noteHtml"],
            )
        )
    return notes

get_all_titles

get_all_titles()

Return a list of all titles across active non-deleted items in the library.

Source code in zocrates/rag/zotero_db.py

def get_all_titles(self) -> list[str]:
    """Return a list of all titles across active non-deleted items in the library."""
    with _open_db(self._sqlite_path) as conn:
        has_deleted_items = self._table_exists(conn, "deletedItems")
        del_clause = "AND i.itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted_items else ""
        sql = f"""
            SELECT idv.value
            FROM items i
            JOIN itemData id ON i.itemID = id.itemID
            JOIN fields f ON id.fieldID = f.fieldID
            JOIN itemDataValues idv ON id.valueID = idv.valueID
            WHERE f.fieldName = 'title'
              {del_clause}
        """
        rows = conn.execute(sql).fetchall()
        titles: list[str] = []
        for r in rows:
            t = (r["value"] or "").strip()
            if t:
                titles.append(t)
        return titles

get_attachment_by_id

get_attachment_by_id(item_id)

Fetch a single file attachment by its item ID.

Parameters:

Name	Type	Description	Default
item_id	int	Numeric Zotero item ID of the attachment.	required

Returns:

Type	Description
ZoteroAttachment | None	ZoteroAttachment with a resolved path, or None if not found
ZoteroAttachment | None	or deleted.

Source code in zocrates/rag/zotero_db.py

def get_attachment_by_id(
    self,
    item_id: int,
) -> ZoteroAttachment | None:
    """Fetch a single file attachment by its item ID.

    Args:
        item_id: Numeric Zotero item ID of the attachment.

    Returns:
        ``ZoteroAttachment`` with a resolved path, or ``None`` if not found
        or deleted.
    """
    sql = """
        SELECT
            i.itemID,
            i.key,
            ia.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(ia.contentType, '') AS contentType,
            COALESCE(ia.path, '') AS path
        FROM items i
        JOIN itemAttachments ia ON i.itemID = ia.itemID
        LEFT JOIN items parentItem ON ia.parentItemID = parentItem.itemID
        WHERE i.itemID = ?
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
    """
    with _open_db(self._sqlite_path) as conn:
        row = conn.execute(sql, (item_id,)).fetchone()
        if row is None:
            return None
        resolved = self._resolve_attachment_path(row["key"], row["path"])
        return ZoteroAttachment(
            item_id=row["itemID"],
            item_key=row["key"],
            parent_item_id=row["parentItemID"],
            parent_item_key=row["parentKey"],
            content_type=row["contentType"],
            path=resolved,
        )

get_attachment_by_key

get_attachment_by_key(item_key)

Fetch a single file attachment by its item key or parent item key.

Parameters:

Name	Type	Description	Default
item_key	str	Zotero item key string (attachment key or parent item key).	required

Returns:

Type	Description
ZoteroAttachment | None	ZoteroAttachment with a resolved path, or None if not found.

Source code in zocrates/rag/zotero_db.py

def get_attachment_by_key(
    self,
    item_key: str,
) -> ZoteroAttachment | None:
    """Fetch a single file attachment by its item key or parent item key.

    Args:
        item_key: Zotero item key string (attachment key or parent item key).

    Returns:
        ``ZoteroAttachment`` with a resolved path, or ``None`` if not found.
    """
    sql = """
        SELECT
            i.itemID,
            i.key,
            ia.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(ia.contentType, '') AS contentType,
            COALESCE(ia.path, '') AS path
        FROM items i
        JOIN itemAttachments ia ON i.itemID = ia.itemID
        LEFT JOIN items parentItem ON ia.parentItemID = parentItem.itemID
        WHERE (i.key = ? OR parentItem.key = ?)
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
        ORDER BY CASE WHEN ia.contentType = 'application/pdf' THEN 0 ELSE 1 END
    """
    with _open_db(self._sqlite_path) as conn:
        row = conn.execute(sql, (item_key, item_key)).fetchone()
        if row:
            resolved = self._resolve_attachment_path(row["key"], row["path"])
            return ZoteroAttachment(
                item_id=row["itemID"],
                item_key=row["key"],
                parent_item_id=row["parentItemID"],
                parent_item_key=row["parentKey"],
                content_type=row["contentType"],
                path=resolved,
            )

    item = self.get_item_by_key(item_key)
    if item:
        attachments = self.get_attachments(item.item_id)
        if attachments:
            return sorted(
                attachments,
                key=lambda a: 0 if a.content_type == "application/pdf" else 1,
            )[0]

    return None

get_attachments

get_attachments(parent_item_id, content_types=('application/pdf', 'text/html', 'text/markdown', 'text/plain'))

Return file attachments for a given parent item.

Parameters:

Name	Type	Description	Default
parent_item_id	int	Numeric ID of the parent Zotero item.	required
content_types	tuple[str, ...]	Tuple of MIME types to include; defaults to PDF, HTML, Markdown, and plain text.	('application/pdf', 'text/html', 'text/markdown', 'text/plain')

Returns:

Type	Description
list[ZoteroAttachment]	List of ZoteroAttachment objects with resolved filesystem
list[ZoteroAttachment]	paths (path is None if the file cannot be resolved).
list[ZoteroAttachment]	The title field is set to the Zotero attachment title if
list[ZoteroAttachment]	available, otherwise the attachment filename.

Source code in zocrates/rag/zotero_db.py

def get_attachments(
    self,
    parent_item_id: int,
    content_types: tuple[str, ...] = ("application/pdf", "text/html", "text/markdown", "text/plain"),
) -> list[ZoteroAttachment]:
    """Return file attachments for a given parent item.

    Args:
        parent_item_id: Numeric ID of the parent Zotero item.
        content_types: Tuple of MIME types to include; defaults to PDF,
            HTML, Markdown, and plain text.

    Returns:
        List of ``ZoteroAttachment`` objects with resolved filesystem
        paths (``path`` is ``None`` if the file cannot be resolved).
        The ``title`` field is set to the Zotero attachment title if
        available, otherwise the attachment filename.
    """
    sql = """
        SELECT
            i.itemID,
            i.key,
            ia.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(ia.contentType, '') AS contentType,
            COALESCE(ia.path, '') AS path,
            COALESCE(att_meta.att_title, '') AS attachmentTitle
        FROM items i
        JOIN itemAttachments ia ON i.itemID = ia.itemID
        LEFT JOIN items parentItem ON ia.parentItemID = parentItem.itemID
        LEFT JOIN (
            SELECT id2.itemID, idv2.value AS att_title
            FROM itemData id2
            JOIN fields f2 ON id2.fieldID = f2.fieldID
            JOIN itemDataValues idv2 ON id2.valueID = idv2.valueID
            WHERE f2.fieldName = 'title'
        ) att_meta ON i.itemID = att_meta.itemID
        WHERE ia.parentItemID = ?
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
    """
    with _open_db(self._sqlite_path) as conn:
        rows = conn.execute(sql, (parent_item_id,)).fetchall()

    attachments: list[ZoteroAttachment] = []
    for row in rows:
        if row["contentType"] not in content_types:
            continue
        resolved = self._resolve_attachment_path(row["key"], row["path"])
        raw_title = (row["attachmentTitle"] or "").strip()
        filename = (resolved.name if resolved and resolved.name else "")
        if raw_title and filename and raw_title.lower() != filename.lower():
            att_title = f"{raw_title} ({filename})"
        elif raw_title:
            att_title = raw_title
        elif filename:
            att_title = filename
        else:
            att_title = "Attachment"
        attachments.append(
            ZoteroAttachment(
                item_id=row["itemID"],
                item_key=row["key"],
                parent_item_id=row["parentItemID"],
                parent_item_key=row["parentKey"],
                content_type=row["contentType"],
                path=resolved,
                title=att_title,
            )
        )
    return attachments

get_item

get_item(item_id)

Fetch a single item with full metadata.

Parameters:

Name	Type	Description	Default
item_id	int	Numeric Zotero item ID.	required

Returns:

Type	Description
ZoteroItem | None	Fully populated ZoteroItem (with creators, tags, collections,
ZoteroItem | None	and relations), or None if not found or deleted.

Source code in zocrates/rag/zotero_db.py

def get_item(self, item_id: int) -> ZoteroItem | None:
    """Fetch a single item with full metadata.

    Args:
        item_id: Numeric Zotero item ID.

    Returns:
        Fully populated ``ZoteroItem`` (with creators, tags, collections,
        and relations), or ``None`` if not found or deleted.
    """
    with _open_db(self._sqlite_path) as conn:
        has_deleted = self._table_exists(conn, "deletedItems")
        del_clause = "AND i.itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted else ""
        sql = f"""
            SELECT i.itemID, i.key, it.typeName
            FROM items i
            JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
            WHERE i.itemID = ?
              {del_clause}
        """
        row = conn.execute(sql, (item_id,)).fetchone()
        if row is None:
            return None
        return self._build_item(conn, row)

get_item_by_key

get_item_by_key(item_key)

Fetch a single item by its string key.

Parameters:

Name	Type	Description	Default
item_key	str	Zotero item key string (e.g. '89X2ELQ6').	required

Returns:

Type	Description
ZoteroItem | None	Fully populated ZoteroItem or None if not found.

Source code in zocrates/rag/zotero_db.py

def get_item_by_key(self, item_key: str) -> ZoteroItem | None:
    """Fetch a single item by its string key.

    Args:
        item_key: Zotero item key string (e.g. '89X2ELQ6').

    Returns:
        Fully populated ``ZoteroItem`` or ``None`` if not found.
    """
    with _open_db(self._sqlite_path) as conn:
        has_deleted = self._table_exists(conn, "deletedItems")
        del_clause = "AND itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted else ""
        sql = f"""
            SELECT itemID FROM items WHERE key = ? {del_clause}
        """
        row = conn.execute(sql, (item_key,)).fetchone()
        if row is None:
            return None
        return self.get_item(row["itemID"])

get_item_date_modified

get_item_date_modified(item_key)

Fetch the last modified timestamp for an item as timezone-aware UTC datetime.

Parameters:

Name	Type	Description	Default
item_key	str	8-character Zotero alphanumeric key.	required

Returns:

Type	Description
datetime | None	Timezone-aware UTC datetime, or None if item not found.

Source code in zocrates/rag/zotero_db.py

def get_item_date_modified(self, item_key: str) -> datetime | None:
    """Fetch the last modified timestamp for an item as timezone-aware UTC datetime.

    Args:
        item_key: 8-character Zotero alphanumeric key.

    Returns:
        Timezone-aware UTC datetime, or None if item not found.
    """
    sql = """
        SELECT COALESCE(clientDateModified, dateModified) AS modified
        FROM items
        WHERE key = ?
          AND itemID NOT IN (SELECT itemID FROM deletedItems)
    """
    with _open_db(self._sqlite_path) as conn:
        row = conn.execute(sql, (item_key,)).fetchone()
        if row and row["modified"]:
            return _parse_utc_datetime(row["modified"])
    return None

get_item_keys_in_collection

get_item_keys_in_collection(collection_name, recursive=True)

Return the unique item keys for all items belonging to a collection name or key.

Parameters:

Name	Type	Description	Default
collection_name	str	Name or key of the collection.	required
recursive	bool	If True and parentCollectionID exists, includes items in all subcollections.	True

Returns:

Type	Description
list[str]	List of unique item keys.

Source code in zocrates/rag/zotero_db.py

def get_item_keys_in_collection(
    self, collection_name: str, recursive: bool = True
) -> list[str]:
    """Return the unique item keys for all items belonging to a collection name or key.

    Args:
        collection_name: Name or key of the collection.
        recursive: If True and parentCollectionID exists, includes items in all subcollections.

    Returns:
        List of unique item keys.
    """
    with _open_db(self._sqlite_path) as conn:
        has_deleted = self._table_exists(conn, "deletedItems")
        del_clause = "AND i.itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted else ""

        col_columns = {c[1] for c in conn.execute("PRAGMA table_info(collections)").fetchall()}
        has_parent_col = "parentCollectionID" in col_columns

        if recursive and has_parent_col:
            sql = f"""
                WITH RECURSIVE subcols(collectionID) AS (
                    SELECT collectionID
                    FROM collections
                    WHERE collectionName = ? OR key = ?
                    UNION
                    SELECT c.collectionID
                    FROM collections c
                    JOIN subcols s ON c.parentCollectionID = s.collectionID
                )
                SELECT DISTINCT i.key
                FROM items i
                JOIN collectionItems ci ON i.itemID = ci.itemID
                JOIN subcols s ON ci.collectionID = s.collectionID
                WHERE 1=1 {del_clause}
            """
        else:
            sql = f"""
                SELECT DISTINCT i.key
                FROM items i
                JOIN collectionItems ci ON i.itemID = ci.itemID
                JOIN collections col ON ci.collectionID = col.collectionID
                WHERE (col.collectionName = ? OR col.key = ?)
                  {del_clause}
            """
        rows = conn.execute(sql, (collection_name, collection_name)).fetchall()
        return [row["key"] for row in rows]

get_item_relations

get_item_relations(item_key_or_id)

Return target item keys related to the specified item.

Source code in zocrates/rag/zotero_db.py

def get_item_relations(self, item_key_or_id: str | int) -> list[str]:
    """Return target item keys related to the specified item."""
    with _open_db(self._sqlite_path) as conn:
        has_deleted_items = self._table_exists(conn, "deletedItems")
        item_id: int | None = None
        if isinstance(item_key_or_id, int):
            if has_deleted_items:
                del_row = conn.execute(
                    "SELECT 1 FROM deletedItems WHERE itemID = ?",
                    (item_key_or_id,),
                ).fetchone()
                if del_row:
                    return []
            item_id = item_key_or_id
        else:
            del_clause = "AND itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted_items else ""
            row = conn.execute(
                f"SELECT itemID FROM items WHERE key = ? {del_clause}",
                (item_key_or_id,),
            ).fetchone()
            if not row:
                return []
            item_id = row["itemID"]

        return self._fetch_relations(conn, item_id)

get_item_tags

get_item_tags(item_key_or_id, tag_type=None)

Return tag names for an item, optionally filtered by tag type (0=user, 1=automatic).

Source code in zocrates/rag/zotero_db.py

def get_item_tags(
    self, item_key_or_id: str | int, tag_type: int | None = None
) -> list[str]:
    """Return tag names for an item, optionally filtered by tag type (0=user, 1=automatic)."""
    with _open_db(self._sqlite_path) as conn:
        has_item_tags = self._table_exists(conn, "itemTags")
        if not has_item_tags:
            return []

        item_id: int | None = None
        has_deleted_items = self._table_exists(conn, "deletedItems")
        if isinstance(item_key_or_id, int):
            if has_deleted_items:
                del_row = conn.execute(
                    "SELECT 1 FROM deletedItems WHERE itemID = ?",
                    (item_key_or_id,),
                ).fetchone()
                if del_row:
                    return []
            item_id = item_key_or_id
        else:
            del_clause = "AND itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted_items else ""
            row = conn.execute(
                f"SELECT itemID FROM items WHERE key = ? {del_clause}",
                (item_key_or_id,),
            ).fetchone()
            if not row:
                return []
            item_id = row["itemID"]

        params: list[Any] = [item_id]
        type_clause = ""
        if tag_type is not None:
            cols = {c[1] for c in conn.execute("PRAGMA table_info(itemTags)").fetchall()}
            if "type" in cols:
                type_clause = " AND it.type = ?"
                params.append(tag_type)

        sql = f"""
            SELECT DISTINCT t.name
            FROM itemTags it
            JOIN tags t ON it.tagID = t.tagID
            WHERE it.itemID = ?{type_clause}
            ORDER BY t.name
        """
        rows = conn.execute(sql, params).fetchall()
        if not rows:
            has_attachments_table = self._table_exists(conn, "itemAttachments")
            if has_attachments_table:
                p_row = conn.execute(
                    "SELECT parentItemID FROM itemAttachments WHERE itemID = ?",
                    (item_id,),
                ).fetchone()
                if p_row and p_row["parentItemID"]:
                    params[0] = p_row["parentItemID"]
                    rows = conn.execute(sql, params).fetchall()

        return [r["name"] for r in rows if r["name"]]

get_items_by_ids

get_items_by_ids(item_ids)

Fetch multiple items by their numeric IDs.

Parameters:

Name	Type	Description	Default
item_ids	list[int]	List of numeric Zotero item IDs.	required

Returns:

Type	Description
list[ZoteroItem]	List of fully populated ZoteroItem objects (with creators,
list[ZoteroItem]	tags, and collections). Items not found or deleted are silently
list[ZoteroItem]	omitted.

Source code in zocrates/rag/zotero_db.py

def get_items_by_ids(self, item_ids: list[int]) -> list[ZoteroItem]:
    """Fetch multiple items by their numeric IDs.

    Args:
        item_ids: List of numeric Zotero item IDs.

    Returns:
        List of fully populated ``ZoteroItem`` objects (with creators,
        tags, and collections). Items not found or deleted are silently
        omitted.
    """
    if not item_ids:
        return []
    batch_size = _SQL_BATCH_SIZE
    with _open_db(self._sqlite_path) as conn:
        has_deleted = self._table_exists(conn, "deletedItems")
        del_clause = "AND i.itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted else ""
        all_rows: list[sqlite3.Row] = []
        for i in range(0, len(item_ids), batch_size):
            batch_chunk = item_ids[i : i + batch_size]
            placeholders = ",".join("?" for _ in batch_chunk)
            sql = f"""
                SELECT i.itemID, i.key, it.typeName
                FROM items i
                JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
                WHERE i.itemID IN ({placeholders})
                  {del_clause}
            """
            all_rows.extend(conn.execute(sql, batch_chunk).fetchall())
        return self._build_items_batched(conn, all_rows)

get_new_item_ids_since

get_new_item_ids_since(last_seen_item_id)

Return item IDs created after last_seen_item_id for incremental sync.

Parameters:

Name	Type	Description	Default
last_seen_item_id	int	The highest item ID already processed.	required

Returns:

Type	Description
list[int]	Sorted list of numeric item IDs greater than last_seen_item_id
list[int]	that are not attachments, notes, or deleted items.

Source code in zocrates/rag/zotero_db.py

def get_new_item_ids_since(self, last_seen_item_id: int) -> list[int]:
    """Return item IDs created after last_seen_item_id for incremental sync.

    Args:
        last_seen_item_id: The highest item ID already processed.

    Returns:
        Sorted list of numeric item IDs greater than ``last_seen_item_id``
        that are not attachments, notes, or deleted items.
    """
    sql = """
        SELECT i.itemID
        FROM items i
        JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
        WHERE it.typeName != 'attachment'
          AND it.typeName != 'note'
          AND i.itemID > ?
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
        ORDER BY i.itemID
    """
    with _open_db(self._sqlite_path) as conn:
        rows = conn.execute(sql, (last_seen_item_id,)).fetchall()
    return [row["itemID"] for row in rows]

get_note_by_id

get_note_by_id(note_id)

Fetch a single note by note item ID.

Source code in zocrates/rag/zotero_db.py

def get_note_by_id(self, note_id: int) -> ZoteroNote | None:
    """Fetch a single note by note item ID."""
    sql = """
        SELECT
            i.itemID AS noteID,
            i.key AS noteKey,
            n.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(n.note, '') AS noteHtml,
            COALESCE(n.title, '') AS noteTitle
        FROM items i
        JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
        JOIN itemNotes n ON i.itemID = n.itemID
        LEFT JOIN items parentItem ON n.parentItemID = parentItem.itemID
        WHERE i.itemID = ?
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
    """
    with _open_db(self._sqlite_path) as conn:
        row = conn.execute(sql, (note_id,)).fetchone()
        if row is None:
            return None
        return ZoteroNote(
            note_id=row["noteID"],
            note_key=row["noteKey"],
            parent_item_id=row["parentItemID"],
            parent_item_key=row["parentKey"],
            title=row["noteTitle"],
            note_html=row["noteHtml"],
        )

get_note_by_key

get_note_by_key(note_key)

Fetch a single note by its Zotero item key.

Parameters:

Name	Type	Description	Default
note_key	str	The alphanumeric Zotero item key of the note.	required

Returns:

Type	Description
ZoteroNote | None	A ZoteroNote if found, or None if the key does not exist
ZoteroNote | None	or the item has been deleted.

Source code in zocrates/rag/zotero_db.py

def get_note_by_key(self, note_key: str) -> "ZoteroNote | None":
    """Fetch a single note by its Zotero item key.

    Args:
        note_key: The alphanumeric Zotero item key of the note.

    Returns:
        A ``ZoteroNote`` if found, or ``None`` if the key does not exist
        or the item has been deleted.
    """
    sql = """
        SELECT
            i.itemID AS noteID,
            i.key AS noteKey,
            n.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(n.note, '') AS noteHtml,
            COALESCE(n.title, '') AS noteTitle
        FROM items i
        JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
        JOIN itemNotes n ON i.itemID = n.itemID
        LEFT JOIN items parentItem ON n.parentItemID = parentItem.itemID
        WHERE i.key = ?
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
    """
    with _open_db(self._sqlite_path) as conn:
        row = conn.execute(sql, (note_key,)).fetchone()
        if row is None:
            return None
        return ZoteroNote(
            note_id=row["noteID"],
            note_key=row["noteKey"],
            parent_item_id=row["parentItemID"],
            parent_item_key=row["parentKey"],
            title=row["noteTitle"],
            note_html=row["noteHtml"],
        )

get_notes_for_item

get_notes_for_item(parent_item_id)

Return child notes attached to a specific parent item.

Parameters:

Name	Type	Description	Default
parent_item_id	int	Numeric ID of the parent Zotero item.	required

Returns:

Type	Description
list[ZoteroNote]	List of ZoteroNote objects whose parent_item_id matches.

Source code in zocrates/rag/zotero_db.py

def get_notes_for_item(self, parent_item_id: int) -> list[ZoteroNote]:
    """Return child notes attached to a specific parent item.

    Args:
        parent_item_id: Numeric ID of the parent Zotero item.

    Returns:
        List of ``ZoteroNote`` objects whose ``parent_item_id`` matches.
    """
    sql = """
        SELECT
            i.itemID AS noteID,
            i.key AS noteKey,
            n.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(n.note, '') AS noteHtml,
            COALESCE(n.title, '') AS noteTitle
        FROM items i
        JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
        JOIN itemNotes n ON i.itemID = n.itemID
        LEFT JOIN items parentItem ON n.parentItemID = parentItem.itemID
        WHERE it.typeName = 'note'
          AND n.parentItemID = ?
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
        ORDER BY i.itemID
    """
    with _open_db(self._sqlite_path) as conn:
        rows = conn.execute(sql, (parent_item_id,)).fetchall()

    notes: list[ZoteroNote] = []
    for row in rows:
        notes.append(
            ZoteroNote(
                note_id=row["noteID"],
                note_key=row["noteKey"],
                parent_item_id=row["parentItemID"],
                parent_item_key=row["parentKey"],
                title=row["noteTitle"],
                note_html=row["noteHtml"],
            )
        )
    return notes

get_stale_item_keys

get_stale_item_keys(converted_map, key_to_atts)

Identify cached items whose Zotero metadata has been modified since conversion.

Parameters:

Name	Type	Description	Default
converted_map	dict[str, str]	Mapping of attachment/note item_key -> converted_at ISO string.	required
key_to_atts	dict[str, list[str]]	Mapping of query key (parent_item_key or direct item_key) to list of associated attachment/note item_keys.	required

Returns:

Type	Description
list[str]	Sorted, unique list of attachment/note item_keys whose metadata is stale.

Source code in zocrates/rag/zotero_db.py

def get_stale_item_keys(
    self,
    converted_map: dict[str, str],
    key_to_atts: dict[str, list[str]],
) -> list[str]:
    """Identify cached items whose Zotero metadata has been modified since conversion.

    Args:
        converted_map: Mapping of attachment/note item_key -> converted_at ISO string.
        key_to_atts: Mapping of query key (parent_item_key or direct item_key)
            to list of associated attachment/note item_keys.

    Returns:
        Sorted, unique list of attachment/note item_keys whose metadata is stale.
    """
    if not key_to_atts or not converted_map:
        return []

    stale_item_keys: set[str] = set()
    query_keys = list(key_to_atts.keys())
    batch_size = 500

    with _open_db(self._sqlite_path) as conn:
        for i in range(0, len(query_keys), batch_size):
            batch = query_keys[i : i + batch_size]
            placeholders = ",".join("?" for _ in batch)
            sql = f"""
                SELECT key, COALESCE(clientDateModified, dateModified) AS modified
                FROM items
                WHERE key IN ({placeholders})
                  AND itemID NOT IN (SELECT itemID FROM deletedItems)
            """
            rows = conn.execute(sql, batch).fetchall()
            for row in rows:
                q_key = row["key"]
                mod_str = row["modified"]
                if not mod_str:
                    continue
                mod_dt = _parse_utc_datetime(mod_str)
                if not mod_dt:
                    continue

                for att_key in key_to_atts.get(q_key, []):
                    conv_str = converted_map.get(att_key)
                    if not conv_str:
                        continue
                    conv_dt = _parse_utc_datetime(conv_str)
                    if not conv_dt:
                        continue

                    if mod_dt > conv_dt:
                        stale_item_keys.add(att_key)

    return sorted(stale_item_keys)

get_standalone_attachments

get_standalone_attachments(content_types=('application/pdf', 'text/html', 'text/markdown', 'text/plain'))

Return standalone file attachments (attachments without a parent item).

Parameters:

Name	Type	Description	Default
content_types	tuple[str, ...]	Tuple of MIME types to include.	('application/pdf', 'text/html', 'text/markdown', 'text/plain')

Returns:

Type	Description
list[ZoteroAttachment]	List of ZoteroAttachment objects whose parent_item_id is None.

Source code in zocrates/rag/zotero_db.py

def get_standalone_attachments(
    self,
    content_types: tuple[str, ...] = ("application/pdf", "text/html", "text/markdown", "text/plain"),
) -> list[ZoteroAttachment]:
    """Return standalone file attachments (attachments without a parent item).

    Args:
        content_types: Tuple of MIME types to include.

    Returns:
        List of ``ZoteroAttachment`` objects whose ``parent_item_id`` is ``None``.
    """
    sql = """
        SELECT
            i.itemID,
            i.key,
            ia.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(ia.contentType, '') AS contentType,
            COALESCE(ia.path, '') AS path
        FROM items i
        JOIN itemAttachments ia ON i.itemID = ia.itemID
        LEFT JOIN items parentItem ON ia.parentItemID = parentItem.itemID
        WHERE ia.parentItemID IS NULL
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
    """
    with _open_db(self._sqlite_path) as conn:
        rows = conn.execute(sql).fetchall()

    attachments: list[ZoteroAttachment] = []
    for row in rows:
        if row["contentType"] not in content_types:
            continue
        resolved = self._resolve_attachment_path(row["key"], row["path"])
        attachments.append(
            ZoteroAttachment(
                item_id=row["itemID"],
                item_key=row["key"],
                parent_item_id=row["parentItemID"],
                parent_item_key=row["parentKey"],
                content_type=row["contentType"],
                path=resolved,
            )
        )
    return attachments

get_standalone_notes

get_standalone_notes()

Return top-level standalone notes (notes without a parent item).

Returns:

Type	Description
list[ZoteroNote]	List of ZoteroNote objects whose parent_item_id is None.

Source code in zocrates/rag/zotero_db.py

def get_standalone_notes(self) -> list[ZoteroNote]:
    """Return top-level standalone notes (notes without a parent item).

    Returns:
        List of ``ZoteroNote`` objects whose ``parent_item_id`` is ``None``.
    """
    sql = """
        SELECT
            i.itemID AS noteID,
            i.key AS noteKey,
            n.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(n.note, '') AS noteHtml,
            COALESCE(n.title, '') AS noteTitle
        FROM items i
        JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
        JOIN itemNotes n ON i.itemID = n.itemID
        LEFT JOIN items parentItem ON n.parentItemID = parentItem.itemID
        WHERE it.typeName = 'note'
          AND n.parentItemID IS NULL
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
        ORDER BY i.itemID
    """
    with _open_db(self._sqlite_path) as conn:
        rows = conn.execute(sql).fetchall()

    notes: list[ZoteroNote] = []
    for row in rows:
        notes.append(
            ZoteroNote(
                note_id=row["noteID"],
                note_key=row["noteKey"],
                parent_item_id=row["parentItemID"],
                parent_item_key=row["parentKey"],
                title=row["noteTitle"],
                note_html=row["noteHtml"],
            )
        )
    return notes

invalidate_schema_cache

invalidate_schema_cache()

Clear cached table existence checks to re-probe schema on next access.

Source code in zocrates/rag/zotero_db.py

def invalidate_schema_cache(self) -> None:
    """Clear cached table existence checks to re-probe schema on next access."""
    self._table_exists_cache.clear()

is_item_active

is_item_active(item_key)

Check if a single item key exists and is not deleted (or parent-deleted).

Parameters:

Name	Type	Description	Default
item_key	str	Zotero item key.	required

Returns:

Type	Description
bool	True if the item exists and is active, False otherwise.

Source code in zocrates/rag/zotero_db.py

def is_item_active(self, item_key: str) -> bool:
    """Check if a single item key exists and is not deleted (or parent-deleted).

    Args:
        item_key: Zotero item key.

    Returns:
        True if the item exists and is active, False otherwise.
    """
    if not item_key:
        return False
    return item_key in self.get_active_item_keys([item_key])

ZoteroItem dataclass

ZoteroItem(item_id, item_key, item_type, title, creators=list(), year=None, publication=None, doi=None, url=None, abstract=None, tags=list(), collections=list(), relations=list(), arxiv_id=None)

Lightweight representation of a Zotero library item.

ZoteroNote dataclass

ZoteroNote(note_id, note_key, parent_item_id, parent_item_key, title, note_html)

A Zotero note item (standalone or attached to a parent item).

_cleanup_sqlite_snapshots

_cleanup_sqlite_snapshots()

Remove all cached snapshot temp files (registered via atexit).

Source code in zocrates/rag/zotero_db.py

def _cleanup_sqlite_snapshots() -> None:
    """Remove all cached snapshot temp files (registered via atexit)."""
    acquired = _sqlite_snapshot_lock.acquire(timeout=1.0)
    try:
        for tmp_path, _sig in list(_sqlite_snapshots.values()):
            for suffix in ("", "-wal", "-shm", "-journal"):
                try:
                    os.remove(tmp_path + suffix)
                except OSError:
                    pass
            try:
                os.rmdir(os.path.dirname(tmp_path))
            except OSError:
                pass
        _sqlite_snapshots.clear()
    finally:
        if acquired:
            _sqlite_snapshot_lock.release()

_get_connection

_get_connection(sqlite_path)

Resolve and open a raw SQLite connection to the Zotero database.

Source code in zocrates/rag/zotero_db.py

def _get_connection(sqlite_path: Path) -> sqlite3.Connection:
    """Resolve and open a raw SQLite connection to the Zotero database."""
    with _sqlite_snapshot_lock:
        cached = _sqlite_snapshots.get(str(sqlite_path))
        if cached is not None and os.path.exists(cached[0]):
            try:
                sig = _sqlite_snapshot_signature(sqlite_path)
                if cached[1] == sig:
                    conn = sqlite3.connect(cached[0], timeout=5.0)
                    conn.row_factory = sqlite3.Row
                    return conn
            except (OSError, sqlite3.Error):
                pass

    from urllib.parse import quote

    encoded_path = quote(str(sqlite_path), safe="/:")
    try:
        conn = sqlite3.connect(
            f"file:{encoded_path}?mode=ro", uri=True, timeout=0.1
        )
        conn.row_factory = sqlite3.Row
        # Probe the read: a mode=ro connection that succeeds but cannot
        # query (edge lock states) should fall through to the snapshot tier
        # instead of raising deep inside a sync pipeline.
        try:
            conn.execute("SELECT 1 FROM sqlite_master LIMIT 1").fetchone()
            conn.execute("PRAGMA busy_timeout = 5000;")
            return conn
        except sqlite3.Error:
            conn.close()
    except sqlite3.Error:
        pass
    try:
        return _open_snapshot_connection(sqlite_path)
    except (OSError, sqlite3.Error):
        pass
    uri = f"file:{encoded_path}?immutable=1"
    conn = sqlite3.connect(uri, uri=True, timeout=1.0)
    conn.row_factory = sqlite3.Row
    return conn

_is_pid_alive

_is_pid_alive(pid)

Check whether a process with the given PID is currently alive.

Source code in zocrates/rag/zotero_db.py

def _is_pid_alive(pid: int) -> bool:
    """Check whether a process with the given PID is currently alive."""
    if pid <= 0:
        return False
    try:
        import psutil

        return psutil.pid_exists(pid)
    except Exception:
        pass
    if sys.platform == "win32":
        return True
    try:
        os.kill(pid, 0)
        return True
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    except OSError:
        return False

_open_db

_open_db(sqlite_path)

Open the Zotero SQLite DB with a read-correct, lock-safe strategy.

Tiered connection strategy (WAL-correct as of the fresh-profile fix):

1. mode=ro: the correct view whenever SQLite can take the shared read path — Zotero running in WAL mode (readers never block writers under WAL), or a cleanly closed database. A read-only connection sees committed WAL frames, so un-checkpointed data is visible.
2. Snapshot copy: when mode=ro fails (e.g. an unclean Zotero shutdown leaves an un-checkpointed -wal sidecar with no -shm — a read-only connection cannot recover the WAL and reports "database is locked"), the DB plus its -wal / -journal sidecars are copied to a private temp file and opened read-write so SQLite performs WAL recovery on the copy. The snapshot is cached per source path and invalidated whenever the main file mtime or WAL size changes (static while Zotero is closed, so the copy cost is paid at most once per Zotero session state).
3. immutable=1 last resort (legacy behavior): bypasses OS locks but silently ignores WAL journals — only reached when both strategies above fail, e.g. truly read-only filesystems.

immutable=1 alone is NOT sufficient in general: Zotero writes into zotero.sqlite-wal and only merges into the main file on checkpoint (typically at shutdown), so a fresh session's committed items can exist exclusively in the WAL. Reading the main file with immutable=1 then reports an empty library even though the user sees items in Zotero.

The file path is URL-encoded because Zotero data directories may contain spaces or special characters that break SQLite URI parameter parsing.

Must be used as a context manager (with _open_db(path) as conn:). The connection and OS file locks are deterministically closed on exit.

Source code in zocrates/rag/zotero_db.py

@contextmanager
def _open_db(sqlite_path: Path) -> Iterator[sqlite3.Connection]:
    """Open the Zotero SQLite DB with a read-correct, lock-safe strategy.

    Tiered connection strategy (WAL-correct as of the fresh-profile fix):

    1. ``mode=ro``: the correct view whenever SQLite can take the shared
       read path — Zotero running in WAL mode (readers never block writers
       under WAL), or a cleanly closed database. A read-only connection
       sees committed WAL frames, so un-checkpointed data is visible.
    2. Snapshot copy: when ``mode=ro`` fails (e.g. an unclean Zotero
       shutdown leaves an un-checkpointed ``-wal`` sidecar with no ``-shm``
       — a read-only connection cannot recover the WAL and reports
       "database is locked"), the DB plus its ``-wal`` / ``-journal``
       sidecars are copied to a private temp file and opened read-write so
       SQLite performs WAL recovery on the copy. The snapshot is cached per
       source path and invalidated whenever the main file mtime or WAL size
       changes (static while Zotero is closed, so the copy cost is paid at
       most once per Zotero session state).
    3. ``immutable=1`` last resort (legacy behavior): bypasses OS locks but
       silently ignores WAL journals — only reached when both strategies
       above fail, e.g. truly read-only filesystems.

    ``immutable=1`` alone is NOT sufficient in general: Zotero writes into
    ``zotero.sqlite-wal`` and only merges into the main file on checkpoint
    (typically at shutdown), so a fresh session's committed items can exist
    exclusively in the WAL. Reading the main file with ``immutable=1`` then
    reports an empty library even though the user sees items in Zotero.

    The file path is URL-encoded because Zotero data directories may contain
    spaces or special characters that break SQLite URI parameter parsing.

    Must be used as a context manager (``with _open_db(path) as conn:``).
    The connection and OS file locks are deterministically closed on exit.
    """
    conn = _get_connection(sqlite_path)
    try:
        yield conn
    finally:
        conn.close()

_open_snapshot_connection

_open_snapshot_connection(sqlite_path)

Open a private recovered copy of the Zotero DB (WAL fallback tier).

Copies the main DB plus -wal / -journal sidecars into a private temp directory and opens the copy read-write, letting SQLite recover committed WAL frames on open. Cached per source path until the main file mtime or WAL size changes (the fallback regime is by definition a static, Zotero-closed state, so the copy cost is paid once).

Source code in zocrates/rag/zotero_db.py

def _open_snapshot_connection(sqlite_path: Path) -> sqlite3.Connection:
    """Open a private recovered copy of the Zotero DB (WAL fallback tier).

    Copies the main DB plus ``-wal`` / ``-journal`` sidecars into a private
    temp directory and opens the copy read-write, letting SQLite recover
    committed WAL frames on open. Cached per source path until the main
    file mtime or WAL size changes (the fallback regime is by definition a
    static, Zotero-closed state, so the copy cost is paid once).
    """
    global _swept_stale_snapshots
    with _sqlite_snapshot_lock:
        if not _swept_stale_snapshots:
            _swept_stale_snapshots = True
            _sweep_stale_sqlite_snapshots()
        sig = _sqlite_snapshot_signature(sqlite_path)
        cached = _sqlite_snapshots.get(str(sqlite_path))
        if cached is not None and cached[1] == sig:
            conn = sqlite3.connect(cached[0], timeout=5.0)
            conn.row_factory = sqlite3.Row
            return conn
        tmp_dir = tempfile.mkdtemp(prefix=f"zocrates-zdb-{os.getpid()}-")
        tmp_path = os.path.join(tmp_dir, "zotero.sqlite")
        conn: sqlite3.Connection | None = None
        try:
            shutil.copy2(sqlite_path, tmp_path)
            for suffix in ("-wal", "-journal"):
                sidecar = Path(str(sqlite_path) + suffix)
                try:
                    if sidecar.exists():
                        shutil.copy2(sidecar, tmp_path + suffix)
                except FileNotFoundError:
                    pass
            conn = sqlite3.connect(tmp_path, timeout=5.0)
            conn.row_factory = sqlite3.Row
            conn.execute("SELECT 1 FROM sqlite_master LIMIT 1").fetchone()
        except BaseException:
            if conn is not None:
                try:
                    conn.close()
                except Exception:
                    pass
            shutil.rmtree(tmp_dir, ignore_errors=True)
            raise

        # Only after the new snapshot is copied and verified:
        if cached is not None:
            for suffix in ("", "-wal", "-shm", "-journal"):
                try:
                    os.remove(cached[0] + suffix)
                except OSError:
                    pass
            try:
                os.rmdir(os.path.dirname(cached[0]))
            except OSError:
                pass
        _sqlite_snapshots[str(sqlite_path)] = (tmp_path, sig)
        return conn

_parse_utc_datetime

_parse_utc_datetime(ts_str)

Parse a date string into a timezone-aware UTC datetime.

Source code in zocrates/rag/zotero_db.py

def _parse_utc_datetime(ts_str: str | None) -> datetime | None:
    """Parse a date string into a timezone-aware UTC datetime."""
    if not ts_str:
        return None
    try:
        ts_clean = ts_str.strip()
        if ts_clean.endswith("Z"):
            ts_clean = ts_clean[:-1] + "+00:00"
        dt = datetime.fromisoformat(ts_clean)
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        else:
            dt = dt.astimezone(timezone.utc)
        return dt
    except Exception:
        return None

_sqlite_snapshot_signature

_sqlite_snapshot_signature(sqlite_path)

Signature used to invalidate cached snapshots: main mtime + WAL size + WAL mtime.

Source code in zocrates/rag/zotero_db.py

def _sqlite_snapshot_signature(sqlite_path: Path) -> tuple[int, int, int]:
    """Signature used to invalidate cached snapshots: main mtime + WAL size + WAL mtime."""
    st = sqlite_path.stat()
    wal_path = Path(str(sqlite_path) + "-wal")
    try:
        wal_st = wal_path.stat()
        wal_size = wal_st.st_size
        wal_mtime = wal_st.st_mtime_ns
    except FileNotFoundError:
        wal_size = 0
        wal_mtime = 0
    return (st.st_mtime_ns, wal_size, wal_mtime)

_sweep_stale_sqlite_snapshots

_sweep_stale_sqlite_snapshots()

Scan and clean up orphaned snapshot temp directories from dead processes or legacy runs.

Source code in zocrates/rag/zotero_db.py

def _sweep_stale_sqlite_snapshots() -> None:
    """Scan and clean up orphaned snapshot temp directories from dead processes or legacy runs."""
    try:
        temp_dir = Path(tempfile.gettempdir())
        now = time.time()
        for entry in temp_dir.iterdir():
            try:
                if not entry.is_dir() or not entry.name.startswith("zocrates-zdb-"):
                    continue
                try:
                    st = entry.stat()
                except OSError:
                    continue
                if hasattr(st, "st_uid") and hasattr(os, "getuid") and st.st_uid != os.getuid():
                    continue
                m = re.match(r"^zocrates-zdb-(\d+)-", entry.name)
                if m:
                    pid = int(m.group(1))
                    if not _is_pid_alive(pid):
                        shutil.rmtree(entry, ignore_errors=True)
                else:
                    # Legacy snapshot directory without PID segment
                    try:
                        if now - st.st_mtime > 86400:
                            shutil.rmtree(entry, ignore_errors=True)
                    except OSError:
                        pass
            except (OSError, PermissionError):
                pass
    except Exception:
        pass

extract_arxiv_id

extract_arxiv_id(extra=None, url=None, doi=None)

Extract and normalize an arXiv ID from extra text, URL, or DOI.

Source code in zocrates/rag/zotero_db.py

def extract_arxiv_id(
    extra: str | None = None,
    url: str | None = None,
    doi: str | None = None,
) -> str | None:
    """Extract and normalize an arXiv ID from extra text, URL, or DOI."""
    if extra:
        m = re.search(
            r"(?i)\barxiv:\s*([a-zA-Z\-]+(?:\.[a-zA-Z\-]+)?\/\d{7}|\d{4}\.\d{4,5}(?:v\d+)?)",
            extra,
        )
        if m:
            return normalize_arxiv_id(m.group(1))
        m_fallback = re.search(r"(?i)\barxiv:\s*(\S+)", extra)
        if m_fallback:
            cand = normalize_arxiv_id(m_fallback.group(1))
            if cand:
                return cand

    if url:
        cand = normalize_arxiv_id(url)
        if cand:
            return cand

    if doi:
        cand = normalize_arxiv_id(doi)
        if cand:
            return cand

    return None

extract_year

extract_year(date_val)

Extract a 4-digit year string from a Zotero date value.

Source code in zocrates/rag/zotero_db.py

def extract_year(date_val: str | None) -> str | None:
    """Extract a 4-digit year string from a Zotero date value."""
    if not date_val:
        return None
    s = date_val.strip()
    m = re.search(r"\b(1[89]\d{2}|20\d{2})\b", s)
    if m:
        return m.group(1)
    if len(s) >= 4 and s[:4].isdigit():
        return s[:4]
    return None

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