Dataclasses with PyMediate

Type-safe, low-boilerplate requests and responses using Python dataclasses — validation, immutability, and mixins.

PyMediate requests and responses are plain classes, which makes Python's dataclasses the natural way to define them: auto-generated __init__/__repr__/__eq__, full mypy support, frozen=True immutability, and __post_init__ validation — all with minimal boilerplate.

from dataclasses import dataclass
from pymediate import Request

@dataclass
class CreateUserResponse:
    user_id: int
    username: str

@dataclass
class CreateUserRequest(Request[CreateUserResponse]):
    username: str
    email: str

The manual equivalent is three methods of hand-written __init__, __repr__, and __eq__ — verbose and easy to get subtly wrong.

Field patterns

Every field needs a type hint (that's a dataclass rule, and PyMediate's type inspection relies on it). Beyond that, the usual dataclass toolkit applies directly:

# Optional fields and defaults
@dataclass
class UpdateUserRequest(Request[UserResponse]):
    user_id: int
    username: str | None = None
    email: str | None = None

@dataclass
class SearchRequest(Request[SearchResponse]):
    query: str
    page: int = 1
    per_page: int = 10
    sort_by: str = "relevance"

Keep secrets out of logs with field(repr=False):

from dataclasses import dataclass, field

@dataclass
class LoginRequest(Request[LoginResponse]):
    username: str
    password: str = field(repr=False)

print(LoginRequest(username="alice", password="secret123"))
# LoginRequest(username='alice')

Mutable defaults need default_factory

tags: list[str] = [] shares one list between every instance. Always use field(default_factory=list) (or dict, or a custom function) for mutable defaults.

import uuid
from datetime import datetime

@dataclass
class TrackedRequest(Request[Response]):
    action: str
    tags: list[str] = field(default_factory=list)
    request_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: datetime = field(default_factory=datetime.now)

Validation with `__post_init__`

Validate at request construction, so an invalid request never reaches a handler:

@dataclass
class CreateUserRequest(Request[UserResponse]):
    username: str
    email: str
    age: int

    def __post_init__(self):
        if not self.username:
            raise ValueError("Username cannot be empty")
        if "@" not in self.email:
            raise ValueError("Invalid email format")
        if self.age < 18:
            raise ValueError("Must be 18 or older")

__post_init__ also works for normalization:

@dataclass
class SearchRequest(Request[SearchResponse]):
    query: str
    filters: list[str]

    def __post_init__(self):
        self.query = self.query.strip().lower()
        self.filters = list(set(self.filters))

For a frozen dataclass, plain assignment in __post_init__ raises FrozenInstanceError — frozen dataclasses override __setattr__ to block it. Use object.__setattr__ for one-time normalization:

@dataclass(frozen=True)
class SearchRequest(Request[SearchResponse]):
    query: str
    filters: list[str]

    def __post_init__(self):
        object.__setattr__(self, "query", self.query.strip().lower())
        object.__setattr__(self, "filters", list(set(self.filters)))

Frozen dataclasses

frozen=True makes requests immutable — no mutation in flight, and instances become hashable, so they work as cache keys:

@dataclass(frozen=True)
class CacheableRequest(Request[Response]):
    user_id: int
    include_details: bool

req = CacheableRequest(user_id=123, include_details=True)
requests_cache[req] = response   # usable as a dict key
req.user_id = 456                # FrozenInstanceError

For high-volume request types, add slots=True as well — it cuts per-instance memory substantially:

@dataclass(slots=True, frozen=True)
class LogEventRequest(Request[LogResponse]):
    event_type: str
    timestamp: datetime
    data: dict

Nested dataclasses

Group related fields into value objects instead of flat field lists:

@dataclass
class OrderItem:
    product_id: int
    quantity: int
    price: float

@dataclass
class ShippingAddress:
    street: str
    city: str
    postal_code: str

@dataclass
class CreateOrderRequest(Request[OrderResponse]):
    user_id: int
    items: list[OrderItem]
    shipping_address: ShippingAddress
    notes: str | None = None

Nested dataclasses validate themselves — each __post_init__ runs when its own instance is constructed:

@dataclass
class Coordinates:
    latitude: float
    longitude: float

    def __post_init__(self):
        if not -90 <= self.latitude <= 90:
            raise ValueError("Latitude must be between -90 and 90")
        if not -180 <= self.longitude <= 180:
            raise ValueError("Longitude must be between -180 and 180")

Mixins

Dataclass mixins share fields and validation across request types:

@dataclass
class PaginationMixin:
    page: int = 1
    per_page: int = 10

    def __post_init__(self):
        if self.page < 1:
            raise ValueError("Page must be >= 1")
        if not 1 <= self.per_page <= 100:
            raise ValueError("Per page must be between 1 and 100")

@dataclass
class SearchUsersRequest(PaginationMixin, Request[SearchResponse]):
    query: str

req = SearchUsersRequest(query="alice", page=2, per_page=20)

Polymorphic request hierarchies

A shared abstract base groups a family of requests:

from abc import ABC

@dataclass
class BaseNotificationRequest(Request[NotificationResponse], ABC):
    user_id: int
    message: str

@dataclass
class EmailNotificationRequest(BaseNotificationRequest):
    email: str
    subject: str

@dataclass
class SMSNotificationRequest(BaseNotificationRequest):
    phone_number: str

This is also what makes selective pipeline behaviors powerful: one PipelineBehavior[BaseNotificationRequest] automatically covers every request in the hierarchy — including channels you add later:

class NotificationMetricsBehavior(PipelineBehavior[BaseNotificationRequest]):
    def __init__(self, metrics):
        self.metrics = metrics

    def __call__(self, request, next):
        start = time.perf_counter()
        response = next()
        self.metrics.histogram(
            "notification.delivery_seconds",
            time.perf_counter() - start,
            tags={"channel": type(request).__name__},
        )
        return response

Useful response shapes

Generic helpers built with PEP 695 syntax compose well with responses:

# Result type — success/failure as data
@dataclass
class Success[T]:
    value: T

@dataclass
class Failure:
    error: str
    error_code: str

type Result[T] = Success[T] | Failure

# Paginated response
@dataclass
class PaginatedResponse[T]:
    items: list[T]
    total: int
    page: int
    per_page: int

    @property
    def total_pages(self) -> int:
        return (self.total + self.per_page - 1) // self.per_page

Best practices

Type every field — required by dataclasses, relied on by PyMediate.
Prefer frozen=True for requests — immutability prevents in-flight mutation bugs and enables hashing.
Use default_factory for mutable defaults — never = [] or = {}.
Validate in __post_init__, not in the handler — fail before dispatch, not after.
Name fields descriptively — search_query, not q.
Group related fields into nested value objects (PriceRange, Address).

Next steps

Requests and responses — message design patterns
Type safety — how static and runtime checks work together
Handlers — processing these requests

Dataclasses with PyMediate

On this page