dlt (Data Load Tool)

​

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

pip install 'cognee[dlt]'

uv pip install 'cognee[dlt]'

​

Quick Start

​

1. Ingest a dlt Resource

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())

​

2. Build and Query the Graph

​

Other Input Modes

​

CSV Auto-Detection

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

​

Database Connection String

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

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

​

Mixed Structured + Unstructured

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",
)

​

Write Dispositions

• 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",
)

​

How It Works

1. Source Detection: cognee identifies dlt resources, CSV files, and connection strings in the input
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
5. LLM Bypass: Structured rows skip chunking, entity extraction, and summarization — the graph is built entirely from schema metadata

​

Foreign Key Resolution

• Target row not loaded: if a foreign key points at a row that wasn’t ingested — most commonly because the target table hit the max_rows_per_table cap — 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

pip install cognee-community-connector-gmail

uv pip install cognee-community-connector-gmail

import cognee
from cognee_community_connector_gmail 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)
)

​

Use Cases

await cognee.remember(
    "postgresql://user:pass@my-cluster.us-east-1.redshift.amazonaws.com:5439/mydb",
    dataset_name="redshift_data",
    primary_key="id",
)

pip install 'cognee[dlt]' 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")