Data Transfer Objects (DTOs)

class AbstractDTO:
    config: DTOConfig

    @classmethod
    def create_for_field_definition(
        cls,
        field_definition: FieldDefinition,
        config: DTOConfig | None = None,
    ) -> type[AbstractDTO]:
        """
        Create a DTO class for a field definition.

        Parameters:
        - field_definition: Field definition to create DTO for
        - config: Optional DTO configuration

        Returns:
        DTO class configured for the field definition
        """

    @classmethod
    def generate_field_definitions(
        cls,
        model_type: type[Any],
    ) -> Generator[DTOFieldDefinition, None, None]:
        """Generate field definitions from model type."""

    def decode_bytes(self, raw: bytes) -> Any:
        """Decode bytes to Python object."""

    def decode_builtins(self, builtins: dict[str, Any] | list[Any]) -> Any:
        """Decode builtin types to Python object."""

    def encode_data(self, data: Any) -> dict[str, Any] | list[dict[str, Any]]:
        """Encode Python object to serializable data."""

class DTOConfig:
    def __init__(
        self,
        *,
        exclude: set[str] | None = None,
        include: set[str] | None = None,
        rename_fields: dict[str, str] | None = None,
        forbid_unknown_fields: bool = False,
        max_nested_depth: int = 1,
        partial: bool = False,
        underscore_fields_private: bool = True,
        experimental_features: list[DTOConfigFeatures] | None = None,
    ):
        """
        Configure DTO behavior.

        Parameters:
        - exclude: Fields to exclude from serialization
        - include: Fields to include in serialization (mutually exclusive with exclude)
        - rename_fields: Mapping of field names to rename
        - forbid_unknown_fields: Reject unknown fields in input data
        - max_nested_depth: Maximum depth for nested object serialization
        - partial: Allow partial updates (fields become optional)
        - underscore_fields_private: Treat underscore-prefixed fields as private
        - experimental_features: List of experimental features to enable
        """

    def create_for_field_definition(
        self,
        field_definition: FieldDefinition,
    ) -> DTOConfig:
        """Create config for specific field definition."""

class DataclassDTO(AbstractDTO):
    """DTO implementation for dataclasses."""

    @classmethod
    def create_for_field_definition(
        cls,
        field_definition: FieldDefinition,
        config: DTOConfig | None = None,
    ) -> type[DataclassDTO]:
        """Create DataclassDTO for field definition."""

class MsgspecDTO(AbstractDTO):
    """DTO implementation for msgspec Structs."""

    @classmethod
    def create_for_field_definition(
        cls,
        field_definition: FieldDefinition,
        config: DTOConfig | None = None,
    ) -> type[MsgspecDTO]:
        """Create MsgspecDTO for field definition."""

class DTOData:
    def __init__(
        self,
        *,
        data_as_builtins: dict[str, Any] | list[dict[str, Any]],
        data_as_bytes: bytes,
        data_as_model_instance: Any,
    ):
        """
        Container for DTO data in different formats.

        Parameters:
        - data_as_builtins: Data as built-in Python types
        - data_as_bytes: Data as encoded bytes
        - data_as_model_instance: Data as model instance
        """

    @property
    def as_builtins(self) -> dict[str, Any] | list[dict[str, Any]]:
        """Get data as built-in Python types."""

    @property
    def as_bytes(self) -> bytes:
        """Get data as encoded bytes."""

    @property
    def as_model_instance(self) -> Any:
        """Get data as model instance."""

class DTOFieldDefinition:
    def __init__(
        self,
        dto_field: DTOField,
        model_name: str,
        field_definition: FieldDefinition | None = None,
        default_value: Any = Empty,
    ):
        """
        Definition of a DTO field.

        Parameters:
        - dto_field: DTO field configuration
        - model_name: Name of the model this field belongs to
        - field_definition: Optional field definition
        - default_value: Default value for the field
        """

    @property
    def serialization_name(self) -> str:
        """Get the serialization name for this field."""

    @property
    def is_required(self) -> bool:
        """Check if this field is required."""

    @property
    def is_excluded(self) -> bool:
        """Check if this field is excluded from serialization."""

