dlt (Data Load Tool)

Ingest structured data into Cognee with dlt.

Ingest structured relational data — databases, CSV files, and dlt resources — directly into cognee’s knowledge graph. Foreign keys become graph edges, tables become schema nodes, and each row becomes a searchable document, all built deterministically from the schema without LLM extraction.

Why Use This Integration

• Schema-Aware Graphs: Foreign key relationships are preserved as first-class edges in the knowledge graph
• Deterministic Graph Construction: Structured data bypasses LLM entity extraction — no hallucination risk
• Mixed Ingestion: Combine structured (dlt) and unstructured (text, PDF) data in the same dataset
• Multiple Input Modes: Pass explicit dlt resources, CSV file paths, or database connection strings
• Write Dispositions: Control how data is synced — merge (upsert), append, or replace

Installation

Nothing to install. dlt[sqlalchemy] and pandas are core cognee dependencies, so every route on this page works on a plain install:

pip install cognee

Or with uv:

uv pip install cognee

cognee[dlt] is kept as an empty extra, so existing install commands, Dockerfiles, and CI jobs that name it stay valid — it simply resolves to nothing. The same applies to the official cognee/cognee Docker image: CSVs added there follow the dlt_csv_loader path described under CSV Files, as they do everywhere else. See Optional Extras and Document Loaders.

Quick Start

1. Ingest a dlt Resource

Define a dlt resource and pass it to cognee.remember(). The dlt-specific structured-ingestion options primary_key, write_disposition, SQL query, and max_rows_per_table are accepted by cognee.remember() and forwarded to the underlying ingestion step. After ingestion, use cognee.recall(...) to query the graph.

import dlt
import cognee
import asyncio

@dlt.resource()
def users_and_pets():
    yield [
        {
            "id": 1,
            "name": "Alice",
            "pets": [
                {"id": 1, "name": "Fluffy", "type": "cat"},
                {"id": 2, "name": "Spot", "type": "dog"},
            ],
        },
        {
            "id": 2,
            "name": "Bob",
            "pets": [{"id": 3, "name": "Fido", "type": "dog"}],
        },
    ]

async def main():
    await cognee.remember(
        users_and_pets,
        dataset_name="users_and_pets",
        primary_key="id",
    )
    results = await cognee.recall(
        query_text="Which pet does Alice have?",
        datasets=["users_and_pets"],
    )
    print(results)

asyncio.run(main())

dlt automatically detects nested structures (like pets inside each user) and creates separate tables with foreign key relationships.

The lower-level cognee.add(...) + cognee.cognify(...) pair still accepts the same dlt kwargs and remains useful when you need to run ingestion and graph building as separate steps. For the runnable end-to-end version of this walkthrough, see examples/demos/ingestion_and_migration/dlt_ingestion_example.py.

2. Build and Query the Graph

Once remember() finishes ingesting and building the graph, use cognee.recall(...) to query it.

Other Input Modes

CSV Files

CSV files are handled by the dlt_csv_loader, which the loader engine registers above the plain-text csv_loader on every install. Selection happens inside the ingest pipeline rather than before it, so the same route applies to every way a CSV can arrive — a local path, a file:// or s3:// location, or an uploaded file:

await cognee.remember(
    "/path/to/employees.csv",
    dataset_name="employees",
)

Each CSV is staged through dlt and becomes one manifest record per original file. The manifest’s identity is derived from the original file name — not the temporary copy an s3:// download or an upload lands in — so it stays stable across runs and a re-add updates the existing record instead of creating a second one. As with any dlt source, the rows skip chunking and LLM entity extraction.

Delimiter detection

The delimiter is detected per file, so semicolon-, tab-, and pipe-separated exports load as proper tables instead of collapsing into a single column — or, when a text field contains commas of its own, failing outright on a ragged split. No configuration is required and there is no knob to set.

Cognee reads up to the first 20 lines of the file, ignoring blank ones, and parses them with each of ,, ;, \t, | in that order, picking the first delimiter that splits every sampled line into the same number of columns, with more than one column. Parsing (rather than counting characters) means quoted fields are respected, so a comma-heavy text column — prose, an embedded JSON list — no longer forces a semicolon-delimited file onto commas. Candidates are tried in order, so a comma wins ties, and , is also the fallback when the file is empty or no candidate produces a consistent column count.

If a genuinely ambiguous file — one where a second candidate also yields a consistent column count — loads with the wrong delimiter, convert it to comma-separated before ingesting, or request the plain-text csv_loader for that call (below).

The DLT options given to remember() / add() in dlt_config reach CSV files too, as do bare keyword arguments (add() takes all six; remember() takes only primary_key, write_disposition, query and max_rows_per_table). Cognee folds them into the dlt_csv_loader entry of preferred_loaders:

await cognee.remember(
    "/path/to/employees.csv",
    dataset_name="employees",
    dlt_config={"primary_key": "id", "write_disposition": "merge"},
)

You can still set the options under the loader directly. dlt_csv_loader accepts primary_key, write_disposition, max_rows_per_table, column_value_columns, and temporal_columns:

await cognee.remember(
    "/path/to/employees.csv",
    dataset_name="employees",
    preferred_loaders=[
        {"dlt_csv_loader": {"primary_key": "id", "write_disposition": "merge"}}
    ],
)

If you set an option both on the call and under preferred_loaders["dlt_csv_loader"], cognee raises a ValueError instead of picking one.

To flatten a CSV into plain text instead, request csv_loader explicitly for that call. Since dlt became a core dependency there is no install shape that gives you text flattening by default, so this per-call override is the only way to get it:

await cognee.remember(
    "/path/to/employees.csv",
    dataset_name="employees",
    preferred_loaders=[{"csv_loader": {}}],
)

A CSV that yields no rows now fails loudly with an IngestionError naming the source, rather than being ingested as an empty record. dlt_csv_loader likewise requires the dataset and user context that the ingest pipeline supplies, so it cannot be invoked outside of remember() / add().

Database Connection String

Ingest tables directly from an existing database:

await cognee.remember(
    "postgresql://user:pass@host/db",
    dataset_name="company_db",
    primary_key="id",
)

The connection string is a source cognee reads from, never a destination it writes to — ingested rows land in cognee’s own configured stores, including vector embeddings of each row, so they’re searchable by semantic similarity as well as graph traversal. To change where cognee stores memory, configure the graph, vector, and relational providers.

