🎯 Object Mapping¶

@dataclass
class Customer:
    id: str
    name: str
    email: str
    phone: str

@dataclass
class CustomerDto:
    id: str      # Automatically mapped
    name: str    # Automatically mapped
    email: str   # Automatically mapped
    phone: str   # Automatically mapped

# Simple mapping - no configuration needed
mapper = Mapper()
customer = Customer("123", "Luigi Mario", "luigi@pizzeria.com", "+1-555-LUIGI")
customer_dto = mapper.map(customer, CustomerDto)

@dataclass
class Address:
    street_address: str
    city_name: str
    postal_code: str
    country_name: str

@dataclass
class AddressDto:
    address_line: str    # Combined field
    city: str           # Different name
    zip_code: str       # Different name
    country: str        # Different name

# Configure custom mappings
mapper.create_map(Address, AddressDto) \
    .map_member("address_line", lambda ctx: ctx.source.street_address) \
    .map_member("city", lambda ctx: ctx.source.city_name) \
    .map_member("zip_code", lambda ctx: ctx.source.postal_code) \
    .map_member("country", lambda ctx: ctx.source.country_name)

@dataclass
class MenuItem:
    name: str
    price: Decimal       # Decimal type
    available: bool
    category_id: int

@dataclass
class MenuItemDto:
    name: str
    price: float         # Converted to float
    available: str       # Converted to string
    category_id: str     # Converted to string

# Automatic type conversion
mapper = Mapper()
item = MenuItem("Margherita", Decimal("15.99"), True, 1)
item_dto = mapper.map(item, MenuItemDto)

assert item_dto.price == 15.99
assert item_dto.available == "True"
assert item_dto.category_id == "1"

from neuroglia.mapping.mapper import MappingProfile

class PizzeriaMappingProfile(MappingProfile):
    """Mapping profile for Mario's Pizzeria domain objects"""

    def configure(self):
        # Order mappings
        self.create_map(Order, OrderDto) \
            .map_member("customer", lambda ctx: ctx.source.customer_name) \
            .map_member("phone", lambda ctx: ctx.source.customer_phone) \
            .map_member("items", lambda ctx: self.map_list(ctx.source.pizzas, PizzaDto)) \
            .map_member("ordered_at", lambda ctx: ctx.source.order_time.isoformat()) \
            .map_member("total", lambda ctx: str(ctx.source.total_amount))

        # Pizza mappings
        self.create_map(Pizza, PizzaDto) \
            .map_member("price", lambda ctx: str(ctx.source.total_price)) \
            .map_member("prep_time", lambda ctx: ctx.source.preparation_time_minutes)

        # Customer mappings
        self.create_map(Customer, CustomerDto)  # Convention-based

        # Address mappings
        self.create_map(Address, AddressDto) \
            .map_member("address_line", lambda ctx: f"{ctx.source.street_address}") \
            .map_member("city", lambda ctx: ctx.source.city_name) \
            .map_member("zip_code", lambda ctx: ctx.source.postal_code)

# Register profile with mapper
mapper = Mapper()
mapper.add_profile(PizzeriaMappingProfile())

from typing import List, Dict

@dataclass
class Menu:
    sections: List[MenuSection]
    featured_items: Dict[str, Pizza]

@dataclass
class MenuDto:
    sections: List[MenuSectionDto]
    featured: Dict[str, PizzaDto]

# Configure collection mappings
mapper.create_map(Menu, MenuDto) \
    .map_member("sections", lambda ctx: mapper.map_list(ctx.source.sections, MenuSectionDto)) \
    .map_member("featured", lambda ctx: {
        k: mapper.map(v, PizzaDto)
        for k, v in ctx.source.featured_items.items()
    })

@dataclass
class OrderSummaryDto:
    id: str
    customer: str
    status: str
    total: str
    special_instructions: str  # Only for certain statuses

# Conditional member mapping
mapper.create_map(Order, OrderSummaryDto) \
    .map_member("special_instructions", lambda ctx:
        getattr(ctx.source, 'special_instructions', '')
        if ctx.source.status in [OrderStatus.COOKING, OrderStatus.READY]
        else None
    )

@dataclass
class OrderWithCustomer:
    id: str
    customer: Customer
    pizzas: List[Pizza]
    status: OrderStatus