class DTOField:
    def __init__(
        self,
        mark: Mark | None = None,
        default: Any = Empty,
    ):
        """
        DTO field configuration.

        Parameters:
        - mark: Field marking for inclusion/exclusion
        - default: Default value for the field
        """

def dto_field(
    mark: Mark | None = None,
    default: Any = Empty,
) -> Any:
    """
    Create a DTO field with configuration.

    Parameters:
    - mark: Field marking for inclusion/exclusion
    - default: Default value for the field

    Returns:
    DTO field configuration
    """

class Mark(Enum):
    """Field marking for DTO behavior."""
    READ_ONLY = "read_only"
    WRITE_ONLY = "write_only"
    PRIVATE = "private"

class RenameStrategy(str, Enum):
    """Field renaming strategies."""
    UPPER = "upper"
    LOWER = "lower"
    CAMEL = "camel"
    PASCAL = "pascal"

def convert_case(value: str, strategy: RenameStrategy) -> str:
    """
    Convert field name case according to strategy.

    Parameters:
    - value: Original field name
    - strategy: Renaming strategy to apply

    Returns:
    Transformed field name
    """

from litestar import Litestar, post, get
from litestar.dto import DataclassDTO, DTOConfig
from dataclasses import dataclass
from typing import Optional

@dataclass
class User:
    name: str
    email: str
    age: int
    id: Optional[int] = None

# Create DTO for User dataclass
UserDTO = DataclassDTO[User]

@post("/users", dto=UserDTO)
def create_user(data: User) -> User:
    # data is automatically validated and converted from request JSON
    # Generate ID for new user
    data.id = 123
    return data

@get("/users/{user_id:int}", return_dto=UserDTO)
def get_user(user_id: int) -> User:
    # Response is automatically serialized using DTO
    return User(id=user_id, name="Alice", email="alice@example.com", age=30)

app = Litestar(route_handlers=[create_user, get_user])

from litestar.dto import DataclassDTO, DTOConfig

@dataclass
class UserWithPassword:
    name: str
    email: str
    password: str
    created_at: datetime
    id: Optional[int] = None

# Exclude sensitive fields from serialization
read_dto_config = DTOConfig(exclude={"password"})
UserReadDTO = DataclassDTO[UserWithPassword].with_config(read_dto_config)

# Only include necessary fields for creation
write_dto_config = DTOConfig(include={"name", "email", "password"})
UserWriteDTO = DataclassDTO[UserWithPassword].with_config(write_dto_config)

@post("/users", dto=UserWriteDTO, return_dto=UserReadDTO)
def create_user_secure(data: UserWithPassword) -> UserWithPassword:
    # Input: only name, email, password allowed
    # Output: password excluded, all other fields included
    data.id = 123
    data.created_at = datetime.utcnow()
    return data

from litestar.dto import DTOConfig, RenameStrategy

@dataclass
class APIResponse:
    user_id: int
    full_name: str
    email_address: str
    is_active: bool

# Convert snake_case to camelCase for API
camel_case_config = DTOConfig(
    rename_fields={
        "user_id": "userId",
        "full_name": "fullName", 
        "email_address": "emailAddress",
        "is_active": "isActive"
    }
)
APIResponseDTO = DataclassDTO[APIResponse].with_config(camel_case_config)

@get("/api/user/{user_id:int}", return_dto=APIResponseDTO)
def get_user_api(user_id: int) -> APIResponse:
    return APIResponse(
        user_id=user_id,
        full_name="Alice Smith",
        email_address="alice@example.com",
        is_active=True
    )
# Response JSON: {"userId": 123, "fullName": "Alice Smith", ...}

from litestar.dto import DTOConfig

@dataclass
class User:
    name: str
    email: str
    age: int
    id: int

# Allow partial updates - all fields become optional
partial_config = DTOConfig(partial=True)
UserPartialDTO = DataclassDTO[User].with_config(partial_config)

@patch("/users/{user_id:int}", dto=UserPartialDTO, return_dto=DataclassDTO[User])
def update_user(user_id: int, data: User) -> User:
    # Only provided fields will be present in data
    # Merge with existing user data
    existing_user = get_user_from_db(user_id)
    
    # Update only provided fields
    if hasattr(data, 'name') and data.name is not None:
        existing_user.name = data.name
    if hasattr(data, 'email') and data.email is not None:
        existing_user.email = data.email
    if hasattr(data, 'age') and data.age is not None:
        existing_user.age = data.age
    
    return existing_user

