litestar-org/litestar-skills/msgspec

import msgspec
# Basic struct
class User(msgspec.Struct):
    id: int
    name: str
    email: str | None = None
# Performance options
class Event(msgspec.Struct, frozen=True, gc=False, cache_hash=True):
    """frozen=True: immutable + hashable. gc=False: skip GC tracking."""
    event_type: str
    payload: dict[str, object]
# Keyword-only + omit defaults on wire (recommended for >2 fields / compact payloads)
class Config(msgspec.Struct, kw_only=True, omit_defaults=True, repr_omit_defaults=True):
    host: str
    port: int = 5432
    ssl: bool = False
# Array-like encoding (tuple encoding, more compact)
class Point(msgspec.Struct, array_like=True):
    x: float
    y: float
# Rename fields for serialization ("lower", "upper", "camel", "pascal", "kebab", callable, or mapping)
class ApiResponse(msgspec.Struct, rename="camel"):
    user_id: int  # serialized as "userId"
    created_at: str  # serialized as "createdAt"
# Per-field wire alias and default_factory via msgspec.field()
class Resource(msgspec.Struct):
    resource_id: int = msgspec.field(name="id")
    tags: list[str] = msgspec.field(default_factory=list)
# Reject unknown fields at API boundaries
class StrictInput(msgspec.Struct, forbid_unknown_fields=True):
    name: str
    value: int

import msgspec
class UserPatch(msgspec.Struct, kw_only=True, forbid_unknown_fields=True):
    name: str | msgspec.UnsetType = msgspec.UNSET
    email: str | None | msgspec.UnsetType = msgspec.UNSET
patch = msgspec.json.decode(b'{"email": null}', type=UserPatch)
assert patch.name is msgspec.UNSET  # omitted by client
assert patch.email is None  # explicitly cleared by client
assert msgspec.json.encode(patch) == b'{"email":null}'

from datetime import datetime
from typing import Annotated
import msgspec
from msgspec import Meta
class Product(msgspec.Struct):
    name: Annotated[str, Meta(min_length=1, max_length=100)]
    price: Annotated[float, Meta(gt=0)]
    quantity: Annotated[int, Meta(ge=0, le=10_000)]
    sku: Annotated[str, Meta(pattern=r"^[A-Z]{2}-\d{4}$")]
    batch_size: Annotated[int, Meta(multiple_of=5)]
    expires_at: Annotated[datetime, Meta(tz=True)]
# Reusable constraint aliases
PositiveInt = Annotated[int, Meta(gt=0)]
NonEmptyStr = Annotated[str, Meta(min_length=1)]
Percentage = Annotated[float, Meta(ge=0.0, le=100.0)]
class Order(msgspec.Struct):
    id: PositiveInt
    label: NonEmptyStr
    discount: Percentage = 0.0

import msgspec
# JSON -- singleton encoder/decoder (cache these!)
# Encoder kwargs: enc_hook=None, decimal_format="string"|"number"|Callable[[Decimal], Any], uuid_format="canonical"|"hex", order=None|"deterministic"|"sorted"
# Decoder kwargs: type=Any, *, strict=True, dec_hook=None, float_hook=None
encoder = msgspec.json.Encoder()
decoder = msgspec.json.Decoder(User)
data = encoder.encode(user)  # bytes
user = decoder.decode(b'{"id":1,"name":"Alice"}')
# Newline-delimited JSON (NDJSON) and zero-allocation buffer writes
ndjson = encoder.encode_lines([user])
users = decoder.decode_lines(ndjson)
buf = bytearray()
encoder.encode_into(user, buf, 0)
pretty = msgspec.json.format(data, indent=2)
# Functional API (convenience, slightly slower)
data = msgspec.json.encode(user, order="deterministic")
user = msgspec.json.decode(data, type=User)
# MessagePack (binary, more compact; uuid_format also accepts "bytes"; Decoder accepts ext_hook and msgspec.msgpack.Ext)
mp_data = msgspec.msgpack.encode(user)
user = msgspec.msgpack.decode(mp_data, type=User)
# YAML (requires msgspec[yaml] / PyYAML) and TOML (requires msgspec[toml] / tomli_w + tomllib/tomli)
yaml_bytes = msgspec.yaml.encode(user)
user_from_yaml = msgspec.yaml.decode(yaml_bytes, type=User)
toml_bytes = msgspec.toml.encode(user)
user_from_toml = msgspec.toml.decode(toml_bytes, type=User)
# Deferred decoding / zero-copy payload slicing with msgspec.Raw
class Envelope(msgspec.Struct):
    kind: str
    payload: msgspec.Raw
