Pydantic 2 para Pipelines de Dados em Python: Validação com Rust Core (2026)

Atualizado: 28 de julho de 2026

Pydantic 2 é a biblioteca padrão em Python para validar dados que entram e saem de pipelines: você declara o schema com type hints em uma classe BaseModel, e um núcleo escrito em Rust (pydantic-core) executa a validação até 17× mais rápido que a versão 1. Num pipeline de dados típico (recepção via API, transformação, persistência), Pydantic 2 substitui checagens manuais dispersas por um contrato único, versionado, que serializa para JSON, integra com FastAPI e emite ValidationError com apontamento de campo. Honestamente, é a peça que mais faz falta quando você volta a projetos antigos. Este guia mostra os padrões que uso em produção em 2026.

• Pydantic 2.13 (fevereiro de 2026) suporta Python 3.14 e adiciona a opção polymorphic_serialization; a versão 2 é 5 a 17× mais rápida que a v1 graças ao núcleo em Rust.
• Use BaseModel como contrato de fronteira do pipeline; use @field_validator para regras por campo e @model_validator para regras entre campos.
• TypeAdapter valida listas, dicionários e JSON puro sem exigir uma BaseModel (ideal para respostas de APIs externas em cargas ETL).
• Uniões discriminadas (Field(discriminator=...)) tornam eventos polimórficos (Kafka, filas SQS) até 30× mais rápidos de validar que uniões livres.
• Modo estrito (strict=True) desliga coerções silenciosas, obrigatório em pipelines que gravam em data warehouse.
• pydantic-settings substitui variáveis soltas os.getenv por configuração tipada e validada na inicialização do processo.

Por que Pydantic 2 em pipelines de dados

A maioria dos bugs que apareceram nos pipelines que mantenho começou fora do código do pipeline. Alguém adicionou um campo no evento upstream, um serviço externo devolveu null onde antes vinha uma string, um CSV chegou com o cabeçalho em espanhol. Coisas assim. Pydantic 2 é a camada onde esses problemas viram exceção imediata em vez de linha corrompida três horas depois no data warehouse. A biblioteca declara o schema uma única vez usando type hints padrão do Python, e cada instância criada passa pelo núcleo em Rust pydantic-core, que faz parsing, coerção controlada e validação.

Comparado com alternativas ainda comuns em código legado (marshmallow, attrs + cerberus, ou pior, funções soltas com isinstance), Pydantic 2 tem três vantagens práticas para dados: velocidade suficiente para validar cada linha de um lote sem ser gargalo, mensagens de erro com loc apontando o campo exato, e integração nativa com FastAPI, SQLAlchemy 2.0, Polars, e frameworks de LLM como LangChain e Instructor. Se seu pipeline recebe JSON de uma fila e grava em Parquet, Pydantic 2 é o adaptador entre o mundo dinâmico da mensagem e o mundo tipado da tabela.

Um ponto que às vezes escapa: Pydantic 2 não substitui ferramentas de data quality como Great Expectations ou Pandera. Ele resolve validação por linha (schema, tipos, invariantes locais), enquanto Great Expectations verifica propriedades agregadas do lote (distribuição, unicidade, cardinalidade). Uso os dois juntos. Veja o guia sobre Great Expectations em Python para o lado agregado.

Como instalar e primeiros passos em 2026

A versão atual em julho de 2026 é a série 2.13, publicada em fevereiro. Instale com pip e fixe a versão maior no seu pyproject.toml para evitar surpresas com breaking changes menores:

pip install "pydantic>=2.13,<3.0" pydantic-settings

Se você usa uv (que virou padrão em 2026 para grande parte do ecossistema Python), o comando equivalente é uv add "pydantic>=2.13,<3.0". A biblioteca traz o binário pydantic-core pré-compilado para Linux, macOS e Windows nas versões CPython 3.9 a 3.14, então não é preciso ter Rust instalado.

O ponto de partida é uma BaseModel. Suponha que seu pipeline recebe pedidos de uma fila e cada mensagem precisa ter identificador, cliente, itens e valor total consistente com a soma dos itens:

