Gmail Ingestion

import asyncio
import os

import cognee
from cognee.tasks.ingestion.connectors import gmail_source

# Keep the inbox in its own dataset so it is easy to inspect and forget.
DATASET_NAME = "gmail_inbox"

# Routing kwargs for remember(). Pass the same ones on every later sync.
#   write_disposition="merge" is REQUIRED: the add pipeline defaults to
#     "replace", which would wipe the whole synced inbox on the next sync.
#   max_rows_per_table=0 guarantees no per-table read cap applies (even if
#     DLT_MAX_ROWS_PER_TABLE is set), so orphan-cleanup (forget-on-delete)
#     compares against the *entire* synced corpus.
#   incremental_loading is left at its default (True) so a re-sync only builds
#     the graph for new or changed messages instead of the whole inbox again.
GMAIL_REMEMBER_KWARGS = {
    "primary_key": "id",
    "write_disposition": "merge",
    "max_rows_per_table": 0,
    "self_improvement": False,
}


async def main():
    credentials_path = os.environ.get("GMAIL_CREDENTIALS_PATH", "credentials.json")
    token_path = os.environ.get("GMAIL_TOKEN_PATH", "token.json")

    if not os.path.exists(credentials_path):
        print(
            f"Gmail OAuth client secrets not found at '{credentials_path}'.\n"
            "See the setup steps in this file's docstring, then re-run."
        )
        return

    # Start from a clean slate so the demo is reproducible.
    await cognee.prune.prune_data()
    await cognee.prune.prune_system(metadata=True)

    # Build the source. Scope it to INBOX and, for the demo, load only the 25
    # newest messages so the first run is quick. Drop ``max_results`` to load
    # the whole label; Gmail's quota allows roughly 250 messages a minute.
    source = gmail_source(
        credentials_path=credentials_path,
        token_path=token_path,
        label_ids=["INBOX"],
        max_results=25,
    )

    # ── Sync: load the newest messages ────────────────────────────────────
    print("\n=== Gmail sync ===")
    result = await cognee.remember(
        source,
        dataset_name=DATASET_NAME,
        **GMAIL_REMEMBER_KWARGS,
    )
    print(result)
    print("Sync stats:", source.cognee_sync_stats)

    answer = await cognee.recall(
        query_text="Summarize the most important emails in my inbox.",
        query_type=cognee.SearchType.GRAPH_COMPLETION,
        datasets=[DATASET_NAME],
    )
    print(answer[0].text)


if __name__ == "__main__":
    asyncio.run(main())

GMAIL_REMEMBER_KWARGS = {
    "primary_key": "id",
    "write_disposition": "merge",
    "max_rows_per_table": 0,
    "self_improvement": False,
}

    source = gmail_source(
        credentials_path=credentials_path,
        token_path=token_path,
        label_ids=["INBOX"],
        max_results=25,
    )

    result = await cognee.remember(
        source,
        dataset_name=DATASET_NAME,
        **GMAIL_REMEMBER_KWARGS,
    )
    print(result)
    print("Sync stats:", source.cognee_sync_stats)

    answer = await cognee.recall(
        query_text="Summarize the most important emails in my inbox.",
        query_type=cognee.SearchType.GRAPH_COMPLETION,
        datasets=[DATASET_NAME],
    )
    print(answer[0].text)

source = gmail_source(label_ids=["INBOX"])
await cognee.remember(source, dataset_name="gmail_inbox", **GMAIL_REMEMBER_KWARGS)
print(source.cognee_sync_stats)  # e.g. {'scanned': 3, 'skipped': 0, 'failed': 0, 'deleted': 1, ...}

source = gmail_source(label_ids=["INBOX"])
await cognee.remember(source, dataset_name="gmail_inbox", **GMAIL_REMEMBER_KWARGS)