Supported databases via auto-detection: SQLite, PostgreSQL, MySQL, MSSQL, Oracle. Hosted Postgres providers such as Neon work with their standard postgresql:// connection strings; keep provider-required SSL parameters such as ?sslmode=require. Amazon Redshift is also compatible since it speaks the PostgreSQL wire protocol — use a standard postgresql:// connection string pointing to your Redshift endpoint.

For Snowflake and Google BigQuery, construct a dlt source directly and pass it to cognee.remember() (see the Cloud Data Warehouses accordion below).

You can optionally filter with a SQL WHERE clause:

await cognee.remember(
    "postgresql://user:pass@host/db",
    dataset_name="engineering_team",
    primary_key="id",
    query="SELECT * FROM employees WHERE department = 'Engineering'",
)

The FROM target may also be schema-qualified. Cognee splits schema.table before building the source — the schema is handed to dlt and the bare table name is what the WHERE clause is matched against — so the filter applies to the qualified table just as it does to a bare one:

await cognee.remember(
    "postgresql://user:pass@host/db",
    dataset_name="engineering_team",
    primary_key="id",
    query="SELECT * FROM public.employees WHERE department = 'Engineering'",
)

The FROM target may also carry a table alias, with or without AS: query="SELECT * FROM employees e WHERE department = 'Engineering'" filters exactly like the unaliased form above; previously the alias broke the parse and the whole table was loaded unfiltered. The WHERE clause is applied as written as long as it does not reference the alias — leave columns unqualified, or qualify them with the table name.

Two query shapes cannot be replayed against the single-table select cognee hands to dlt, so they raise a ValueError instead of loading the table unfiltered:

• A WHERE that references the table alias — SELECT * FROM employees e WHERE e.department = 'Engineering'. The filter is replayed against the bare table, where the alias does not exist. Rewrite the columns bare (department = 'Engineering') or qualify them with the table name (employees.department = 'Engineering'). The check is a textual scan for <alias>.<column> across the whole clause, so a string literal that happens to match trips it too — under the alias f, WHERE path = 'f.txt' raises even though no column is qualified. A longer alias (emp rather than e) avoids the collision.
• A JOIN in the FROM clause — SELECT o.id FROM orders o JOIN customers c ON o.customer_id = c.id WHERE c.active. The source targets a single table, so a join across tables has no representation. Create a database view over the joined result and ingest that view, or ingest the table without a filter. Only the JOIN keyword is recognized: a comma-style join (SELECT * FROM orders o, customers c WHERE c.active) still parses as a plain select on the first table with its WHERE dropped, loading that table unfiltered — write joins with JOIN so the error fires, or ingest a view.

query accepts a SELECT with an unquoted FROM target — either a bare table name (employees) or a dot-qualified one (public.employees) — an optional table alias (employees e / employees AS e), and an optional WHERE clause. Quoted identifiers such as "public"."employees" are not parsed and raise a ValueError. Omitting WHERE loads the whole table.

Mixed Structured + Unstructured

Combine dlt resources with unstructured text in a single dataset:

text = """Alice has two pets: a cat named Fluffy and a dog named Spot.
Bob has a dog named Fido, who is friendly with both Fluffy and Spot."""

await cognee.remember(
    [text, users_and_pets],
    dataset_name="users_and_pets_with_text",
    primary_key="id",
)

Structured data creates deterministic graph nodes from the schema, while unstructured text goes through LLM-based entity extraction. Both are combined in the same knowledge graph.

DLT Options

Options that apply only to dlt sources (connection strings, dlt resources, and CSV files) can be grouped in one dlt_config dict on add(). remember() passes it through to add():

await cognee.remember(
    "postgresql://user:pass@host/db",
    dataset_name="company_db",
    dlt_config={
        "primary_key": "id",
        "write_disposition": "merge",
        "max_rows_per_table": 10_000,
        "temporal_columns": {"orders": ["order_date"], "*": ["*"]},
    },
)

Key	Type	What it does
primary_key	str	Upsert key. See Write Dispositions.
write_disposition	str	replace, merge, or append.
query	str	SQL SELECT … WHERE filter for a connection string.
max_rows_per_table	int	Per-table row cap for this call. See Bounding Ingestion.
column_value_columns	dict[str, list[str]]	Columns whose cells become shared ColumnValue nodes.
temporal_columns	dict[str, list[str]]	Columns whose date cells become Timestamp edges. See Time Anchoring.

• The read options primary_key, write_disposition and max_rows_per_table apply to every dlt source, connectors included, while query applies only to connection strings. The exception is that document-mode connector sources always read their whole table and ignore max_rows_per_table. Document-mode connector sources also ignore the two column selections (column_value_columns, temporal_columns).
• A key you leave out falls back to its default: DLT_MAX_ROWS_PER_TABLE, DLT_COLUMN_VALUE_COLUMNS, or DLT_TEMPORAL_COLUMNS for the three keys that have an environment variable, and the built-in default for the others.
• The dict is validated. An unknown key raises a pydantic ValidationError.
• The bare keyword arguments (primary_key="id", …) are still accepted for compatibility. If you give the same option both in dlt_config and as a keyword argument, cognee raises a ValueError instead of merging them.

Time Anchoring

Cognee anchors dlt rows in time by default. When a cell in a selected column holds an ISO 8601 month (2024-03), date (2024-03-01), or date-time (2024-03-01 10:15:00), the row gets an edge to the shared Timestamp node for that period. The edge is named after the column. Prose, relative phrases, and bare years are never read from a cell. Foreign-key columns are skipped, because they already link rows. A date primary key is kept, because a time series is keyed by the date it is about. Fractional seconds are dropped. A date-time with a zone (Z, +02:00) is shifted to UTC.

The Timestamp node is the same one that text extraction produces for that instant. A row dated 2024-03-01 and a document that says “1 March 2024” therefore share one node.

The temporal_columns selection decides which columns count. It has the same shape as column_value_columns: a map from table name to column names, with "*" as a wildcard on either side. A table you name gets exactly its list. An empty list turns that table off. Only tables you don’t name fall back to the "*" entry.

• Default {"*": ["*"]}: every date-shaped cell of every table.
• Narrow it when a table has bookkeeping dates. For example, {"orders": ["order_date"], "*": ["*"]} stops an updated_at column from anchoring orders rows to the time they were revised.
• {} turns time edges off.