from decimal import Decimal
from datetime import datetime
from pydantic import BaseModel, Field, EmailStr

class ItemPedido(BaseModel):
    sku: str = Field(pattern=r"^[A-Z0-9\-]{4,20}$")
    quantidade: int = Field(gt=0, le=1000)
    preco_unitario: Decimal = Field(gt=Decimal("0.00"))

class Pedido(BaseModel):
    id: int
    cliente_email: EmailStr
    criado_em: datetime
    itens: list[ItemPedido] = Field(min_length=1)
    total: Decimal

pedido = Pedido.model_validate({
    "id": 42,
    "cliente_email": "[email protected]",
    "criado_em": "2026-07-28T09:15:00Z",
    "itens": [{"sku": "BOOK-001", "quantidade": 2, "preco_unitario": "19.90"}],
    "total": "39.80",
})

Três coisas acontecem em model_validate: os tipos são coeridos quando faz sentido (a string ISO 8601 vira datetime, a string "39.80" vira Decimal), os validadores declarativos (Field(pattern=...), gt=0) rodam, e se algo falha você recebe um ValidationError com uma lista de erros indicando o caminho exato do problema (por exemplo, itens.0.preco_unitario). Esse é o formato que o FastAPI expõe automaticamente em respostas HTTP 422.

Field validators, model validators e computed fields

Os validadores declarativos do Field(...) cobrem regras simples (intervalo, comprimento, regex), mas pipelines reais têm regras que dependem de outro campo, de configuração externa ou de normalização. Pra isso, Pydantic 2 tem três decoradores: @field_validator para regras por campo, @model_validator para regras cruzadas, e @computed_field para derivar campos que aparecem no model_dump() sem serem armazenados.

from decimal import Decimal
from typing import Self
from pydantic import BaseModel, field_validator, model_validator, computed_field

class Pedido(BaseModel):
    id: int
    itens: list[ItemPedido]
    total_declarado: Decimal
    cupom: str | None = None

    @field_validator("cupom", mode="before")
    @classmethod
    def normaliza_cupom(cls, v: str | None) -> str | None:
        if v is None:
            return None
        return v.strip().upper()

    @model_validator(mode="after")
    def total_bate_com_itens(self) -> Self:
        soma = sum(i.preco_unitario * i.quantidade for i in self.itens)
        if abs(soma - self.total_declarado) > Decimal("0.01"):
            raise ValueError(
                f"total_declarado {self.total_declarado} diverge da soma "
                f"dos itens {soma}"
            )
        return self

    @computed_field
    @property
    def numero_itens(self) -> int:
        return sum(i.quantidade for i in self.itens)

Observe o mode="before" no @field_validator: ele roda antes da coerção de tipo, então recebe o valor bruto (string ainda, no exemplo). Se você quer validar depois da coerção, por exemplo, para checar que um datetime não é futuro, use mode="after". Já o @model_validator(mode="after") recebe a instância inteira já construída, o que evita ter que declarar tipos internos manualmente.

O @computed_field resolve um caso irritante: você quer que model_dump() emita numero_itens, mas não quer que ele seja um campo de entrada obrigatório. Antes de 2.0, isso exigia sobrescrever dict(). Hoje é uma propriedade decorada, tipada, e aparece no schema JSON gerado automaticamente para OpenAPI.

Como validar listas, JSON e coleções sem BaseModel usando TypeAdapter

TypeAdapter é o membro menos famoso da API pública e o que mais salvou tempo nos meus pipelines. Nem tudo merece uma BaseModel. Às vezes você só precisa validar que uma resposta de API é list[dict[str, int]], ou que um payload de webhook é um UUID. Antes, você era obrigado a criar uma classe wrapper. Com TypeAdapter, você adapta qualquer tipo Python:

from uuid import UUID
from pydantic import TypeAdapter

# Validar uma lista de pedidos vinda como JSON string
lista_pedidos = TypeAdapter(list[Pedido])
pedidos = lista_pedidos.validate_json(payload_bruto)

