mem0-python-server/mem0core/storage.py

import sqlite3

from .config import SQLITE_PATH


def sqlite_delete_ids(memory_ids: list[str]) -> int:
    """Delete history rows by memory ID from SQLite and return deleted count."""
    if not memory_ids:
        return 0
    try:
        conn = sqlite3.connect(SQLITE_PATH)
        cur = conn.cursor()
        placeholders = ",".join("?" * len(memory_ids))
        cur.execute(f"DELETE FROM history WHERE memory_id IN ({placeholders})", memory_ids)
        deleted = cur.rowcount
        conn.commit()
        conn.close()
        return deleted
    except Exception as exc:
        print(f"[sqlite] warning: {exc}")
        return 0


def chroma_get_all(collection, user_id: str, include: list | None = None) -> list[dict]:
    """Page through a Chroma collection and return all rows for one user."""
    if include is None:
        include = ["metadatas"]

    results = []
    batch = 500
    offset = 0

    while True:
        page = collection.get(
            where={"user_id": {"$eq": user_id}},
            limit=batch,
            offset=offset,
            include=include,
        )
        ids = page.get("ids", [])
        if not ids:
            break

        for i, item_id in enumerate(ids):
            row = {"id": item_id}
            for field in include:
                values = page.get(field, [])
                row[field[:-1]] = values[i] if i < len(values) else None
            results.append(row)

        offset += len(ids)
        if len(ids) < batch:
            break

    return results