Instance emission (Pydantic, TypedDict, ORM)

What you’ll learn: How to emit Pydantic models or ORM objects directly instead of plain dicts, enabling validation and type safety.

Prerequisites: Basic understanding of TableEmit and Field from the README Quick Start.

What is instance emission?

Instead of getting back plain dicts from your mapping, you can tell etielle to create your Pydantic models, TypedDicts, or ORM objects directly.

This means you get:

  • Validated data: Pydantic validates as it builds
  • Type safety: Your IDE knows the exact type of each instance
  • ORM integration: Create database objects without manual conversion

Progressive construction means you can have multiple traversals updating the same instance. For example:

  • Traversal 1 sets id and name from users array
  • Traversal 2 adds email from profiles array
  • Both updates merge into one User instance with matching join_keys

TableEmit vs. InstanceEmit

With TableEmit (basic approach):

from etielle.core import MappingSpec, TraversalSpec, TableEmit, Field
from etielle.transforms import get
from etielle.executor import run_mapping

data = {"users": [{"id": "u1", "email": "alice@example.com"}]}

spec = MappingSpec(traversals=[TraversalSpec(
    path=["users"],
    mode="auto",
    emits=[TableEmit(
        table="users",
        join_keys=[get("id")],
        fields=[Field("id", get("id")), Field("email", get("email"))]
    )]
)])
result = run_mapping(data, spec)
# result["users"].instances = {("u1",): {"id": "u1", "email": "alice@example.com"}}
#                                        ↑ plain dict

With InstanceEmit (typed approach):

from etielle.core import MappingSpec, TraversalSpec, field_of
from etielle.transforms import get
from etielle.instances import InstanceEmit, FieldSpec, PydanticBuilder
from etielle.executor import run_mapping
from pydantic import BaseModel

class User(BaseModel):
    id: str
    email: str

data = {"users": [{"id": "u1", "email": "alice@example.com"}]}

spec = MappingSpec(traversals=[TraversalSpec(
    path=["users"],
    mode="auto",
    emits=[InstanceEmit[User](
        table="users",
        join_keys=[get("id")],
        builder=PydanticBuilder(User),
        fields=[
            FieldSpec(selector=field_of(User, lambda u: u.id), transform=get("id")),
            FieldSpec(selector=field_of(User, lambda u: u.email), transform=get("email")),
        ]
    )]
)])
result = run_mapping(data, spec)
# result["users"].instances = {("u1",): User(id="u1", email="alice@example.com")}
#                                        ↑ Pydantic model instance

Builders

from etielle.instances import InstanceEmit, FieldSpec, PydanticBuilder, TypedDictBuilder
from etielle.transforms import get
from etielle.core import MappingSpec, TraversalSpec, field_of
from pydantic import BaseModel

class User(BaseModel):
    id: str
    email: str

emit = InstanceEmit[User](
    table="users",
    join_keys=[get("id")],
    fields=[
        FieldSpec(selector=field_of(User, lambda u: u.id), transform=get("id")),
        FieldSpec(selector=field_of(User, lambda u: u.email), transform=get("email")),
    ],
    builder=PydanticBuilder(User),
)

# Minimal runnable demo
root = {"users": [{"id": "u1", "email": "alice@example.com"}]}
mapping = MappingSpec(traversals=[TraversalSpec(path=["users"], mode="auto", emits=[emit])])
from etielle.executor import run_mapping
res = run_mapping(root, mapping)
print(sorted([(k, v.email) for k, v in res["users"].instances.items()]))
[(('u1',), 'alice@example.com')]

TypedDict without Pydantic:

from typing import TypedDict

class UserTD(TypedDict):
    id: str
    email: str

emit_td = InstanceEmit[UserTD](
    table="users",
    join_keys=[get("id")],
    fields=[
        FieldSpec(selector="id", transform=get("id")),
        FieldSpec(selector="email", transform=get("email")),
    ],
    builder=TypedDictBuilder(lambda d: UserTD(**d)),
)

# Minimal runnable demo
root = {"users": [{"id": "u1", "email": "alice@example.com"}]}
mapping = MappingSpec(traversals=[TraversalSpec(path=["users"], mode="auto", emits=[emit_td])])
from etielle.executor import run_mapping
res = run_mapping(root, mapping)
print(list(res["users"].instances.values()))
[{'id': 'u1', 'email': 'alice@example.com'}]

Choosing a builder

  • PydanticBuilder(Model): Use for Pydantic models with validation
  • PydanticPartialBuilder(Model): Use when some fields might be missing (creates partial models)
  • TypedDictBuilder(factory): Use for plain dicts or when you don’t want Pydantic
  • Custom builder: Implement the builder protocol for ORM objects or custom types

Strictness and error collection

Builders collect update-time and finalize-time errors; the executor returns them in MappingResult per table.

from etielle.executor import run_mapping

result = run_mapping(root_err, mapping_err)
mr = result["users"]
print(mr.update_errors)
print(mr.finalize_errors)
{('u1',): ["table=users key=('u1',) field emali: unknown field; did you mean email?"]}
{('u1',): ["table=users key=('u1',) 1 validation error for User\nemail\n  Field required [type=missing, input_value={'id': 'u1'}, input_type=dict]\n    For further information visit https://errors.pydantic.dev/2.12/v/missing"]}

Merge policies

By default, if two traversals update the same field, the last write wins. You can change this behavior—see Merge policies for details.

Reference

  • InstanceEmit[T]
  • FieldSpec[T]
  • PydanticBuilder, PydanticPartialBuilder, TypedDictBuilder

See also