from dataclasses import dataclass
from typing import List

@dataclass
class Address:
    street: str
    city: str
    country: str

@dataclass
class Company:
    name: str
    address: Address

@dataclass
class Employee:
    name: str
    company: Company
    addresses: List[Address]

# Configure nested depth
nested_config = DTOConfig(max_nested_depth=2)
EmployeeDTO = DataclassDTO[Employee].with_config(nested_config)

@get("/employees/{emp_id:int}", return_dto=EmployeeDTO)
def get_employee(emp_id: int) -> Employee:
    return Employee(
        name="John Doe",
        company=Company(
            name="Tech Corp",
            address=Address("123 Main St", "City", "Country")
        ),
        addresses=[
            Address("456 Home St", "Home City", "Country"),
            Address("789 Vacation Ave", "Vacation City", "Country")
        ]
    )

import msgspec
from litestar.dto import MsgspecDTO

@msgspec.defstruct
class Product:
    name: str
    price: float
    category: str
    in_stock: bool = True

ProductDTO = MsgspecDTO[Product]

@post("/products", dto=ProductDTO, return_dto=ProductDTO)
def create_product(data: Product) -> Product:
    # msgspec provides very fast serialization
    return data

@get("/products", return_dto=MsgspecDTO[List[Product]])
def list_products() -> List[Product]:
    return [
        Product("Laptop", 999.99, "Electronics"),
        Product("Book", 29.99, "Education", False)
    ]

from litestar.dto import AbstractDTO, DTOConfig
from typing import TypeVar, Generic

T = TypeVar("T")

class CustomDTO(AbstractDTO, Generic[T]):
    """Custom DTO with additional validation."""
    
    def decode_builtins(self, builtins: dict[str, Any] | list[Any]) -> T:
        # Custom deserialization logic
        data = super().decode_builtins(builtins)
        
        # Add custom validation
        if hasattr(data, 'email') and '@' not in data.email:
            raise ValueError("Invalid email format")
        
        return data
    
    def encode_data(self, data: T) -> dict[str, Any] | list[dict[str, Any]]:
        # Custom serialization logic
        encoded = super().encode_data(data)
        
        # Add metadata
        if isinstance(encoded, dict):
            encoded['_serialized_at'] = datetime.utcnow().isoformat()
        
        return encoded

@post("/custom", dto=CustomDTO[User])
def create_with_custom_dto(data: User) -> dict:
    return {"status": "created", "user": data}

from litestar.exceptions import ValidationException

@dataclass
class CreateUserRequest:
    name: str
    email: str
    age: int

    def __post_init__(self):
        # Custom validation in dataclass
        if not self.name.strip():
            raise ValueError("Name cannot be empty")
        if self.age < 0:
            raise ValueError("Age must be positive")
        if '@' not in self.email:
            raise ValueError("Invalid email format")

UserCreateDTO = DataclassDTO[CreateUserRequest]

@post("/users/validated", dto=UserCreateDTO)
def create_user_validated(data: CreateUserRequest) -> dict:
    # DTO automatically handles validation
    # Any validation errors are converted to ValidationException
    return {"message": f"User {data.name} created successfully"}

# Custom exception handler for DTO validation errors
def dto_validation_handler(request: Request, exc: ValidationException) -> Response:
    return Response(
        content={
            "error": "Validation failed",
            "details": exc.extra,
            "message": exc.detail
        },
        status_code=422
    )

# Generic DTO type
DTOType = TypeVar("DTOType", bound=AbstractDTO)

# DTO configuration features
class DTOConfigFeatures(str, Enum):
    EXPERIMENTAL_GENERIC_SUPPORT = "experimental_generic_support"

# Field definition types
FieldDefinition = Any  # From litestar._signature module

# Default values
class Empty:
    """Sentinel value for empty/unset fields."""
    
# Encoding types  
EncodedData = bytes | str | dict[str, Any] | list[dict[str, Any]]
DecodedData = Any