Release highlights: 1.22

Breaking changes

• Pydantic v1 support removed. All Pydantic v1 compatibility code has been removed and dlt now requires Pydantic v2 (#3572). Update any models still using v1 APIs before upgrading.
• data_type contract now covers full column type. The data_type schema contract now applies to changes in precision, scale, nullability, and timezone on existing complete columns, not only to variant columns (#3572). Pipelines with data_type set to freeze that changed these properties on existing columns will now be blocked.
• Compound hints now replace instead of accumulate. merge_columns previously merged additively, so compound hints like merge_key and primary_key were added rather than replaced (#3431). Redefining these hints on an already extracted resource now replaces the previous configuration instead of accumulating old and new keys.

Validate event streams with Pydantic discriminated unions

The Pydantic support was reworked (#3572). A RootModel wrapping a discriminated union validates a single resource that emits different item types and dispatches each to its own table, resolving the variant by its discriminator field. Mark the model authoritative with DltConfig, then route items with dlt.mark.with_table_name.

from typing import ClassVar, Literal, Union
from typing_extensions import Annotated
from pydantic import BaseModel, Field, RootModel
from dlt.common.libs.pydantic import DltConfig
import dlt


class EventBase(BaseModel):
    kind: str
    id: int

class ClickEvent(EventBase):
    kind: Literal["click"]
    element_id: str

class PurchaseEvent(EventBase):
    kind: Literal["purchase"]
    amount: float

EventUnion = Annotated[Union[ClickEvent, PurchaseEvent], Field(discriminator="kind")]

class Event(RootModel[EventUnion]):
    dlt_config: ClassVar[DltConfig] = {"is_authoritative_model": True}


TABLE_MAP = {"click": "click_events", "purchase": "purchase_events"}

@dlt.resource(name="events", columns=Event, schema_contract={"data_type": "discard_row"})
def event_stream(items):
    for item in items:
        yield dlt.mark.with_table_name(item, TABLE_MAP[item["kind"]])

Parallelize dependent resources in rest_api

Set parallelized to True on a dependent resource (transformer) so child fetches for many parent items run concurrently in dlt's thread pool instead of one after another (#3574). All pages for a single parent are collected in memory before being yielded, so expect higher memory use for parents with many child pages.

import dlt
from dlt.sources.rest_api import rest_api_source

source = rest_api_source({
    "client": {"base_url": "https://api.example.com"},
    "resources": [
        "posts",
        {
            "name": "post_comments",
            "parallelized": True,
            "endpoint": {"path": "posts/{resources.posts.id}/comments"},
        },
    ],
})

Filter dataset relations by load ID

dataset.table() now takes a load_ids argument to return only rows from specific loads, joining nested tables back to their root when needed (#3547). List available loads with dataset.load_ids() and get the newest with dataset.latest_load_id(). The feature is experimental.

import dlt

pipeline = dlt.pipeline("my_pipeline", destination="duckdb")
dataset = pipeline.dataset()

all_loads = dataset.load_ids()            # every load id in the dataset
recent_users = dataset.table("users", load_ids=all_loads[-1:])
df = recent_users.df()

Zero-downtime replace on Snowflake with atomic swap

With enable_atomic_swap set on the Snowflake destination, the staging-optimized replace strategy swaps production and staging tables using ALTER TABLE ... SWAP WITH instead of cloning them (#3540). The swap is near-instant, so tables stay available during replacement. Each table keeps its existing permissions, and dlt does not drop the old staging table.

import dlt

pipeline = dlt.pipeline(
    pipeline_name="my_pipeline",
    destination=dlt.destinations.snowflake(enable_atomic_swap=True),
)

Compact config layout for renamed sources

When a source's section differs from its name, for example after .clone(name=..., section=...), dlt now accepts a compact sources.<name>.<key> config path in addition to the full sources.<section>.<name>.<key> (#3664). The full path and the section path still take precedence.

# compact layout: just the source name
[sources.my_db.credentials]
password = "..."

# full layout: section + name, takes precedence
[sources.my_db_module.my_db.credentials]
password = "..."

Load to Hugging Face with the hf:// protocol

The filesystem destination now supports the hf:// protocol, loading directly to Hugging Face dataset repositories (#3669). Install dlt[hf], point bucket_url at hf://datasets/<namespace>, and dlt creates one repo per dataset, commits each table atomically, and writes a dataset card with per-table subsets (#3689) for the dataset viewer.

[destination.filesystem]
bucket_url = "hf://datasets/my-org"

[destination.filesystem.credentials]
hf_token = "hf_..."

import dlt

@dlt.resource
def my_data():
    yield [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]

pipeline = dlt.pipeline(
    pipeline_name="my_pipeline",
    destination="filesystem",
    dataset_name="my_dataset",
)
pipeline.run(my_data(), write_disposition="append")

Shout-out to new contributors

Big thanks to our newest contributors:

• @Shadesfear
• @ShreyasGS
• @arel
• @karlanka
• @kien-truong
• @thecaptain789

Full release notes

View the 1.22.0 release notes