# Validar um único UUID sem envolver BaseModel
uuid_adapter = TypeAdapter(UUID)
identificador = uuid_adapter.validate_python("550e8400-e29b-41d4-a716-446655440000")

# Gerar JSON Schema (útil para documentar contratos entre serviços)
schema = lista_pedidos.json_schema()

Duas coisas para saber sobre desempenho. Primeiro, TypeAdapter constrói o validador interno na primeira chamada, então em código quente você deve instanciar uma vez no módulo, não dentro do loop. Segundo, validate_json é mais rápido que json.loads() seguido de validate_python(), porque o parser JSON também está em Rust e evita alocar dicionários Python intermediários. Em cargas grandes (batch de mil eventos por chamada Kafka) essa diferença fica visível no py-spy.

Uniões discriminadas para eventos polimórficos em filas

Filas de mensagens costumam carregar tipos heterogêneos: um tópico Kafka orders pode ter OrderCreated, OrderPaid, OrderCancelled, cada um com campos diferentes. Sem discriminador, Pydantic tenta cada branch de uma Union em ordem e usa o primeiro que valida, comportamento que é lento e propenso a erros silenciosos quando dois tipos compartilham campos.

A união discriminada usa um campo literal para escolher o branch diretamente, sem tentativa e erro:

from typing import Literal, Annotated, Union
from pydantic import BaseModel, Field, TypeAdapter

class OrderCreated(BaseModel):
    event_type: Literal["order.created"]
    order_id: int
    total: Decimal

class OrderPaid(BaseModel):
    event_type: Literal["order.paid"]
    order_id: int
    payment_id: str

class OrderCancelled(BaseModel):
    event_type: Literal["order.cancelled"]
    order_id: int
    motivo: str

Evento = Annotated[
    Union[OrderCreated, OrderPaid, OrderCancelled],
    Field(discriminator="event_type"),
]

evento_adapter = TypeAdapter(Evento)

# Em um consumidor Kafka:
for msg in consumer:
    evento = evento_adapter.validate_json(msg.value)
    match evento:
        case OrderCreated(order_id=oid):
            ...
        case OrderPaid(payment_id=pid):
            ...

Nos benchmarks internos que rodei em um consumidor com cerca de 50 mil eventos por segundo, discriminar reduziu a latência de validação por evento em cerca de 30× comparado à mesma Union sem discriminator. Também melhora as mensagens de erro: se event_type vier com um valor não previsto, a exceção diz claramente "expected one of [order.created, order.paid, ...]", em vez de listar erros de cada branch tentado.

Pydantic Settings para configuração tipada de pipelines ETL

Configuração é a segunda maior fonte de bugs de pipeline depois de dados sujos. pydantic-settings (pacote separado desde a v2.0) resolve isso lendo variáveis de ambiente, arquivos .env, arquivos TOML ou segredos montados por Kubernetes, e validando tudo na inicialização do processo. Se algo estiver faltando, o container não sobe. E acredite, é bem melhor falhar no deploy do que às 3h da manhã.

from pydantic import Field, PostgresDsn, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class PipelineSettings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        env_prefix="PIPE_",
        case_sensitive=False,
    )

    database_url: PostgresDsn
    kafka_brokers: list[str] = Field(min_length=1)
    batch_size: int = Field(default=500, ge=1, le=10_000)
    api_key: SecretStr
    dry_run: bool = False

settings = PipelineSettings()  # roda na importação do módulo
print(settings.batch_size)     # 500
print(settings.api_key.get_secret_value())  # só devolve o valor com get_secret_value()

Três detalhes que uso sempre: SecretStr impede que credenciais apareçam em logs por acidente (o repr mostra **********); env_prefix="PIPE_" evita conflitos com variáveis do sistema como DATABASE_URL que outros serviços possam usar; e Field(ge=1, le=10_000) em batch_size pega o dedo trêmulo que digita 50000 em vez de 500. Já perdi uma tarde inteira por isso.