@dataclass
class FlatOrderDto:
    order_id: str
    customer_name: str      # Flattened from customer
    customer_email: str     # Flattened from customer
    pizza_count: int        # Computed field
    status: str

# Flattening mapping
mapper.create_map(OrderWithCustomer, FlatOrderDto) \
    .map_member("order_id", lambda ctx: ctx.source.id) \
    .map_member("customer_name", lambda ctx: ctx.source.customer.name) \
    .map_member("customer_email", lambda ctx: ctx.source.customer.email) \
    .map_member("pizza_count", lambda ctx: len(ctx.source.pizzas))

import pytest
from neuroglia.mapping.mapper import Mapper

class TestPizzeriaMapping:

    def setup_method(self):
        self.mapper = Mapper()
        self.mapper.add_profile(PizzeriaMappingProfile())

    def test_pizza_to_dto_mapping(self):
        """Test Pizza to PizzaDto mapping"""
        # Arrange
        pizza = Pizza(
            id="pizza-123",
            name="Margherita",
            size="large",
            base_price=Decimal("15.99"),
            toppings=["basil", "mozzarella"],
            preparation_time_minutes=18
        )

        # Act
        pizza_dto = self.mapper.map(pizza, PizzaDto)

        # Assert
        assert pizza_dto.id == "pizza-123"
        assert pizza_dto.name == "Margherita"
        assert pizza_dto.price == "18.99"  # base_price + toppings
        assert pizza_dto.prep_time == 18
        assert pizza_dto.toppings == ["basil", "mozzarella"]

    def test_order_to_dto_mapping_preserves_structure(self):
        """Test complex Order to OrderDto mapping"""
        # Arrange
        order = create_sample_order_with_multiple_pizzas()

        # Act
        order_dto = self.mapper.map(order, OrderDto)

        # Assert
        assert order_dto.id == order.id
        assert order_dto.customer == order.customer_name
        assert len(order_dto.items) == len(order.pizzas)
        assert order_dto.total == str(order.total_amount)

    def test_mapping_handles_none_values(self):
        """Test mapping with None values"""
        # Arrange
        customer = Customer(
            id="123",
            name="Test Customer",
            email=None,  # None value
            phone="+1-555-TEST"
        )

        # Act
        customer_dto = self.mapper.map(customer, CustomerDto)

        # Assert
        assert customer_dto.email is None
        assert customer_dto.name == "Test Customer"

    def test_collection_mapping_preserves_order(self):
        """Test that collection mapping preserves order"""
        # Arrange
        pizzas = [
            create_pizza("Margherita"),
            create_pizza("Pepperoni"),
            create_pizza("Hawaiian")
        ]

        # Act
        pizza_dtos = self.mapper.map_list(pizzas, PizzaDto)

        # Assert
        assert len(pizza_dtos) == 3
        assert pizza_dtos[0].name == "Margherita"
        assert pizza_dtos[1].name == "Pepperoni"
        assert pizza_dtos[2].name == "Hawaiian"

from neuroglia.mvc import ControllerBase
from fastapi import HTTPException

class OrdersController(ControllerBase):
    def __init__(self,
                 service_provider: ServiceProviderBase,
                 mapper: Mapper,
                 mediator: Mediator,
                 order_service: OrderService):
        super().__init__(service_provider, mapper, mediator)
        self.order_service = order_service

    @get("/{order_id}")
    async def get_order(self, order_id: str) -> OrderDto:
        """Get order by ID with automatic DTO mapping"""
        order = await self.order_service.get_by_id_async(order_id)

        if not order:
            raise HTTPException(status_code=404, detail="Order not found")

        # Map domain entity to DTO
        return self.mapper.map(order, OrderDto)

    @post("/")
    async def create_order(self, create_order_request: CreateOrderRequest) -> OrderDto:
        """Create new order with request mapping"""
        # Map request to command
        command = self.mapper.map(create_order_request, CreateOrderCommand)

        # Execute command
        result = await self.mediator.execute_async(command)

        if not result.is_success:
            raise HTTPException(status_code=400, detail=result.error_message)

        # Map result to DTO
        return self.mapper.map(result.value, OrderDto)

from neuroglia.mediation import Command, CommandHandler

