Propagate primary_key from root to nested tables

About this Example

You learn how to propagate specific fields (e.g., primary, foreign keys) from a parent record into each child record.

In this example, we'll explore how to:

• Add parent_id into each child record using add_parent_id function
• Use the add_map function to apply this custom logic to every record in the dataset

important

Please note that dlt metadata, including _dlt_id and _dlt_load_id, will still be loaded into the tables.

Full source code

from typing import Any, Generator
import dlt


# Define a dlt resource with write disposition to 'merge'
@dlt.resource(name="parent_with_children", write_disposition={"disposition": "merge"})
def data_source() -> Generator[list[dict[str, Any]], None, None]:
    # Example data
    data = [
        {
            "parent_id": 1,
            "parent_name": "Alice",
            "children": [
                {"child_id": 1, "child_name": "Child 1"},
                {"child_id": 2, "child_name": "Child 2"},
            ],
        },
        {
            "parent_id": 2,
            "parent_name": "Bob",
            "children": [{"child_id": 3, "child_name": "Child 3"}],
        },
    ]

    yield data


# Function to add parent_id to each child record within a parent record
def add_parent_id(record: dict[str, Any]) -> dict[str, Any]:
    parent_id_key = "parent_id"
    for child in record["children"]:
        child[parent_id_key] = record[parent_id_key]
    return record


# Create and configure the dlt pipeline
pipeline = dlt.pipeline(
    pipeline_name="generic_pipeline",
    destination="duckdb",
    dataset_name="dataset",
)

# Run the pipeline
load_info = pipeline.run(data_source().add_map(add_parent_id), primary_key="parent_id")
# print(load_info)

# the primary key `parent_id` is propagated into the nested `children` table
row_counts = pipeline.last_trace.last_normalize_info.row_counts
assert row_counts["parent_with_children"] == 2
assert row_counts["parent_with_children__children"] == 3