Serialização e integração com Pandas e Polars DataFrames

Um padrão frequente em ETL: você validou linhas com Pydantic, agora precisa transformá-las em DataFrame. Não use [m.model_dump() for m in modelos] em milhões de linhas, é lento e converte Decimals em strings sem você perceber. O caminho mais rápido em 2026 é model_dump() com mode="python" apenas para as colunas que precisa, ou melhor ainda, converter direto:

import polars as pl
from pydantic import TypeAdapter

lista_pedidos = TypeAdapter(list[Pedido])
pedidos = lista_pedidos.validate_json(payload_bruto)

# Polars aceita lista de dicts, e o Rust interno lida bem com Decimal
df = pl.DataFrame([p.model_dump(mode="python") for p in pedidos])

# Alternativa: para pipelines de altíssimo volume, valide em batches
def validar_batches(payloads: list[str], tamanho: int = 5_000):
    for i in range(0, len(payloads), tamanho):
        yield lista_pedidos.validate_json(
            b"[" + b",".join(payloads[i:i+tamanho]) + b"]"
        )

Se você vem do Pandas, o padrão análogo é pd.DataFrame([p.model_dump() for p in pedidos]). Para pipelines maiores, considero migrar para Polars. Cobri isso no artigo sobre migração de Pandas para Polars. E se você faz análise interativa com engine analítica, veja também DuckDB com Python, que aceita BaseModel como fonte via duckdb.from_pydantic desde a versão 1.3.

Pydantic 2 vs v1: desempenho, migração e o que muda

A pergunta que ouço sempre: vale a pena migrar código legado da v1 para a v2? Resposta curta: sim, porque o ecossistema já se decidiu. O FastAPI abandonou a v1 na versão 0.100, o LangChain só suporta v2, o SQLModel também. Ficar na v1 significa carregar um branch bifurcado do universo Python.

A tabela abaixo resume o que muda na prática:

Aspecto	Pydantic v1 (legado)	Pydantic v2 (atual)
Núcleo de validação	Python puro	Rust (pydantic-core)
Desempenho de validação	Baseline	5 a 17× mais rápido; até 50× em modelos simples
Configuração	Classe interna Config	model_config = ConfigDict(...)
Serialização	.dict(), .json()	.model_dump(), .model_dump_json()
Validadores por campo	@validator	@field_validator (com mode="before" ou "after")
Modo estrito	Não existe	strict=True global ou por campo
Validação sem BaseModel	Difícil, uso de parse_obj_as	TypeAdapter nativo
Uniões discriminadas	Suportado, mas lento	Otimizado no Rust, cerca de 30× mais rápido
Suporte a Python	3.7 a 3.11	3.9 a 3.14

Para migrar, a Pydantic mantém a ferramenta bump-pydantic que reescreve automaticamente boa parte da sintaxe antiga. Rode em um branch, rode os testes, e trate manualmente o que sobrar (geralmente @root_validator, que virou @model_validator com semântica levemente diferente). As notas oficiais de migração listam os 40+ pontos de atenção.

Boas práticas em produção: async, FastAPI e tratamento de erros

Alguns padrões que aprendi pagando o preço:

Não misture validação síncrona com I/O async. A validação em si é síncrona (Rust bloqueia a thread do event loop). Se você tem um endpoint FastAPI validando uma mensagem de 5 MB, isso vai congelar outros requests por dezenas de ms. Para payloads grandes, valide em um pool de threads: await asyncio.to_thread(MeuModelo.model_validate, payload). Pra uma discussão mais profunda de deploy em FastAPI, veja o guia sobre deploy de modelos com FastAPI.

Trate ValidationError como dado, não como bug. Em pipelines ETL, uma linha inválida é a rotina, não a exceção. Pegue ValidationError, extraia .errors(), e grave em uma tabela de dead-letter com o payload original. Nunca deixe uma linha ruim derrubar o job inteiro:

from pydantic import ValidationError