@dataclass
class CreateOrderRequest:
    customer_name: str
    customer_phone: str
    pizza_requests: List[PizzaRequest]

@dataclass
class CreateOrderCommand(Command[Order]):
    customer_name: str
    customer_phone: str
    pizza_items: List[PizzaOrderItem]

# Map request to command
class OrderMappingProfile(MappingProfile):
    def configure(self):
        self.create_map(CreateOrderRequest, CreateOrderCommand) \
            .map_member("pizza_items", lambda ctx:
                [self.map(req, PizzaOrderItem) for req in ctx.source.pizza_requests]
            )

from neuroglia.eventing import DomainEvent

@dataclass
class OrderStatusChangedEvent(DomainEvent):
    order_id: str
    old_status: str
    new_status: str
    customer_email: str
    notification_data: dict

class OrderEventService:
    def __init__(self, mapper: Mapper):
        self.mapper = mapper

    def create_status_change_event(self, order: Order, old_status: OrderStatus) -> OrderStatusChangedEvent:
        """Create event with mapped data"""

        # Map order data to event notification data
        notification_data = {
            "order_summary": self.mapper.map(order, OrderSummaryDto),
            "estimated_time": order.estimated_ready_time.isoformat(),
            "total_amount": str(order.total_amount)
        }

        return OrderStatusChangedEvent(
            order_id=order.id,
            old_status=old_status.value,
            new_status=order.status.value,
            customer_email=order.customer.email,
            notification_data=notification_data
        )

class OptimizedMappingService:
    def __init__(self, mapper: Mapper):
        self.mapper = mapper
        # Pre-compile mappings for better performance
        self._initialize_mappings()

    def _initialize_mappings(self):
        """Pre-configure frequently used mappings"""
        # Frequently used mappings
        self.mapper.create_map(Order, OrderDto)
        self.mapper.create_map(Pizza, PizzaDto)
        self.mapper.create_map(Customer, CustomerDto)

        # Warm up mapper with sample objects
        sample_order = create_sample_order()
        self.mapper.map(sample_order, OrderDto)

    def bulk_map_orders(self, orders: List[Order]) -> List[OrderDto]:
        """Efficiently map large collections"""
        return [self.mapper.map(order, OrderDto) for order in orders]

    def map_with_caching(self, source: Any, target_type: Type[T]) -> T:
        """Map with result caching for immutable objects"""
        cache_key = f"{type(source)}-{target_type}-{hash(source)}"

        if cache_key not in self._mapping_cache:
            self._mapping_cache[cache_key] = self.mapper.map(source, target_type)

        return self._mapping_cache[cache_key]

class OrderApiService:
    def __init__(self, mapper: Mapper, serializer: JsonSerializer):
        self.mapper = mapper
        self.serializer = serializer

    def export_orders_json(self, orders: List[Order]) -> str:
        """Export orders as JSON with DTO mapping"""
        # Map to DTOs first
        order_dtos = self.mapper.map_list(orders, OrderDto)

        # Then serialize
        return self.serializer.serialize_to_text(order_dtos)

    def import_orders_json(self, json_data: str) -> List[Order]:
        """Import orders from JSON with DTO mapping"""
        # Deserialize to DTOs
        order_dtos = self.serializer.deserialize_from_text(json_data, List[OrderDto])

        # Map to domain entities
        return self.mapper.map_list(order_dtos, Order)

from neuroglia.hosting import WebApplicationBuilder

def configure_mapping(builder: WebApplicationBuilder):
    """Configure object mapping services"""

    # Register mapper as singleton
    mapper = Mapper()

    # Add mapping profiles
    mapper.add_profile(PizzeriaMappingProfile())
    mapper.add_profile(CustomerMappingProfile())
    mapper.add_profile(EventMappingProfile())

    builder.services.add_singleton(Mapper, lambda: mapper)

    # Register mapping services
    builder.services.add_scoped(OrderMappingService)
    builder.services.add_scoped(CustomerMappingService)

# Usage in controllers
class MenuController(ControllerBase):
    def __init__(self,
                 service_provider: ServiceProviderBase,
                 mapper: Mapper,  # Injected automatically
                 mediator: Mediator):
        super().__init__(service_provider, mapper, mediator)