Skip to content

04 – Schema and Payload

Schema declares the allowed shape of a Payload, and is enforced whenever a State is constructed or forked.

🎯 Goal

Validate a document with a nested address object, using unions for optional fields.

🧱 The schema

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from dagpipe import Payload, Schema

AddressSchema = Schema({
    "city": str,
    "zip": int | None,
})

UserSchema = Schema({
    "name": str,
    "address": AddressSchema,
})

DeepItemSchema = Schema({
    "raw": object,
    "user": UserSchema,
})

Outer keys inside a Schema tree point at one of:

  • a type (str, int, ...) — single allowed type
  • a union (str | None) — any of the members
  • a nested Schema — the value must be a mapping following that sub-schema
  • object — unrestricted

✅ Valid payloads

1
2
3
4
5
6
7
8
ok = Payload({
    "raw": {"anything": "goes"},
    "user": {
        "name": "John",
        "address": {"city": "Mumbai", "zip": 400001},
    },
})
DeepItemSchema.validate_payload(ok)   # no-op

💥 Violations

validate_payload raises SchemaError for any of:

Case Example Error
Undefined key "phone" not declared Invalid path 'phone' not defined in schema
Wrong scalar type "name": 42 Path 'user.name' must be str
Union violation "zip": "abc" Path 'user.address.zip' must be one of (int, NoneType)
Non-container for nested schema "user": "joe" Path 'user' must be a container 
1
2
3
4
5
6
7
8
from dagpipe import SchemaError

try:
    DeepItemSchema.validate_payload(
        Payload({"raw": {}, "user": {"name": 42, "address": {"city": "M"}}})
    )
except SchemaError as e:
    print(e)   # Path 'user.name' must be str

🔍 Updates are validated too

State.fork(payload_update=...) calls validate_update, which checks that every dot-path in the update is declared — before anything is copied:

1
2
UserSchema.validate_update({"address.city": "Paris"})   # OK
UserSchema.validate_update({"address.country": "FR"})   # SchemaError: not in schema

🔍 Dot-path access

Payload gives typed, immutable access to nested values:

1
2
3
4
5
6
7
p = Payload({"user": {"address": {"city": "Mumbai", "zip": None}}})

p.get("user.address.city")     # 'Mumbai'
p.get("user.address.zip")      # None
p.has("user.address.zip")      # True
p.has("user.phone")            # False
p.update({"user.phone": "123"})  # Schema may reject this on the State level!

Payload.update itself is schema-unaware — the shape check belongs to the State/Schema layer. Use State.fork when you want validation.

💡 Tips

  • Schema is immutable and reusable across all state instances.
  • SchemaError raised during __post_init__ means a State can never exist with an invalid payload — catch it early at construction.
  • Keep union syntax consistent with your Python version (PEP-604 str | None for 3.10+, Union[str, None] otherwise).