git:20260606.cac7da7 to git:20260723.07bc8a4

---

name: msgspec

---

# msgspec Skill

## Code Style Rules

- Use PEP 604 for unions: `T | None` (not `Optional[T]`)

- **`from __future__ import annotations` rule** — Library/shared modules that define runtime-introspected `msgspec.Struct` subclasses should avoid postponed annotations unless the consuming tool resolves them. Consumer modules that only use Structs MAY use future annotations.

- Use `kw_only=True` for Structs with more than 2 fields

## Quick Reference

### Struct Definition

```python

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):

"""frozen=True: immutable + hashable. gc=False: skip GC for short-lived objects."""

event_type: str

payload: dict[str, object]

# Keyword-only (recommended for >2 fields)

class Config(msgspec.Struct, kw_only=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

class ApiResponse(msgspec.Struct, rename="camel"):

user_id: int # serialized as "userId"

created_at: str # serialized as "createdAt"

# Reject unknown fields at API boundaries

class StrictInput(msgspec.Struct, forbid_unknown_fields=True):

name: str

value: int

```

### Validation Constraints

```python

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}$")]

# 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

```

### Serialization

```python

import msgspec

# JSON -- singleton encoder/decoder (cache these!)

encoder = msgspec.json.Encoder()

decoder = msgspec.json.Decoder(User)

data = encoder.encode(user) # bytes

user = decoder.decode(b'{"id":1,"name":"Alice"}')

# Functional API (convenience, slightly slower)

data = msgspec.json.encode(user)

user = msgspec.json.decode(b'...', type=User)

# MessagePack (binary, more compact)

data = msgspec.msgpack.encode(user)

user = msgspec.msgpack.decode(data, type=User)

def enc_hook(obj: object) -> object:

encoder = msgspec.json.Encoder(enc_hook=enc_hook)

decoder = msgspec.json.Decoder(MyStruct, dec_hook=dec_hook)

```

### Canonical Litestar serializers (match-your-stack)

Litestar apps typically need `to_json(value, as_bytes=True)` that handles UUID / datetime / Enum / Decimal for Channels broadcasts, log contexts, and JSONB writes. Pick the branch that matches your project.

**Branch A — sqlspec is in-stack.** Re-export sqlspec's serializer; it already installs an `enc_hook` covering UUID, datetime, Enum, Decimal, Pydantic, dataclasses, attrs, and msgspec.Struct.

```python

# myapp/utils/serialization.py

from sqlspec.utils.serializers import from_json, to_json

__all__ = ("from_json", "to_json")

```

Usage:

```python

from myapp.utils.serialization import to_json

payload = to_json(order, as_bytes=True)

await backend.publish(payload, channels=[f"orders:{order.id}:events"])

```

```python

# myapp/utils/serialization.py

from typing import Any

import msgspec

def to_json(value: Any) -> bytes:

if isinstance(value, bytes):

return value

return _encoder.encode(value)

```

### Type Coercion with convert()

```python

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

user = msgspec.convert(raw, User, strict=False) # id coerced to 42

# str_keys: dict keys are strings (useful for JSON-loaded dicts)

data = {"1": "Alice", "2": "Bob"}

result = msgspec.convert(data, dict[int, str], str_keys=True)

from dataclasses import dataclass

@dataclass

class LegacyUser:

id: int

name: str

legacy = LegacyUser(id=1, name="Alice")

```

### Dynamic Struct Creation

```python

import msgspec

# Runtime struct from field definitions

fields = [

("id", int),

("name", str),

("score", Annotated[float, Meta(ge=0.0)]),

]

DynamicModel = msgspec.defstruct("DynamicModel", fields, kw_only=True)

# With defaults

fields_with_defaults = [

("id", int),

("active", bool, True), # (name, type, default)

]

FlexModel = msgspec.defstruct("FlexModel", fields_with_defaults)

```

### Tagged Unions (Discriminated Unions)

```python

import msgspec

# Default tag field is "type", tag value is the class name

class Dog(msgspec.Struct, tag=True):

name: str

breed: str

class Cat(msgspec.Struct, tag=True):

name: str

indoor: bool

Animal = Dog | Cat

# Deserialize: inspects "type" field to pick correct class

animal = msgspec.json.decode(b'{"type":"Dog","name":"Rex","breed":"Lab"}', type=Animal)

# Custom tag values

class CreateEvent(msgspec.Struct, tag="create"):

resource: str

class DeleteEvent(msgspec.Struct, tag="delete"):

resource: str

soft: bool = True

Event = CreateEvent | DeleteEvent

# Custom tag field name

class V1Request(msgspec.Struct, tag="v1", tag_field="version"):

payload: str

class V2Request(msgspec.Struct, tag="v2", tag_field="version"):

payload: str

metadata: dict[str, str] = {}

Request = V1Request | V2Request

```

<workflow>

## Workflow

### Step 1: Define Structs

Create msgspec Structs for all data shapes. Use `kw_only=True` for Structs with more than 2 fields. Use `frozen=True` for immutable value objects. Use `forbid_unknown_fields=True` for API-boundary input validation.

### Step 2: Add Constraints

