mappers

name	mappers
description	Mappers pattern for persistence. Use when converting between domain aggregates and database rows, handling nested value objects, child entities, and polymorphic types.
user-invocable	false
disable-model-invocation	false

Mappers

Type: Primary

Purpose

• Convert between domain aggregates/value objects and database representations.
• Handle nested value objects and entities through nested mapper delegation.
• Normalize data during conversion when needed (e.g., datetime serialization).

Aggregate Mappers

Full Aggregate Mapper (mapper.py.j2)

The standard aggregate mapper. Each framework-managed block renders only when the aggregate's domain class actually declares the matching attributes; business scalars and JSONB value-object attributes enumerate dynamically. One template covers aggregates with any combination of: tenant, status, timestamps, polymorphic nested data, and arbitrary scalar/VO business fields.

• to_dict(aggregate) converts domain aggregate to dictionary for database insertion.
• from_row(row) converts single database row to domain aggregate.
• Handles optional nested polymorphic fields (kind + entity pattern).

Minimal Aggregate Mapper (mapper_minimal.py.j2)

For simple aggregates without status or timestamps.

• to_dict(aggregate) converts only id, tenant_id, and additional fields.
• from_row(row) reconstructs aggregate with default value handling.
• Use when aggregate has no status tracking or audit timestamps.

Aggregate Mapper with Children (mapper_with_children.py.j2)

For aggregates that own child entities stored in separate tables.

• to_dict(aggregate) converts only the aggregate's own fields (not children).
• from_row(row) reconstructs aggregate without children (for simple lookups).
• from_rows(aggregate_row, child_rows) reconstructs full aggregate with children.

Value Object Mappers

Simple Value Object Mapper (value_object_mapper.py.j2)

For value objects stored in JSONB columns. Generic over the VO's attribute list — works for any number of Guard-declared attributes.

• to_json(value_object) converts to JSON-compatible dict; serializes datetime/date attributes via .isoformat().
• from_json(data) reconstructs value object; parses datetime/date attributes via datetime.fromisoformat.
• Signatures track the optionality contract (see below): | None is added to parameter and return types iff the owning aggregate's attribute is Optional (or the JSONB column is nullable).

Complex Value Object Mapper (complex_value_object_mapper.py.j2)

For value objects containing multiple optional nested value objects.

• Uses getattr(obj, "attr", None) pattern for optional fields.
• Delegates to nested field mapper for each optional field.
• Handles None checks on deserialization with data.get("field").

Value Object with Collection Mapper (value_object_with_collection_mapper.py.j2)

For value objects containing a collection of nested value objects.

• Handles both scalar fields and collection fields.
• Iterates over collection for serialization/deserialization.
• Uses hasattr() check for optional collections.

Entity Mappers

Child Entity Mapper (child_entity_mapper.py.j2)

For entities owned by an aggregate, stored in child tables.

• to_dict(entity, parent_id, tenant_id) includes parent context in output.
• from_row(row) reconstructs entity without parent reference.

Polymorphic Mapper (polymorphic_mapper.py.j2)

For type hierarchies stored in JSONB with a discriminator field.

• to_json(kind, entity) dispatches to type-specific mapper based on discriminator.
• from_json(kind, data) reconstructs correct type based on discriminator.
• Returns None when kind or data is missing.

Optionality Contract

A value-object or polymorphic mapper's signature must match the nullability of the owning aggregate's attribute (or, equivalently, the JSONB column it persists into):

• If the aggregate declares the attribute as Guard[VO](VO) (required, non-nullable column) → mapper signature is to_json(vo: VO) -> dict[str, Any] and from_json(data: dict[str, Any]) -> VO. No None short-circuits.
• If the aggregate declares it as Guard[VO | None](VO) / Optional[...] (or the column is nullable) → mapper signature is to_json(vo: VO | None) -> dict[str, Any] | None and from_json(data: dict[str, Any] | None) -> VO | None, with if vo is None: return None / if data is None: return None guards.