Set it per call with dlt_config={"temporal_columns": {...}} (or the bare temporal_columns= keyword on add()). To change the process-wide default, set the DLT_TEMPORAL_COLUMNS environment variable as JSON:

DLT_TEMPORAL_COLUMNS='{"orders": ["order_date"], "audit_log": [], "*": ["*"]}'

This is on by default. dlt sources you ingest or re-ingest after upgrading get Timestamp edges for their date cells, which can change what TEMPORAL search returns. Set DLT_TEMPORAL_COLUMNS='{}' to keep building graphs without them.

Rows in Hybrid and TEMPORAL Search

When a dataset holds dlt rows, the hybrid chunk lane also searches the row vectors (the DltRow collection) next to document chunks. This applies to HYBRID_COMPLETION and TEMPORAL. In TEMPORAL search, rows anchored to Timestamp nodes inside the question’s time window join the candidates and rank ahead of undated ones; only rows dated more coarsely than the question rank lower. A question like “which orders were placed in March 2024?” can therefore reach the matching rows even when their text alone does not match the question well.

Write Dispositions

Control how data is synced on repeated runs using the write_disposition parameter:

• replace (default): Drop and recreate tables on each run. Use for full snapshot refreshes.
• merge: Upsert by primary key — updates existing rows, inserts new ones. Best for data that changes over time.
• append: Always insert without deduplication. Use for time-series data and event logs.

# Append mode — every call adds new rows, no dedup
await cognee.remember(
    event_resource,
    dataset_name="events",
    primary_key="id",
    write_disposition="append",
)

write_disposition only controls dlt’s staging snapshot. It does not by itself make a re-run pick up rows a source has gained, because the ingestion skip in add() short-circuits an already-ingested source first — see Re-Ingesting a Source and Re-ingesting a source that keeps growing.

How It Works

1. Source Detection: cognee identifies dlt resources and connection strings in the input. .csv files are not detected here — they are routed by the loader engine’s dlt_csv_loader inside the ingest pipeline (see CSV Files) and join the flow from step 2 onward
2. Pipeline Execution: A dlt pipeline loads data into a per-dataset staging database
3. Schema Extraction: Table schemas, primary keys, and foreign keys are extracted
4. Graph Construction: Each row becomes a document node; foreign keys become edges between nodes, and date-shaped cells become edges to shared Timestamp nodes (see Time Anchoring)
5. LLM Bypass: Structured rows skip chunking, entity extraction, and summarization — the graph is built entirely from schema metadata and cell values

The primary_key parameter controls upsert behavior when you use write_disposition="merge". If not specified, cognee auto-detects from an id column or falls back to the first column.

Bounding Ingestion

Two opt-in caps guard against large sources; both default to 0, meaning no cap.

• Rows per table: pass the max_rows_per_table kwarg to remember() / add() to bound the per-table row count for a single call, or set the DLT_MAX_ROWS_PER_TABLE environment variable to change the process-wide default.
• Column-value length: DLT_MAX_COLUMN_VALUE_LENGTH bounds the length of the cell values that become shared ColumnValue nodes — cell-level graph nodes that link rows sharing the same value in a selected column. (Columns are selected with column_value_columns — in dlt_config on remember() / add(), or as a bare add() kwarg — or with DLT_COLUMN_VALUE_COLUMNS; nothing is selected by default.) A positive value skips selected cells longer than that many characters, dropping rather than truncating them — worthwhile for free-text-heavy columns or wildcard ("*") selection, since long one-off values make poor shared nodes and each unique value costs one embedding. Unlike max_rows_per_table, this cap has no per-call kwarg: set the environment variable or the dlt_max_column_value_length ingestion config field. Cognee previously applied a fixed 256-character cap unconditionally; set DLT_MAX_COLUMN_VALUE_LENGTH=256 to keep that behavior.

Re-Ingesting a Source

Each relational dlt source is stored as a single record whose identity is stable: it is derived from the dataset name and the source name (dlt_source:{dataset_name}:{source_name}) and does not depend on the data. Re-running remember() / add() on the same source therefore never creates a second copy — but what happens to the existing record depends on how you call it:

• Plain re-add (the default): add() is idempotent. A source that has already been ingested keeps its record, content hash, and cognify status as-is — whether or not the upstream data changed — and nothing is reprocessed.
• Explicit re-ingest: to pick up upstream changes, call add(..., incremental_loading=False, data_cache=False) — the completed-skip runs whenever either flag is on, so both must be off — or use update() with the record’s UUID. The record then updates in place under the same identity (a content hash over the source’s tables and rows tracks the change), so the source is never absent from the store, and the next cognify() purges the source’s previously derived artifacts from the graph and vector stores before re-emitting the current rows. Rows deleted upstream disappear, and changed rows do not keep their stale values alongside the new ones. Between the re-ingesting add() and cognify(), searches still return the source’s previous rows; they are replaced only once the re-cognify completes.

The purge is a real delete, re-authorized as one: re-ingesting a changed dlt source requires delete permission on the dataset. If that permission is missing, the run fails instead of continuing — silently skipping the purge would leave stale rows in the graph and present them as current.

The purge also reaches the session cache: cached session turns whose answers used the purged graph elements are removed, along with the feedback and session-context lessons distilled from them, so a re-ingest cannot replay the source’s old rows out of session memory. This part is best-effort — a session-cache error is logged as a warning and never fails the run. See Invalidation when the underlying data is deleted for how the targeted pass decides what is contaminated.

Renaming a Source

Because the identity is keyed on the dataset and source names, renaming either one is a remove + add, not an in-place rename. The new name ingests from scratch as a fresh source, and the records under the old name stay in the dataset — a re-ingest only reconciles the source names it just ingested, so it will not treat the old name’s records as orphans. Delete them explicitly (for example with cognee.forget(...)) if you do not want both.

For the same reason, two dlt sources that share a name within a single add() call would resolve to the same identity. cognee raises an error rather than letting one silently overwrite the other, so give each source a distinct name.

CSVs are keyed the same way, but through dlt_csv_loader rather than that check: a CSV’s source name is its file-name stem, with runs of characters other than letters, digits, and underscores replaced by _, leading and trailing separators stripped, and the result lowercased. Two CSVs whose stems normalize to the same string (for example Employees 2024.csv and employees_2024.csv) therefore share one manifest record within a dataset, with the later ingest updating the earlier one. Give CSV files names that stay distinct under that normalization.