Annotate fields with `Annotated[Type, Meta(...)]` for numeric ranges, string lengths, and regex patterns. Define reusable constraint aliases at module level to avoid repetition.

### Step 3: Choose Serialization Strategy

### Step 4: Handle Polymorphism

Use tagged unions (`tag=True` or `tag="value"`) for discriminated unions. Define a union type alias (`Event = CreateEvent | DeleteEvent`) and decode against it. Use `tag_field` to customize the discriminator field name.

### Step 5: Validate

Test round-trip encode/decode. Confirm `ValidationError` is raised for constraint violations. Verify tag dispatch selects the correct Struct type for all union variants.

</workflow>

<guardrails>

## Guardrails

- **Use `kw_only=True` for Structs with >2 fields** -- prevents positional argument confusion and makes instantiation self-documenting.

- **Use `forbid_unknown_fields=True` at API boundaries** -- rejects payloads with unexpected keys, preventing silent data loss.

- **Use `gc=False` for short-lived, non-circular objects** -- eliminates GC overhead for hot-path objects like request/response shapes.

- **Tagged unions for polymorphism** -- faster than manual dispatch and eliminates `isinstance` chains.

- **`from __future__ import annotations` rule** — Library/shared modules that define runtime-introspected types (advanced-alchemy models, sqlspec configs, msgspec Structs, dishka providers) avoid postponed annotations unless their consumers resolve them. Consumer applications MAY use it. The restriction applies only to modules that define introspected types, not handler/service/test modules that use them.

- **Use `strict=False` only at trust boundaries** -- lax coercion is useful for converting legacy dicts but can mask type errors in internal code.

- **Prefer `sqlspec.utils.serializers.to_json` when sqlspec is in-stack** — its built-in enc_hook covers UUID, datetime, Enum, Decimal, Pydantic, msgspec.Struct, dataclasses, and attrs in one import. Hand-rolling is only needed when sqlspec is not a dependency.

</guardrails>

<validation>

### Validation Checkpoint

Before delivering msgspec code, verify:

- [ ] All Struct fields have explicit type annotations

- [ ] If this library/shared module defines runtime-introspected types, avoid `from __future__ import annotations` unless all consumers resolve postponed annotations. Consumer modules may use it.

- [ ] Encoder/Decoder instances are module-level singletons (not created per-request)

- [ ] API-boundary Structs use `forbid_unknown_fields=True`

- [ ] Numeric/string constraints use `Meta` (not manual `if` checks)

- [ ] Tagged union tag values are unique across all variants in a union

- [ ] `kw_only=True` on Structs with more than 2 fields

- [ ] If sqlspec is in-stack, to_json is imported from sqlspec.utils.serializers (not hand-rolled)

</validation>

<example>

## Example

**Task:** Define an event system with tagged unions, constraints, and JSON serialization.

```python

# Library/shared modules that define runtime-introspected Structs usually avoid postponed annotations.

```

```python

# events.py

import uuid

import msgspec

from msgspec import Meta

# --- Constraint aliases ---

NonEmptyStr = Annotated[str, Meta(min_length=1, max_length=255)]

PositiveInt = Annotated[int, Meta(gt=0)]

# --- Event variants (tagged union) ---

class UserCreatedEvent(msgspec.Struct, tag="user.created", tag_field="event_type", kw_only=True, gc=False):

event_id: uuid.UUID

user_id: PositiveInt

email: NonEmptyStr

occurred_at: datetime

class UserDeletedEvent(msgspec.Struct, tag="user.deleted", tag_field="event_type", kw_only=True, gc=False):

event_id: uuid.UUID

user_id: PositiveInt

occurred_at: datetime

reason: str | None = None

UserEvent = UserCreatedEvent | UserDeletedEvent

def encode_event(event: UserEvent) -> bytes:

return _encoder.encode(event)

def decode_event(data: bytes) -> UserEvent:

return _decoder.decode(data)

# --- Usage ---

event = UserCreatedEvent(

event_id=uuid.uuid4(),

user_id=42,

email="alice@example.com",

)

payload = encode_event(event)

# b'{"event_type":"user.created","event_id":"...","user_id":42,"email":"alice@example.com","occurred_at":"..."}'

recovered = decode_event(payload)

assert isinstance(recovered, UserCreatedEvent)

```

</example>

---

## References Index

For detailed guides and reference tables, refer to the following documents in `references/`:

- **[Meta Constraints Reference](references/constraints.md)** -- Full table of all Meta constraint parameters with examples for numeric, string, bytes, and OpenAPI metadata.

- **[Tagged Union Patterns](references/tagged-unions.md)** -- Discriminated union patterns: default tags, custom tag fields/values, nested unions, API versioning, and event systems.

- **[Litestar Patterns](references/litestar-patterns.md)** — CamelizedBaseStruct, sqlspec-vs-manual to_json branches, hybrid msgspec + Pydantic schema pattern, `__post_init__` validation for Litestar apps.

---

## Official References

## Shared Styleguide Baseline

- Use shared styleguides for generic language/framework rules to reduce duplication in this skill.

- [General Principles](../litestar-styleguide/references/general.md)

- [Python](../litestar-styleguide/references/python.md)

- Keep this skill focused on tool-specific workflows, edge cases, and integration details.