The Polymorphic Mapper is always rendered as Optional — its discriminator pair (<attr>_kind, <attr>_data) is conventionally nullable.

Datetime Handling

JSONB columns require explicit datetime serialization:

import contextlib
from datetime import datetime

# Serialization
if isinstance(value, datetime):
    value = value.isoformat()

# Deserialization (preferred pattern)
if isinstance(value, str) and "T" in value:
    with contextlib.suppress(ValueError, TypeError):
        value = datetime.fromisoformat(value)

Testing guidance

• Write unit tests that verify round-trip conversion: aggregate → dict → aggregate.
• Test optional attribute handling (None values, missing attributes).
• Test nested mapper delegation for child entities and value objects.
• Verify datetime serialization produces ISO format strings.
• Test collection mappers with empty, single, and multiple items.

---

Template

Full Aggregate Mapper

Rendering rules:

• Imports. Always emit from collections.abc import Mapping and from typing import Any (used by the rendered annotations). Emit one from .{{ vo_mapper_module }} import {{ vo_mapper_class }} per entry in {{ business_vo_attributes }}. Emit from .{{ nested_mapper }} import {{ nested_mapper_class }} only when {{ has_polymorphic_field }} is true. Do not emit from datetime import datetime or import contextlib — the canonical body round-trips datetimes through SQLAlchemy DateTime(timezone=True) columns without local conversion.
• Multi-tenancy. When {{ is_multi_tenant }} is true, emit the tenant_id row in to_dict's returned dict and the tenant_id=row["{{ tenant_id_column }}"] kwarg in from_row's constructor call. When false, omit both.
• Status block. When {{ has_status }} is true, emit the status and status_error rows in to_dict, the status_error = row.get("{{ status_error_column }}") helper before the constructor call, and the status_value= and status_error= kwargs in from_row. When false, omit all of those.
• Timestamps block. When {{ has_timestamps }} is true, emit created_at and updated_at rows in to_dict and matching kwargs in from_row. When false, omit both.
• Business scalar attributes. For each entry <attr> in {{ business_scalar_attributes }} (declaration order from the aggregate's class body, excluding identity/framework-managed/polymorphic/VO attrs), emit one direct-passthrough row in to_dict ("<attr>": <agg>.<attr>,) and one kwarg in from_row (<attr>=row["<attr>"]).
• Business value-object attributes. For each entry (<attr>, <vo_mapper_module>, <vo_mapper_class>, <vo_fields>) in {{ business_vo_attributes }} (declaration order, JSONB-resident VOs composed by the aggregate, excluding the polymorphic field if any):
  • In to_dict, emit one mapper-delegated row: "<attr>": <vo_mapper_class>.to_json(<agg>.<attr>),. The aggregate's <attr> property returns the Guard-projected VO instance, which the VO mapper can serialize directly.
  • In from_row, emit one VO-reconstruction helper line before the constructor call: <attr> = <vo_mapper_class>.from_json(row["<attr>"]).
  • In the constructor call, emit one kwarg per entry in <vo_fields> (the VO's own primitive attributes, in declaration order): <vo_field>=<attr>.<vo_field>,. This matches the framework's domain-spec:flat-constructor-arguments convention — the aggregate's __init__ accepts the VO's primitives as flat kwargs and re-projects them into the VO internally (e.g. self.details = Details.new(name, description)), so the mapper must pass the primitives, not the VO instance.
  • Each VO requires a sibling-mapper import.
• Polymorphic block. When {{ has_polymorphic_field }} is true, emit the if/else dispatch in to_dict and the <field>_entity / <field>_kind resolution in from_row plus the matching kwargs in the constructor call. When false, omit all of those.
• Constructor argument order. id_, tenant_id (if multi-tenant), polymorphic params (if any), status_value/status_error (if status), business scalars (declaration order), business VO flat-fields (one VO at a time, fields in VO declaration order), created_at/updated_at (if timestamps).

{{# imports — only the ones the rendered body actually uses #}}
from collections.abc import Mapping
from typing import Any

from {{ domain_module }} import {{ aggregate_name }}

{{# for vo_attr, vo_mapper_module, vo_mapper_class in business_vo_attributes #}}
from .{{ vo_mapper_module }} import {{ vo_mapper_class }}
{{# end for #}}
{{# if has_polymorphic_field #}}
from .{{ nested_mapper }} import {{ nested_mapper_class }}
{{# end if #}}

__all__ = ["{{ mapper_class }}"]

class {{ mapper_class }}:
    @staticmethod
    def to_dict({{ aggregate_name_lower }}: {{ aggregate_name }}) -> dict[str, Any]:
        result: dict[str, Any] = {
            "{{ id_column }}": {{ aggregate_name_lower }}.id,
            {{# if is_multi_tenant #}}
            "{{ tenant_id_column }}": {{ aggregate_name_lower }}.tenant_id,
            {{# end if #}}
            {{# if has_status #}}
            "{{ status_column }}": {{ aggregate_name_lower }}.status.status,
            "{{ status_error_column }}": {{ aggregate_name_lower }}.status.error if {{ aggregate_name_lower }}.status.error else None,
            {{# end if #}}
            {{# for scalar_attr in business_scalar_attributes #}}
            "{{ scalar_attr }}": {{ aggregate_name_lower }}.{{ scalar_attr }},
            {{# end for #}}
            {{# for vo_attr, vo_mapper_module, vo_mapper_class in business_vo_attributes #}}
            "{{ vo_attr }}": {{ vo_mapper_class }}.to_json({{ aggregate_name_lower }}.{{ vo_attr }}),
            {{# end for #}}
            {{# if has_timestamps #}}
            "created_at": {{ aggregate_name_lower }}.created_at,
            "updated_at": {{ aggregate_name_lower }}.updated_at,
            {{# end if #}}
        }

        {{# if has_polymorphic_field #}}
        if {{ aggregate_name_lower }}.{{ nested_field }}:
            result["{{ nested_kind_column }}"] = {{ aggregate_name_lower }}.{{ nested_field }}.kind
            result["{{ nested_data_column }}"] = {{ nested_mapper_class }}.to_json(
                {{ aggregate_name_lower }}.{{ nested_field }}.kind,
                {{ aggregate_name_lower }}.{{ nested_field }}.entity,
            )
        else:
            result["{{ nested_kind_column }}"] = None
            result["{{ nested_data_column }}"] = None
        {{# end if #}}

        return result

    @staticmethod
    def from_row(row: Mapping[str, Any]) -> {{ aggregate_name }}:
        {{# if has_polymorphic_field #}}
        {{ nested_field }}_entity = {{ nested_mapper_class }}.from_json(
            row.get("{{ nested_kind_column }}"),
            row.get("{{ nested_data_column }}"),
        )
        {{ nested_field }}_kind = row.get("{{ nested_kind_column }}")
        {{# end if #}}

        {{# if has_status #}}
        status_error = row.get("{{ status_error_column }}")
        {{# end if #}}

        {{# for vo_attr, vo_mapper_module, vo_mapper_class, vo_fields in business_vo_attributes #}}
        {{ vo_attr }} = {{ vo_mapper_class }}.from_json(row["{{ vo_attr }}"])
        {{# end for #}}

        return {{ aggregate_name }}(
            id_=row["{{ id_column }}"],
            {{# if is_multi_tenant #}}
            tenant_id=row["{{ tenant_id_column }}"],
            {{# end if #}}
            {{# if has_polymorphic_field #}}
            {{ nested_kind_param }}={{ nested_field }}_kind,
            {{ nested_entity_param }}={{ nested_field }}_entity,
            {{# end if #}}
            {{# if has_status #}}
            status_value=row["{{ status_column }}"],
            status_error=status_error,
            {{# end if #}}
            {{# for scalar_attr in business_scalar_attributes #}}
            {{ scalar_attr }}=row["{{ scalar_attr }}"],
            {{# end for #}}
            {{# for vo_attr, vo_mapper_module, vo_mapper_class, vo_fields in business_vo_attributes #}}
            {{# for vo_field in vo_fields #}}
            {{ vo_field }}={{ vo_attr }}.{{ vo_field }},
            {{# end for #}}
            {{# end for #}}
            {{# if has_timestamps #}}
            created_at=row["created_at"],
            updated_at=row["updated_at"],
            {{# end if #}}
        )

Minimal Aggregate Mapper

Rendering rules:

• Multi-tenancy. When {{ is_multi_tenant }} is true, emit the tenant_id row in to_dict and the tenant_id=row["{{ tenant_id_column }}"] kwarg in from_row's constructor call. When false, omit both.

from collections.abc import Mapping
from typing import Any

from {{ domain_module }} import {{ aggregate_name }}

__all__ = ["{{ mapper_class }}"]

class {{ mapper_class }}:
    @staticmethod
    def to_dict({{ aggregate_name_lower }}: {{ aggregate_name }}) -> dict[str, Any]:
        return {
            "{{ id_column }}": {{ aggregate_name_lower }}.id,
            {{# if is_multi_tenant #}}
            "{{ tenant_id_column }}": {{ aggregate_name_lower }}.tenant_id,
            {{# end if #}}
            "{{ additional_column }}": {{ aggregate_name_lower }}.{{ additional_attribute }},
        }

    @staticmethod
    def from_row(row: Mapping[str, Any]) -> {{ aggregate_name }}:
        {{ additional_attribute }} = row.get("{{ additional_column }}")
        if {{ additional_attribute }} is None:
            {{ additional_attribute }} = {{ additional_default }}

        return {{ aggregate_name }}(
            id_=row["{{ id_column }}"],
            {{# if is_multi_tenant #}}
            tenant_id=row["{{ tenant_id_column }}"],
            {{# end if #}}
            {{ additional_attribute }}={{ additional_attribute }},
        )

Aggregate Mapper with Children

Rendering rules:

• Multi-tenancy. When {{ is_multi_tenant }} is true, emit the tenant_id row in to_dict and the tenant_id=... kwargs in both from_row and from_rows constructor calls. When false, omit them.

from collections.abc import Mapping
from typing import Any

from {{ domain_module }} import {{ aggregate_name }}

from .{{ child_mapper_module }} import {{ child_mapper_class }}

__all__ = ["{{ mapper_class }}"]

class {{ mapper_class }}:
    @staticmethod
    def to_dict({{ aggregate_name_lower }}: {{ aggregate_name }}) -> dict[str, Any]:
        return {
            "{{ id_column }}": {{ aggregate_name_lower }}.id,
            {{# if is_multi_tenant #}}
            "{{ tenant_id_column }}": {{ aggregate_name_lower }}.tenant_id,
            {{# end if #}}
            "{{ additional_column }}": {{ aggregate_name_lower }}.{{ additional_attribute }},
        }

    @staticmethod
    def from_row(row: Mapping[str, Any]) -> {{ aggregate_name }}:
        {{ additional_attribute }} = row.get("{{ additional_column }}")
        if {{ additional_attribute }} is None:
            {{ additional_attribute }} = {{ additional_default }}

        return {{ aggregate_name }}(
            id_=row["{{ id_column }}"],
            {{# if is_multi_tenant #}}
            tenant_id=row["{{ tenant_id_column }}"],
            {{# end if #}}
            {{ additional_attribute }}={{ additional_attribute }},
            {{ children_attribute }}=None,
        )

    @staticmethod
    def from_rows(aggregate_row: Mapping[str, Any], child_rows: list[Mapping[str, Any]]) -> {{ aggregate_name }}:
        children = [{{ child_mapper_class }}.from_row(row) for row in child_rows]

        {{ additional_attribute }} = aggregate_row.get("{{ additional_column }}")
        if {{ additional_attribute }} is None:
            {{ additional_attribute }} = {{ additional_default }}

        return {{ aggregate_name }}(
            id_=aggregate_row["{{ id_column }}"],
            {{# if is_multi_tenant #}}
            tenant_id=aggregate_row["{{ tenant_id_column }}"],
            {{# end if #}}
            {{ additional_attribute }}={{ additional_attribute }},
            {{ children_attribute }}=children,
        )

Simple Value Object Mapper

Rendering rules:

• Imports. Emit from datetime import datetime only when at least one entry in {{ attributes }} has a datetime/date type. Emit import contextlib only if the chosen from_json body uses contextlib.suppress (the canonical body below does not — it relies on isinstance checks instead).
• Optionality. When {{ optional }} is true, append | None to the parameter type and to the return type of both to_json and from_json, and prepend the corresponding if X is None: return None guard. When false, emit neither.
• Attribute loop. For each attribute in {{ attributes }} (declaration order from the VO's class body), emit one entry in to_json's returned dict and one kwarg in from_json's constructor call. Use the datetime branch for datetime/date types, the plain branch otherwise. Do not emit unused branches.

{{# imports — only the ones the body actually uses #}}
from typing import Any

from {{ domain_module }} import {{ value_object_name }}

__all__ = ["{{ mapper_class }}"]

class {{ mapper_class }}:
    @staticmethod
    def to_json({{ value_object_name_lower }}: {{ value_object_name }}{{ "| None" if optional }}) -> dict[str, Any]{{ "| None" if optional }}:
        {{ "if " ~ value_object_name_lower ~ " is None: return None" if optional }}

        return {
            {{# for attr in attributes #}}
            {{# datetime branch  #}} "{{ attr }}": {{ value_object_name_lower }}.{{ attr }}.isoformat() if {{ value_object_name_lower }}.{{ attr }} is not None else None,
            {{# plain branch     #}} "{{ attr }}": {{ value_object_name_lower }}.{{ attr }},
            {{# end for          #}}
        }

    @staticmethod
    def from_json(data: dict[str, Any]{{ "| None" if optional }}) -> {{ value_object_name }}{{ "| None" if optional }}:
        {{ "if data is None: return None" if optional }}

        return {{ value_object_name }}(
            {{# for attr in attributes #}}
            {{# datetime branch  #}} {{ attr }}=datetime.fromisoformat(data["{{ attr }}"]) if isinstance(data["{{ attr }}"], str) else data["{{ attr }}"],
            {{# plain branch     #}} {{ attr }}=data["{{ attr }}"],
            {{# end for          #}}
        )

Complex Value Object Mapper

from typing import Any

from {{ domain_module }} import {{ value_object_name }}

from .{{ field_mapper_module }} import {{ field_mapper_class }}

__all__ = ["{{ mapper_class }}"]

class {{ mapper_class }}:
    @staticmethod
    def to_json({{ value_object_name_lower }}: {{ value_object_name }}) -> dict[str, Any]:
        return {
            "{{ type_field }}": {{ value_object_name_lower }}.{{ type_field }},
            "{{ field_1 }}": {{ field_mapper_class }}.to_json(getattr({{ value_object_name_lower }}, "{{ field_1 }}", None)),
            "{{ field_2 }}": {{ field_mapper_class }}.to_json(getattr({{ value_object_name_lower }}, "{{ field_2 }}", None)),
            "{{ field_3 }}": {{ field_mapper_class }}.to_json(getattr({{ value_object_name_lower }}, "{{ field_3 }}", None)),
            "{{ field_4 }}": {{ field_mapper_class }}.to_json(getattr({{ value_object_name_lower }}, "{{ field_4 }}", None)),
            "{{ field_5 }}": {{ field_mapper_class }}.to_json(getattr({{ value_object_name_lower }}, "{{ field_5 }}", None)),
        }

    @staticmethod
    def from_json(data: dict[str, Any]) -> {{ value_object_name }}:
        return {{ value_object_name }}(
            {{ type_field }}=data["{{ type_field }}"],
            {{ field_1 }}={{ field_mapper_class }}.from_json(data["{{ field_1 }}"]) if data.get("{{ field_1 }}") else None,
            {{ field_2 }}={{ field_mapper_class }}.from_json(data["{{ field_2 }}"]) if data.get("{{ field_2 }}") else None,
            {{ field_3 }}={{ field_mapper_class }}.from_json(data["{{ field_3 }}"]) if data.get("{{ field_3 }}") else None,
            {{ field_4 }}={{ field_mapper_class }}.from_json(data["{{ field_4 }}"]) if data.get("{{ field_4 }}") else None,
            {{ field_5 }}={{ field_mapper_class }}.from_json(data["{{ field_5 }}"]) if data.get("{{ field_5 }}") else None,
        )

Value Object with Collection Mapper

from typing import Any

from {{ domain_module }} import {{ value_object_name }}, {{ nested_item_name }}

from .{{ field_mapper_module }} import {{ field_mapper_class }}

__all__ = ["{{ mapper_class }}"]

class {{ mapper_class }}:
    @staticmethod
    def to_json({{ value_object_name_lower }}: {{ value_object_name }}) -> dict[str, Any]:
        {{ collection_field }}_data = []
        if hasattr({{ value_object_name_lower }}, "{{ collection_field }}") and {{ value_object_name_lower }}.{{ collection_field }}:
            for item in {{ value_object_name_lower }}.{{ collection_field }}:
                {{ collection_field }}_data.append({
                    "{{ item_field_1 }}": {{ field_mapper_class }}.to_json(getattr(item, "{{ item_field_1 }}", None)),
                    "{{ item_field_2 }}": {{ field_mapper_class }}.to_json(getattr(item, "{{ item_field_2 }}", None)),
                    "{{ item_field_3 }}": {{ field_mapper_class }}.to_json(getattr(item, "{{ item_field_3 }}", None)),
                })

        return {
            "{{ type_field }}": {{ value_object_name_lower }}.{{ type_field }},
            "{{ scalar_field_1 }}": {{ field_mapper_class }}.to_json(getattr({{ value_object_name_lower }}, "{{ scalar_field_1 }}", None)),
            "{{ scalar_field_2 }}": {{ field_mapper_class }}.to_json(getattr({{ value_object_name_lower }}, "{{ scalar_field_2 }}", None)),
            "{{ collection_field }}": {{ collection_field }}_data,
        }

    @staticmethod
    def from_json(data: dict[str, Any]) -> {{ value_object_name }}:
        {{ collection_field }}_list = []
        if data.get("{{ collection_field }}"):
            for item_data in data["{{ collection_field }}"]:
                {{ collection_field }}_list.append(
                    {{ nested_item_name }}(
                        {{ item_field_1 }}={{ field_mapper_class }}.from_json(item_data["{{ item_field_1 }}"]) if item_data.get("{{ item_field_1 }}") else None,
                        {{ item_field_2 }}={{ field_mapper_class }}.from_json(item_data["{{ item_field_2 }}"]) if item_data.get("{{ item_field_2 }}") else None,
                        {{ item_field_3 }}={{ field_mapper_class }}.from_json(item_data["{{ item_field_3 }}"]) if item_data.get("{{ item_field_3 }}") else None,
                    )
                )

        return {{ value_object_name }}(
            {{ type_field }}=data["{{ type_field }}"],
            {{ scalar_field_1 }}={{ field_mapper_class }}.from_json(data["{{ scalar_field_1 }}"]) if data.get("{{ scalar_field_1 }}") else None,
            {{ scalar_field_2 }}={{ field_mapper_class }}.from_json(data["{{ scalar_field_2 }}"]) if data.get("{{ scalar_field_2 }}") else None,
            {{ collection_field }}={{ collection_field }}_list,
        )

Child Entity Mapper

Rendering rules:

• Multi-tenancy. When {{ is_multi_tenant }} is true, include the tenant_id: str parameter on to_dict and emit the "{{ tenant_id_column }}": tenant_id row in the returned dict. When false, drop the parameter and omit the row.
• Status block. When {{ has_status }} is true, emit the status and status_error rows in to_dict, the status_error = row.get("{{ status_error_column }}") helper before the constructor call, and the status_value= and status_error= kwargs in from_row. When false, omit all of those.

from collections.abc import Mapping
from typing import Any

from {{ domain_module }} import {{ entity_name }}

__all__ = ["{{ mapper_class }}"]

class {{ mapper_class }}:
    @staticmethod
    def to_dict({{ entity_name_lower }}: {{ entity_name }}, {{ parent_id_param }}: str{{ ", tenant_id: str" if is_multi_tenant }}) -> dict[str, Any]:
        return {
            "{{ id_column }}": {{ entity_name_lower }}.id,
            "{{ parent_id_column }}": {{ parent_id_param }},
            {{# if is_multi_tenant #}}
            "{{ tenant_id_column }}": tenant_id,
            {{# end if #}}
            "{{ additional_column }}": {{ entity_name_lower }}.{{ additional_attribute }},
            {{# if has_status #}}
            "{{ status_column }}": {{ entity_name_lower }}.status.status,
            "{{ status_error_column }}": {{ entity_name_lower }}.status.error if {{ entity_name_lower }}.status.error else None,
            {{# end if #}}
        }

    @staticmethod
    def from_row(row: Mapping[str, Any]) -> {{ entity_name }}:
        {{# if has_status #}}
        status_error = row.get("{{ status_error_column }}")
        {{# end if #}}

        return {{ entity_name }}(
            id_=row["{{ id_column }}"],
            {{ additional_attribute }}=row["{{ additional_column }}"],
            {{# if has_status #}}
            status_value=row["{{ status_column }}"],
            status_error=status_error,
            {{# end if #}}
        )

Polymorphic Mapper

from typing import Any

from {{ domain_module }} import {{ type_a_name }}, {{ type_b_name }}

from .{{ type_a_mapper_module }} import {{ type_a_mapper_class }}
from .{{ type_b_mapper_module }} import {{ type_b_mapper_class }}

__all__ = ["{{ mapper_class }}"]

class {{ mapper_class }}:
    @staticmethod
    def to_json(kind: str, entity: {{ type_a_name }} | {{ type_b_name }}) -> dict[str, Any]:
        if kind == "{{ type_a_discriminator }}":
            return {{ type_a_mapper_class }}.to_json(entity)
        else:
            return {{ type_b_mapper_class }}.to_json(entity)

    @staticmethod
    def from_json(kind: str | None, data: dict[str, Any] | None) -> {{ type_a_name }} | {{ type_b_name }} | None:
        if not kind or not data:
            return None

        if kind == "{{ type_a_discriminator }}":
            return {{ type_a_mapper_class }}.from_json(data)
        else:
            return {{ type_b_mapper_class }}.from_json(data)

Placeholders

Placeholder	Description	Example
{{ domain_module }}	Domain module path	domain.aggregates, domain.value_objects
{{ aggregate_name }}	Aggregate class name	Order, Profile
{{ aggregate_name_lower }}	Aggregate name in snake_case	order, profile
{{ mapper_class }}	Mapper class name	OrderMapper, ProfileMapper
{{ is_multi_tenant }}	Whether the owning aggregate carries tenant_id (Section 1 Multi-tenant flag, or presence of a tenant_id attribute on the domain class)	true, false
{{ has_status }}	Whether the domain class declares a status attribute typed as a <<Value Object>> (the framework Status VO convention)	true, false
{{ has_timestamps }}	Whether the domain class declares both created_at and updated_at datetime attributes	true, false
{{ has_polymorphic_field }}	Whether the domain class declares a Union-typed (A | B) attribute that the polymorphic mapper dispatches on	true, false
{{ business_scalar_attributes }}	Ordered list of non-identity, non-framework-managed, non-VO scalar attribute names on the domain class	["enabled"], ["count", "label"]
{{ business_vo_attributes }}	Ordered list of (<attr>, <vo_mapper_module>, <vo_mapper_class>, <vo_fields>) quadruples — one per JSONB-resident value-object attribute on the domain class, excluding the polymorphic field. <vo_fields> is the ordered list of the VO's own primitive attribute names, used to expand the aggregate's flat-constructor kwargs.	[("details", "details_mapper", "DetailsMapper", ["name", "description"])]
{{ id_column }}	Primary key column name	id, order_id
{{ tenant_id_column }}	Tenant ID column name	tenant_id
{{ status_column }}	Status column name	status
{{ status_error_column }}	Status error column name	status_error
{{ nested_field }}	Nested polymorphic field name	info, metadata
{{ nested_kind_column }}	Nested kind column name	info_kind
{{ nested_data_column }}	Nested data column name	info_data
{{ nested_mapper_class }}	Nested mapper class	InfoMapper
{{ nested_kind_param }}	Constructor parameter for kind	info_kind
{{ nested_entity_param }}	Constructor parameter for entity	info_entity
{{ additional_column }}	Additional column name	name, description
{{ additional_attribute }}	Additional attribute name	name, description
{{ additional_default }}	Default value for optional attribute	"", [], None
{{ child_mapper_module }}	Child mapper module path	mappers.order_item_mapper
{{ child_mapper_class }}	Child mapper class	OrderItemMapper
{{ children_attribute }}	Children collection attribute	items, addresses
{{ value_object_name }}	Value object class name	OrderInfo, Address
{{ value_object_name_lower }}	Value object name in snake_case	order_info, address
{{ attributes }}	Ordered list of the VO's Guard-declared attribute names (Simple VO Mapper)	["name", "description"]
{{ optional }}	Whether the mapper signature carries | None (per Optionality Contract)	true, false
{{ field_mapper_module }}	Field mapper module	mappers.field_mapper
{{ field_mapper_class }}	Field mapper class	FieldMapper
{{ type_field }}	Type discriminator field	type, kind
{{ field_1 }} through {{ field_5 }}	Optional field names	field1, field2
{{ collection_field }}	Collection field name	items, tags
{{ nested_item_name }}	Nested item class name	OrderItem, Tag
{{ item_field_1 }} through {{ item_field_3 }}	Item field names	name, value
{{ scalar_field_1 }}, {{ scalar_field_2 }}	Scalar field names	name, description
{{ entity_name }}	Entity class name	OrderItem, Address
{{ entity_name_lower }}	Entity name in snake_case	order_item, address
{{ parent_id_param }}	Parent ID parameter name	order_id, parent_id
{{ parent_id_column }}	Parent ID column name	order_id, parent_id
{{ type_a_name }}, {{ type_b_name }}	Polymorphic type names	TypeA, TypeB
{{ type_a_mapper_module }}	Type A mapper module	mappers.type_a_mapper
{{ type_a_mapper_class }}	Type A mapper class	TypeAMapper
{{ type_b_mapper_module }}	Type B mapper module	mappers.type_b_mapper
{{ type_b_mapper_class }}	Type B mapper class	TypeBMapper
{{ type_a_discriminator }}	Type A discriminator value	"type_a", "standard"