Foreign Key Resolution

A foreign key becomes a graph edge only when both the source row and the target row are loaded in the same ingestion run. Two edge cases are worth knowing about — cognee now logs a warning in each so they are diagnosable rather than silent:

• Target row not loaded: if a foreign key points at a row that wasn’t ingested — most commonly because the target table hit a max_rows_per_table cap you set — the reference is dropped and no edge is created. The warning identifies the dropped references as source_table.column -> ref_table:value. If you see missing edges, raise max_rows_per_table so the referenced rows are included.
• Duplicate primary keys within a table: if multiple rows in a table share the same primary key, foreign key edges that target that key resolve to the last such row loaded; earlier rows with the same key are shadowed for FK targeting. The warning names the affected table and pk.

Connectors

Connectors are dlt sources for a specific system. The list below keeps the current connector packages visible; the routing details are tucked away for reference.

Connector Modes

Cognee supports two DLT connector modes:

• Relational connectors take the default dlt path described above: each row becomes a schema-context document and foreign keys become edges, all built deterministically without LLM extraction.
• Document-mode connectors opt each row into normal cognify instead: the row is turned into a text document that goes through LLM entity extraction, just like unstructured text passed to remember().

A dlt source opts into document mode by setting the cognee_document_source attribute (via the document_source_tag() helper in cognee.tasks.ingestion.dlt_utils) to a short source tag. cognee then routes every row from that source through cognify rather than the relational schema-context path:

• Each row is built from its title and content columns (rendered as # {title}\n\n{content}, or just the content when there is no title), with optional url and id columns preserved in metadata.
• system_metadata["source"] is set to the connector’s own tag (for example "notion") instead of "dlt", alongside title and, when present, url and external_id (from the row’s id).
• The rendered row text is stored literally. A row whose content happens to be nothing but a URL or an existing local file path is kept as that text — cognee never fetches or reads it the way it would a string you passed to add() yourself — and such a row is never recorded as having originated from that location. This applies to every document-mode row, titled or not, because the content comes from the provider rather than from the caller.

Because the tag is connector-provided, the shared ingestion engine stays connector-agnostic: a connector declares its own nature rather than being hard-coded by name.

Per-row node sets. A document-mode row can name its own node sets in the reserved cognee_node_set column (NODE_SET_COLUMN in cognee.tasks.ingestion.dlt_utils). The column must hold a JSON list of names; cognee applies a dlt json type hint to it on every document source, so the list stays on the row instead of becoming a child table.

• Each name is prefixed with the source tag (<source_tag>:) unless it already starts with it, so provider data can never target cognee’s own node sets such as skills or user_context. Names that resolve to the same node set (for example Project A and project_a) are deduplicated, and the first spelling is kept.
• The row’s node sets are merged with the node_set you pass to add() / remember(), so the document, its chunks, and the entities extracted from it land in both, and node_name filtering in recall works per row. The Notion connector, for example, tags each page with ["notion:<workspace_id>:<root_id>"].
• An unset value (None or [], which read back as NULL, "[]", or "null") is left out of the row’s content hash, so adding the column without any tags does not change existing document ids or re-ingest rows. Do not emit an empty string: dlt stores it in a variant column and the sync stops. Changing a row’s actual node sets re-ingests it like any other column change.

Any other shape stops the whole sync: a bare string, number, or object, a value dlt moved into a cognee_node_set__v_* variant column, text that is not JSON, or an entry that is not a non-empty string. cognee raises a CogneeValidationError named DltNodeSetColumnNotAList that names the source, table, and row, and no document from that sync is stored. Emit a list even for a single name (["name"]) and re-sync the affected rows.

Hosts and connectors can check DOCUMENT_SYNC_VERSION in cognee.tasks.ingestion.dlt_utils to gate on core capability. Version 2 adds the per-row cognee_node_set column; the Notion connector refuses to run on an older core, while the Gmail, Google Drive, Google Calendar, and Linear connectors still need only version 1.

Sync and orphan cleanup. Document sources read back their full current snapshot (max_rows_per_table=0) and honor the write_disposition you pass: use replace for snapshot sources with no delete feed and merge with a hard-delete tombstone column for incremental sources that emit real deletions. primary_key defaults to id. Orphan cleanup is scoped to the source tag and the staging table the source loaded, so reconciling one document source (or one Drive folder) only removes that table’s rows, and relational ("dlt") rows and other sources’ rows are never cross-deleted in a mixed dataset. A table that tombstones have emptied still counts as loaded, so deleting the last message or file upstream forgets its document too. An incremental sync that brings no changes still re-reads the source’s own staged rows, so a dataset whose processing failed, or whose documents were removed locally, recovers on the next sync without anything changing upstream. Documents ingested before staging tables were recorded on each row are kept rather than guessed at. Deleting an orphan also invalidates the cached session turns that used that row’s graph elements, so a row removed upstream stops surfacing from session memory too; that step is best-effort and logged as a warning if it fails.

Orphan cleanup runs in the foreground of add() / remember(), and only after the pipeline finishes successfully — a run that returns an error leaves the previous records in place. Background runs (run_in_background=True) skip it, logging Skipping DLT orphan cleanup for background ingestion, because the detached pipeline has not committed when the call returns. To propagate upstream deletions, sync the affected source in the foreground.

Gmail connector

Gmail Connector

The Gmail connector is a first-class dlt source that turns your inbox into cognee memory. It reuses the same remember() + dlt path described above, so it gets incremental re-sync and forget-on-delete for free. gmail_source() returns a dlt resource that you hand directly to cognee.remember().

This connector reads the content of your email. It is strictly opt-in — nothing is fetched until you construct a source and call remember(). Scope what you ingest with label_ids, keep the OAuth token file (token.json) private, and prefer a dedicated dataset so you can wipe it with a single cognee.forget(...).

The Gmail connector is bundled with the Cognee SDK (from v1.6.1) and imported from cognee.tasks.ingestion.connectors. It supersedes the standalone cognee-community-connector-gmail package, so no community package is needed. To let users connect their mailboxes through Cognee’s API instead of a local token file, see the Gmail integration.

Installation

pip install "cognee[gmail]"

Or with uv:

uv pip install "cognee[gmail]"

The gmail extra pulls in google-api-python-client, google-auth, and google-auth-oauthlib. They are imported lazily: importing cognee.tasks.ingestion.connectors works without the extra, and the ImportError naming it is raised only when the Gmail client is built, during the remember() call.

One-Time OAuth Setup

The connector authenticates with Gmail via the OAuth2 installed-app (Desktop app) flow using the read-only scope https://www.googleapis.com/auth/gmail.readonly — it never modifies your mailbox.

1. In the Google Cloud Console, enable the Gmail API, configure an OAuth consent screen (add yourself as a test user), and create an OAuth 2.0 Client ID of type Desktop app.
2. Download the client-secret JSON and save it as credentials.json (or point credentials_path at it).
3. The first run opens a browser to consent and caches the resulting user token at token.json (token_path). Later runs reuse and silently refresh that token.

Usage

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

source = gmail_source(
    credentials_path="credentials.json",
    token_path="token.json",
    label_ids=["INBOX"],
)

await cognee.remember(
    source,
    dataset_name="gmail_inbox",
    primary_key="id",
    write_disposition="merge",   # incremental upsert by message id
    max_rows_per_table=0,        # 0 = no row cap (see note below)
)

gmail_source() accepts these keyword-only parameters:

Parameter	Default	Description
credentials_path	"credentials.json"	Path to the OAuth client-secret JSON (Desktop app).
token_path	"token.json"	Where the cached user token is read/written.
label_ids	None	Restrict to messages carrying all of these Gmail label ids (e.g. ["INBOX"]). None covers the whole mailbox except Spam and Trash.
max_results	None	Load only the newest N messages (handy for demos/tests). Each capped run starts again from the newest message and saves no incremental cursor, so later runs don’t sync only the changes. None = no cap.
resource_name	"gmail_messages"	dlt resource (and staging table) name. Give each mailbox its own name when several share one dataset or pipeline, so their cursors stay separate.
service	None	A pre-built Gmail API client, used instead of the local OAuth flow. build_gmail_service_from_access_token(token) builds one from an access token you already hold.
check_active	None	Optional callable run during extraction; raising from it aborts the sync (the integrations API uses it to stop when a connection is revoked mid-sync).
quota_units_per_minute	5000	Gmail API quota budget. Gmail allows 6,000 units per user per minute and fetching one message costs 20, so the default fetches about 250 messages a minute. Sources with the same resource_name in one process share one budget, and the first value set for that name wins.

The query and include_spam_trash parameters of the community package are gone.

The returned resource (named resource_name, gmail_messages by default) is preconfigured with primary_key="id", write_disposition="merge", and a _deleted hard-delete column, so combined with primary_key="id" on remember() it performs idempotent upserts by Gmail message id.

Gmail is a document source, so Cognee always reads its whole staging table back, whatever max_rows_per_table or DLT_MAX_ROWS_PER_TABLE says. Forget-on-delete therefore always compares against the whole synced corpus; passing max_rows_per_table=0 just makes that explicit.

How It Works

• Incremental sync: The first run lists the label from the newest message down. If it wasn’t capped with max_results and finished, it records the mailbox historyId in dlt’s per-resource state; later runs call users.history.list(startHistoryId=...) and emit only the delta (added / changed / deleted messages).
• Forget-on-delete: Messages reported as deleted or trashed by the History API are emitted with the _deleted hard-delete marker. dlt removes those rows from its destination on merge, and cognee’s existing orphan_cleanup then purges them from the graph, vector, and relational stores.
• Scope changes: Changing the set of label_ids makes the next run do a full backfill of the new scope, so older messages that now match are picked up; reordering the same labels keeps the cursor. That backfill also forgets every previously synced message outside the new scope, so narrowing label_ids removes the mail that no longer matches. On incremental runs, a message that leaves the scope (for example, trashed INBOX mail) is forgotten the same way. A backfill capped by max_results, or one that fails midway, forgets nothing.
• History expiry: Gmail expires history after roughly a week. On the resulting 404 the connector loads every message in the label again rather than silently stalling.
• Transient errors: every Gmail API call retries up to 6 times. A quota rejection (a 429, or a 403 with reason rateLimitExceeded or userRateLimitExceeded) that survives those retries then waits out a full 60-second window and tries again, up to five times. A call that still fails — a message fetch or the History API request — aborts the run before the cursor advances, so the next sync retries from the same point. Only a real 404 from the History API counts as the expired history above, and only a 404/410 on a message fetch counts as a deletion.

Deletions only propagate from a foreground run — see the orphan-cleanup note above. For a runnable walkthrough, and for when the cursor carries over, what capped syncs and prune do, and how to load a whole inbox, see the Gmail Ingestion guide.

Linear connector

The Linear connector turns one Linear team’s issues, comments, and projects into cognee memory. linear_source() returns a dlt resource you hand directly to cognee.remember(), so it takes the document-mode path described above (its source tag is linear) and reuses the same merge upserts by row id. It does not emit deletions yet — see What Is Ingested below.

This connector reads the content of your Linear workspace. It is strictly opt-in — nothing is fetched until you construct a source and call remember(). The token’s visibility is the scope, so private teams it can see are included, and anyone with read access to the target dataset can see what was ingested from them. Prefer a dedicated dataset so you can wipe it with a single cognee.forget(...).

This is the connector you drive from your own code with a token you already hold. It is not the Linear integration, which installs Cognee into a workspace as an OAuth agent.

Installation

Nothing to install. Unlike Gmail, Google Drive, and Google Calendar, the Linear connector needs no extra — dlt is a core cognee dependency — so cognee.tasks.ingestion.connectors carries it out of the box. All you need is an access token.

Usage

import cognee
from cognee.tasks.ingestion.connectors import linear_source

await cognee.remember(
    linear_source(
        team_id="<team uuid>",
        access_token="<oauth token or lin_api_... key>",
    ),
    dataset_name="linear",
    primary_key="id",
    write_disposition="merge",   # REQUIRED — the add pipeline defaults to "replace"
    max_rows_per_table=0,        # 0 = no row cap
)

Pass the team’s id (the uuid), not its key. One source is one team, so give each team its own linear_source() call. Run it again against the same dataset to sync only what changed.

write_disposition="merge" is mandatory. With the default "replace", the second sync drops and reloads the table instead of upserting into it.

linear_source() accepts these keyword-only parameters:

Parameter	Default	Description
team_id	— (required)	The Linear team’s id — the uuid, not the team key.
access_token	None	An OAuth access token or a personal lin_api_... API key. Required unless you pass service.
service	None	A pre-built Linear client used instead of building one from access_token. build_linear_service(token) builds one; mainly an injection point for tests and hosts.
resource_name	derived from team_id	dlt resource (and staging table) name. Defaults to linear_ followed by the lowercased team id with runs of non-alphanumeric characters replaced by _. Keep the linear_ prefix if a host tracks the table by it.
check_active	None	Optional callable run before each request; raising from it aborts the sync (hosts use it to stop when a connection is revoked mid-sync).
include_archived	True	Include archived issues, projects, and comments. Linear archives closed issues automatically, so leaving this on keeps most of a team’s history. Trashed issues are always skipped.
page_size	50	Rows per page, and also the maximum — larger values are clamped. A page of issues with their nested comments has to stay under Linear’s 10,000-point per-query complexity limit.
max_requests	1500	Request budget for a single run. A bigger team finishes over several runs.

The returned resource is preconfigured with primary_key="id", write_disposition="merge", and a _deleted hard-delete column.

What Is Ingested

Rows are flat {id, title, content, url, _deleted} documents whose ids are prefixed by kind:

Row	Id	Title	Content
Issue	issue:<uuid>	ENG-12 Title	state, assignee, priority, labels, project, parent, description, and the issue’s comments
Project	project:<uuid>	project name	status, lead, start and target dates, description, content

A comment is part of its issue’s document rather than a row of its own, so asking about a discussion finds the issue it belongs to.

• Sub-teams are not included. Select the sub-team itself to ingest its issues.
• An issue moved to another team stays in the old team’s table.
• A project shared by several teams is ingested once per selected team.
• Deletions are not propagated yet. Trashed issues are skipped, but anything already ingested stays in memory: the _deleted column is declared and always False. A comment deleted in Linear is hard-deleted there, so it is never seen again and stays in the issue’s document until the issue changes next.

How It Works

• Incremental by updatedAt: Linear sorts newest first and has no change feed, so each of the three streams — issues, comments, projects — keeps its own floor, plus the ids seen exactly at that floor, in dlt’s per-resource state. A run with no changes yields nothing.
• Comments are their own stream: an edited comment does not reliably bump its issue’s updatedAt, so changed comments are read separately and their parent issues are re-rendered.
• A run is bounded: max_requests (1,500 by default) caps one run. Linear allows 5,000 requests and 2,000,000 complexity points per hour for an OAuth app user (2,500 and 3,000,000 for a personal API key), shared with everything else that user does.
• It stops cleanly rather than raising or sleeping: on a Linear rate limit (a gateway HTTP 429 counts as one), remaining quota below a reserve, a spent request budget, or a token that expires after progress was made, the source stops where it is, keeps the state it reached, and records it in the cognee_sync_stats dict on the returned resource — failed, plus failed_rate_limit, failed_budget, or failed_auth. Raising would make dlt roll the run’s state back, so a team too big for one hour of quota would never finish; sleeping would hold the process-wide staging lock.
• The next run resumes from the cut point, down to the page cursor it stopped at, so even a large group of rows sharing one timestamp is read across several runs.
• Errors that do raise: a token rejected before any progress was made, a team that cannot be found, a pagination response that does not advance, and other API failures raise LinearAuthError, LinearTeamNotFoundError, or LinearAPIError. Their messages carry codes only — never the token, the query variables, or a response body.

Tokens

The source takes a plain access token and never refreshes it; a host that owns the OAuth flow passes a fresh token per run. OAuth tokens are sent as Bearer, personal API keys (lin_api_...) are sent as is. Surrounding whitespace is stripped, and a token that is empty or contains spaces, non-ASCII, or non-printable characters is refused when the source is built, without echoing the value.

Google Calendar connector

The Google Calendar connector syncs your calendars into cognee memory, incrementally, and forgets events when they are deleted. google_calendar_source() returns a dlt resource you hand directly to cognee.remember(). It takes the document-mode path described above (its source tag is google_calendar): each event becomes one document that goes through chunking and LLM entity extraction. The calendar name, organizer, and attendees with their responses are written into the document, so the same person is linked across meetings.

This connector reads personal data from the calendars you select. It is strictly opt-in — nothing is fetched until you construct a source and call remember(). Keep the token file private and prefer a dedicated dataset so you can wipe it with a single cognee.forget(...).

The connector is bundled with the Cognee SDK and imported from cognee.tasks.ingestion.connectors. It replaces the cognee-community Google Calendar package, so no community package is needed.

Installation

pip install "cognee[google-calendar]"

The google-calendar extra pulls in google-api-python-client, google-auth, and google-auth-oauthlib. They are imported lazily: importing cognee.tasks.ingestion.connectors works without the extra, and the ImportError naming it is raised only when the Calendar client is built.

One-Time OAuth Setup

The connector uses the OAuth2 installed-app (Desktop app) flow with the read-only scope https://www.googleapis.com/auth/calendar.readonly only. It never creates, changes, or deletes anything in Google Calendar.

1. In the Google Cloud Console, create a project and enable the Google Calendar API.

2. Under Google Auth Platform, set up the consent screen as External, keep it in Testing, add your own account as a test user, and add the calendar.readonly scope.

3. Under Clients, create an OAuth client of type Desktop app and download its JSON.

4. Point the connector at it and sign in once (a browser opens for consent):

   export GOOGLE_CALENDAR_CREDENTIALS_PATH=~/client_secret.json
   export GOOGLE_CALENDAR_TOKEN_PATH=~/.cognee_google_calendar_token.json
   
   import os
   
   from cognee.tasks.ingestion.connectors.google_calendar import (
       build_calendar_service,
       list_calendars,
   )
   
   # Caches the token in the second file, written with mode 0600.
   service = build_calendar_service(
       os.path.expanduser(os.environ["GOOGLE_CALENDAR_CREDENTIALS_PATH"]),
       os.path.expanduser(os.environ["GOOGLE_CALENDAR_TOKEN_PATH"]),
   )
   print(list_calendars(service))  # the calendar ids you can sync
   

Apps left in Testing get refresh tokens that expire after 7 days. Publish the app, or sign in again, for longer use.

Headless or hosted: instead of the browser flow, set GOOGLE_CALENDAR_REFRESH_TOKEN with GOOGLE_CALENDAR_CLIENT_ID and GOOGLE_CALENDAR_CLIENT_SECRET, or pass a ready-made access token (access_token= or GOOGLE_CALENDAR_ACCESS_TOKEN), which is never refreshed. build_calendar_service_from_access_token(token) builds a client from a token you already hold. When several are set, an access token wins, then the refresh token, then the installed-app flow. A source built with no credentials at all raises a ValueError right away, before any pipeline work starts.

Usage

import cognee
from cognee.tasks.ingestion.connectors import google_calendar_source

await cognee.remember(
    google_calendar_source(calendar_ids=["primary"]),  # GOOGLE_CALENDAR_* settings for auth
    dataset_name="my_calendar",
    primary_key="id",
    write_disposition="merge",   # REQUIRED — incremental upserts and deletes
    max_rows_per_table=0,        # 0 = no row cap
)

answer = await cognee.recall("Who is in my Thursday meetings?", datasets=["my_calendar"])

Run the same remember() call again whenever you want to sync.

Parameter	Setting (env var)	Default	Description
calendar_ids	GOOGLE_CALENDAR_IDS (comma-separated)	primary	Calendars to sync. An empty list is refused, because it would forget every synced event. primary and the real id of the primary calendar are the same calendar and give one set of documents.
settings	GOOGLE_CALENDAR_CLIENT_ID, _CLIENT_SECRET, _REFRESH_TOKEN, _CREDENTIALS_PATH, _TOKEN_PATH	read from the environment	A GoogleCalendarSettings object. token_path defaults to ~/.cognee_google_calendar_token.json.
access_token	GOOGLE_CALENDAR_ACCESS_TOKEN	None	A ready-made token for hosts and quick tries; never refreshed.
resource_name	—	"google_calendar_events"	dlt resource (and staging table) name. Give each account its own name when several sync into one dataset, so their state stays separate.
check_active	—	None	Optional callable run before every API request; raising from it aborts the sync.
service	—	None	A pre-built Calendar client, mainly for tests.

All parameters except calendar_ids are keyword-only. The returned resource is preconfigured with primary_key="id", write_disposition="merge", and a _deleted hard-delete column.

How Sync and Forget-on-Delete Work

• Sync tokens: the first run lists every event of each calendar; later runs pass that calendar’s nextSyncToken and get only what changed. The token is saved in dlt state once the whole round has finished.
• Recurring events: a series is one document with its RRULE, and a changed single occurrence is its own document.
• All-day events show their last day: a one-day event reads all day, 2026-11-08, a three-day one all day, 2026-11-08 to 2026-11-10.
• Deleted events come back as cancelled and are forgotten if they ever reached memory. A cancelled single occurrence forgets only that occurrence; deleting a whole series also forgets its changed occurrences. An event created and deleted between two syncs is neither stored nor counted.
• Expired sync token: a 410 Gone lists that calendar again and reconciles it like a snapshot, so events deleted in the meantime are still forgotten.
• Removed calendars: removing a calendar from calendar_ids forgets its events. A calendar is also forgotten when it was synced before and has left the account’s calendar list. A 404 for a calendar that is still listed, or that was never synced (most likely a typo), is a failure, not a deletion.
• Failures never delete: a calendar that fails keeps its old state and is retried on the next run, while the other calendars finish and save theirs. The run raises only when no calendar finished. A 401, a 429, or a 403 with a rate-limit reason ends the run early, since the remaining calendars would fail the same way; any other 403 (lost access) fails only its own calendar. This holds for HTTP responses only: a transport error, or an auth error raised by the Google client library before any HTTP call (such as google-auth’s RefreshError), ends the whole run.

Deletions only propagate from a foreground run — see the orphan-cleanup note above.

After a run, the cognee_sync_stats dict on the returned resource holds counts only — never titles or content: scanned, skipped, deleted, calendars, and failed, plus failed_auth (HTTP 401), failed_rate_limit (HTTP 429, or 403 with a rate-limit reason), or failed_calendar_sync (anything else) for the reason a calendar failed.

What Is Read

• From the selected calendars only: title, calendar name, start and end with time zone, location, organizer, attendees with name, email, and response, the recurrence rule, status, event type, the Google Meet link, the description, and the event link.
• Not read: attachments, conference details beyond the Meet link, reminders, colors, and other calendars. workingLocation events are skipped.
• dlt state stores each calendar’s id, its sync token, and its event ids — not titles or descriptions. Logs and errors carry counts and HTTP statuses only; calendar ids and request URLs are left out.
• Event text becomes ordinary cognee documents, so during cognify it is sent to your configured LLM and embedding providers like any other data. Use local models if that matters to you.
• The token file holds a refresh token and is written with mode 0600. Never commit it; revoke access any time at myaccount.google.com/permissions and delete the file.

Limitations

• The full history of each calendar is read on the first run; the sync-token API allows no date window.
• workingLocation events are skipped and attachments are not ingested.

Notion connector

The Notion connector syncs the Notion pages and database rows you choose into cognee memory, incrementally and with forget-on-delete. notion_source() returns a dlt resource you hand directly to cognee.remember(), so it takes the document-mode path described above (its source tag is notion): each page becomes a text document that goes through normal chunking and LLM graph extraction.

This connector reads the content of your Notion pages. It is strictly opt-in — nothing is fetched until you construct a source and call remember(), and only the roots you name (plus everything nested under them) are read. Prefer a dedicated dataset so you can wipe it with a single cognee.forget(...).

Installation

Nothing to install. The Notion connector talks to the Notion REST API over httpx, a core cognee dependency, so like Linear it needs no extra and ships in cognee.tasks.ingestion.connectors with a plain cognee install.

Setup

1. Create a Notion internal integration and copy its token.
2. Share every page or database you want synced with that integration (... > Connections > Add connection).
3. Set NOTION_API_KEY, or pass token= directly, plus your LLM_API_KEY as for any other cognee run.

Usage

import cognee
from cognee.tasks.ingestion.connectors import notion_source

await cognee.remember(
    notion_source(
        root_page_ids=["<page id from the page URL>"],
        resource_name="notion_<connection id>",
    ),
    dataset_name="my_notion_space",
    primary_key="id",
    write_disposition="merge",   # incremental upsert + tombstones
    max_rows_per_table=0,        # 0 = no row cap
)

A database (or one of its data sources) works the same way:

notion_source(root_data_source_ids=["<database or data source id>"])

Run it again against the same dataset to pick up edits and forget pages that are gone.

notion_source() accepts these parameters:

Parameter	Default	Description
token	None	Notion integration token. Falls back to the NOTION_API_KEY environment variable; with neither, a ValueError is raised.
root_page_ids	None	Page ids to walk, plus everything nested under them.
root_data_source_ids	None	Data source ids to sync, plus everything nested under their rows. A database id is also accepted and resolved into that database’s data sources. At least one of root_page_ids / root_data_source_ids is required, or a ValueError is raised.
workspace_id	None	Stable workspace id used in each row’s node set. When omitted it is resolved once from the integration’s bot user (users/me); the sync fails rather than emitting untagged rows if it cannot be determined.
resource_name	"notion_pages"	dlt resource name — also the staging table and the key of the incremental state. See the warning below.
check_active	None	Optional callable run during extraction; raising from it aborts the sync (hosts use it to stop when a connection is revoked mid-sync).
http_client	None	A pre-built httpx client, mainly a test-injection point; when omitted one is built for the sync.

The returned resource is preconfigured with primary_key="id", write_disposition="merge", and a _deleted hard-delete column. Counts-only diagnostics (never titles or content) are exposed in the cognee_sync_stats dict on the returned resource.

Every Notion source that syncs into the same dataset needs its own resource_name. Two sources sharing one share state, so each run tombstones and forgets the other’s pages. A run refuses to continue when the stored state belongs to a different workspace, but two root sets in one workspace under one name cannot be told apart from an edited selection. Renaming a source starts it over, and documents synced under the old name are not cleaned up.

What Is Ingested

• One document per page. Rows are {id, title, content, url, _deleted}. A database row is a page too: its properties (rich_text, select, multi_select, status, date, people, number, checkbox, url, email, relation ids) are rendered as Name: value lines at the top of its content. Relation and people values stay raw ids/names in that text; they are not resolved into graph edges.
• Nested content is followed: sub-pages, databases embedded in pages, and content inside columns, toggles, callouts, lists, and synced blocks, at any depth. A synced block that copies another page’s original is not searched; its sub-pages are only synced when the original’s page is under a selected root. Archived and trashed pages are skipped.
• Node sets: each row carries cognee_node_set = ["notion:<workspace_id>:<root_id>"] — always ids, never titles — so the page’s document and chunks belong to the root you configured it under. When a page is reachable from two selected roots, the nearest root wins. Its position inside the tree (parent page, block) is not recorded.
• Attachments (file, image, PDF, video blocks) become a name/caption reference line only; the binary is not downloaded.

How It Works

• Re-walk every run: Notion has no delete feed, so each run walks the selected trees again to rediscover what still exists. A page’s content is only re-rendered when its last_edited_time or its root changed since the previous run; unchanged pages are not re-ingested.
• Forget-on-delete: pages that vanished (unshared, trashed, or moved out of the selected roots) are emitted as _deleted tombstones; dlt drops them on merge and cognee’s orphan_cleanup forgets them from the graph, vector, and relational stores (foreground runs only — see the orphan-cleanup note above).
• Moving a page between two selected roots, or changing roots so that another one reaches it, changes its node set (part of the row’s identity), so it is re-ingested once under the new root.
• Cost: a run makes at least one request per page even when nothing changed — a 20,000-page tree is tens of thousands of requests, hours at Notion’s rate limit of about 3 requests per second. The connector does not throttle itself; HTTP 429 and 5xx responses are retried (up to 5 attempts) using Retry-After or exponential backoff.

Errors

A run aborts before any tombstone is written — so a partial walk can never look like a mass deletion — when:

• a transient error (network failure, or a 429/5xx that exhausts the retries) occurs;
• listing a readable page’s blocks or querying a data source fails, or Notion reports more results without a next_cursor;
• one of your explicit roots returns 403/404 (a configuration problem, not a deletion);
• a data source has more than 10,000 rows (see limitations).

A 403/404 on a page or database reached through the tree, by contrast, means that item is gone, and it is tombstoned.

Limitations

• Data sources over 10,000 rows abort every run: Notion cuts a query off there, and syncing only the first 10,000 would forget the rest.
• The walk holds cognee’s process-wide dlt staging lock, so while a large tree syncs every other dlt ingestion in the same process (Drive, Gmail, Google Calendar, Linear, CSV) waits.
• Teamspaces are not exposed by Notion’s public REST API, so pages are grouped by selected root, never by teamspace.
• Text shown from other objects goes stale: a page is only re-rendered when it changes itself, so sub-page and database titles, synced copies, and mentions stay as they were until the page is edited.
• Pinned to Notion API version 2025-09-03, the first with data sources as first-class objects.

Use Cases

CRM and Relational Data

Load customer, order, and product tables from a database. Foreign keys between tables (e.g., order.customer_id → customer.id) become graph edges, enabling cross-table queries like “Which customers ordered product X?”

CSV Analytics Pipeline

Point cognee at CSV exports from analytics tools. Each row becomes a searchable node in the graph, and you can combine them with unstructured reports in the same dataset.

Event Log Ingestion

Use write_disposition="append" to stream event batches into cognee without deduplication. Query across the full event history with natural language.

Database Mirroring

Use write_disposition="merge" to keep cognee’s graph in sync with a live database. Rows that are removed upstream are cleaned up best-effort; any orphaned rows that fail to delete are logged and retried on the next ingest.

Cloud Data Warehouses (Snowflake, Redshift, BigQuery)

Amazon Redshift speaks the PostgreSQL wire protocol, so the standard connection string auto-detection works:

await cognee.remember(
    "postgresql://user:[e-mail]:5439/mydb",
    dataset_name="redshift_data",
    primary_key="id",
)

Snowflake requires constructing a dlt sql_database source manually (install snowflake-sqlalchemy first):

pip install cognee snowflake-sqlalchemy

from dlt.sources.sql_database import sql_database
import cognee

source = sql_database(
    credentials="snowflake://user:password@account_identifier/database/schema?warehouse=MY_WH",
    table_names=["orders", "customers"],
)

await cognee.remember(source, dataset_name="snowflake_data", primary_key="id")

The account_identifier is the part before .snowflakecomputing.com in your Snowflake URL (e.g. myorg-myaccount). Omit table_names to ingest all tables in the schema.

Google BigQuery works the same way using dlt’s BigQuery connector — construct the source and pass it directly to cognee.remember(). See the dlt sql_database docs for connector-specific setup.

Remember Operation

Learn more about data ingestion in cognee

dlt Documentation

Official dlt documentation and guides