env = msgspec.json.decode(b'{"kind":"user","payload":{"id":1,"name":"Alice"}}', type=Envelope)
detached_raw = env.payload.copy()  # detach from input buffer if retaining long-term
inner_user = msgspec.json.decode(detached_raw, type=User)
# Hooks are only for unsupported custom types. datetime, date, time, timedelta,
# UUID, Decimal, and Enum are already supported natively.
def enc_hook(obj: object) -> object:
    if isinstance(obj, complex):
        return (obj.real, obj.imag)
    raise NotImplementedError(f"Unsupported type: {type(obj)}")
def dec_hook(target_type: type, obj: object) -> object:
    if target_type is complex:
        real, imag = obj
        return complex(real, imag)
    raise NotImplementedError(f"Unsupported type: {target_type}")
custom_encoder = msgspec.json.Encoder(enc_hook=enc_hook)
custom_decoder = msgspec.json.Decoder(MyStruct, dec_hook=dec_hook)

# myapp/utils/serialization.py
from sqlspec.utils.serializers import from_json, to_json
__all__ = ("from_json", "to_json")

from myapp.utils.serialization import to_json
payload = to_json(order, as_bytes=True)
await backend.publish(payload, channels=[f"orders:{order.id}:events"])

# myapp/utils/serialization.py
from typing import Any
import msgspec
_encoder = msgspec.json.Encoder()
def to_json(value: Any) -> bytes:
    if isinstance(value, bytes):
        return value
    return _encoder.encode(value)

import msgspec
raw = {"id": "42", "name": "Alice"}  # id is a string
# Strict mode (default): raises on type mismatch
user = msgspec.convert(raw, User)  # ValidationError: id must be int
# Lax mode: coerces compatible types (e.g. str -> int, 0/1 -> bool, epoch int/float -> datetime)
user = msgspec.convert(raw, User, strict=False)  # id coerced to 42
# str_keys: dict keys are strings (useful for JSON/TOML-loaded dicts)
data = {"1": "Alice", "2": "Bob"}
result = msgspec.convert(data, dict[int, str], str_keys=True)
# Convert with dec_hook or custom builtin_types passthrough
measurement = msgspec.convert(raw_measurement, Measurement, dec_hook=dec_hook, builtin_types=(bytes,))
# Convert a dataclass, attrs instance, or ORM model to a Struct by reading attributes
from dataclasses import dataclass
@dataclass
class LegacyUser:
    id: int
    name: str
legacy = LegacyUser(id=1, name="Alice")
user = msgspec.convert(legacy, User, from_attributes=True)
# Convert a Struct or supported type graph back to Python builtin types
builtins_dict = msgspec.to_builtins(user, str_keys=True, order="deterministic")

import copy
import msgspec
# Shallow conversions (keyed by Python attribute names; preserves nested Structs and UNSET values)
field_map = msgspec.structs.asdict(user)
field_tuple = msgspec.structs.astuple(user)
# Copy-and-replace (both call __post_init__ as of msgspec 0.21.0)
updated = msgspec.structs.replace(user, name="Bob")
updated_std = copy.replace(user, name="Bob")
# Mutate a frozen=True Struct (e.g. inside __post_init__ to normalize a field)
msgspec.structs.force_setattr(event, "event_type", event.event_type.lower())