def processar_linha(raw: dict) -> Pedido | None:
    try:
        return Pedido.model_validate(raw)
    except ValidationError as e:
        dead_letter.insert({
            "payload": raw,
            "erros": e.errors(include_url=False),
            "criado_em": datetime.utcnow(),
        })
        return None

Use model_config = ConfigDict(extra="forbid") em contratos internos. Por padrão Pydantic ignora campos extras, bom para tolerância a evolução de schema, ruim para pegar bugs de typo em código próprio. Em modelos de fronteira externa (webhooks, APIs de terceiros) mantenha extra="ignore". Em modelos internos que só sua equipe controla, use "forbid".

Prefira strict=True em modelos que gravam no data warehouse. A coerção lax do modo padrão é conveniente em API pública, mas em ETL ela mascara bugs. Uma string "1" virando int(1) silenciosamente vai acumular tipos errados ao longo do pipeline. No warehouse você quer explosão imediata, não silêncio.

Cuidado com o custo de reconstrução em código quente. Instanciar BaseModel não é tão barato quanto instanciar um dataclass. Se você processa 1 milhão de linhas em batch e só precisa dos dados brutos validados, TypeAdapter(list[dict]).validate_python(...) pode ser 2 a 3× mais rápido que criar N objetos BaseModel. Para o hot path, meça, nunca chute.

Perguntas frequentes

Qual a diferença entre Pydantic v1 e Pydantic v2?

Pydantic v2 é uma reescrita completa do núcleo em Rust (pydantic-core), o que a torna entre 5× e 50× mais rápida que a v1 dependendo do modelo. Além do desempenho, a v2 traz TypeAdapter para validar tipos sem BaseModel, uniões discriminadas otimizadas, modo estrito (strict=True) e uma API renomeada (.model_dump() em vez de .dict(), @field_validator em vez de @validator). A v1 recebe apenas correções de segurança em 2026.

Pydantic é mais rápido que dataclasses ou attrs?

Para criar instâncias sem validação, dataclass e attrs são mais rápidos porque não fazem checagens. Mas se você precisa validar tipos, converter strings para datetime, ou aplicar regras de campo, Pydantic 2 vence com folga: a validação em Rust é mais rápida que qualquer código Python equivalente que você escrevesse manualmente. Como regra, use dataclass para estruturas internas confiáveis e Pydantic no edge do sistema (entrada, saída, deserialização).

Como validar um DataFrame Pandas com Pydantic?

Pydantic valida linhas, não DataFrames inteiros. O padrão é iterar as linhas (df.to_dict(orient="records")), validar cada uma com MinhaLinha.model_validate, e coletar erros. Para validação a nível de DataFrame (schema, distribuição, unicidade), use Pandera ou Great Expectations em vez de Pydantic. São ferramentas complementares, não substitutas.

Como validar uma lista de objetos JSON sem criar uma BaseModel wrapper?

Use TypeAdapter: adapter = TypeAdapter(list[MinhaModel]) seguido de adapter.validate_json(payload). O TypeAdapter aceita qualquer tipo Python (listas, dicionários, uniões, UUID, datetime) e é a maneira idiomática em v2 de validar coleções sem envolver uma classe extra apenas para o wrapping.

Vale a pena migrar código legado de Pydantic v1 para v2?

Sim. O FastAPI, LangChain, SQLModel e a maioria do ecossistema abandonaram a v1. Você ganha 5 a 17× em desempenho de validação, mensagens de erro melhores e suporte a Python 3.12 a 3.14. Use a ferramenta oficial bump-pydantic para reescrever a sintaxe automaticamente e cubra o resto com testes. A maioria das migrações leva um sprint.

Posso usar Pydantic 2 com código assíncrono e FastAPI?

Sim, e o FastAPI já exige v2 desde a versão 0.100. A validação em si é síncrona (roda em Rust, bloqueando a thread), então para payloads muito grandes (>1 MB) prefira validar em um pool com await asyncio.to_thread(Modelo.model_validate, payload). Para payloads pequenos, o custo é sub-milissegundo e a chamada direta no handler async é segura.