Serialization and Validation

import dataclasses
import datetime as dt
import enum
import json
import uuid
from pathlib import Path
from typing import Any

def to_jsonable(obj: Any) -> Any:
    """`default=` hook for json.dumps: handles what asdict leaves behind."""
    if dataclasses.is_dataclass(obj) and not isinstance(obj, type):
        return {f.name: getattr(obj, f.name) for f in dataclasses.fields(obj)}   # shallow, json recurses
    if isinstance(obj, enum.Enum):
        return obj.value
    if isinstance(obj, (dt.datetime, dt.date, dt.time)):
        return obj.isoformat()
    if isinstance(obj, (uuid.UUID, Path)):
        return str(obj)
    if isinstance(obj, (set, frozenset)):
        return sorted(obj)
    if hasattr(obj, "tolist"):          # numpy arrays / scalars, torch tensors (cpu)
        return obj.tolist()
    raise TypeError(f"{type(obj).__name__} is not JSON serializable")

def dumps(obj: Any, **kw: Any) -> str:
    return json.dumps(obj, default=to_jsonable, ensure_ascii=False, **kw)

import dataclasses
from typing import Any

def export(obj: Any, *, drop_none: bool = True) -> dict[str, Any]:
    out: dict[str, Any] = {}
    for f in dataclasses.fields(obj):
        if f.name.startswith("_") or f.metadata.get("export") is False:
            continue
        v = getattr(obj, f.name)
        if drop_none and v is None:
            continue
        if dataclasses.is_dataclass(v) and not isinstance(v, type):
            v = export(v, drop_none=drop_none)
        elif isinstance(v, list):
            v = [export(x, drop_none=drop_none) if dataclasses.is_dataclass(x) else x for x in v]
        out[f.metadata.get("json_key", f.name)] = v
    return out

from dataclasses import dataclass, field
from typing import Any, Optional

@dataclass
class Usage:
    prompt_tokens: int
    completion_tokens: int

    @classmethod
    def from_dict(cls, d: dict[str, Any]) -> "Usage":
        return cls(prompt_tokens=d["prompt_tokens"], completion_tokens=d["completion_tokens"])

@dataclass
class Completion:
    id: str
    text: str
    usage: Optional[Usage] = None
    stop_reason: Optional[str] = field(default=None, metadata={"json_key": "stopReason"})

    @classmethod
    def from_dict(cls, d: dict[str, Any]) -> "Completion":
        return cls(
            id=d["id"],
            text=d["text"],
            usage=Usage.from_dict(d["usage"]) if d.get("usage") is not None else None,
            stop_reason=d.get("stopReason"),
        )

import dataclasses
import enum
import sys
import typing
from typing import Any, Optional, Type, TypeVar, Union

T = TypeVar("T")
_UNION_TYPES: tuple = (Union,)
if sys.version_info >= (3, 10):
    import types
    _UNION_TYPES = (Union, types.UnionType)

def from_data(tp: Any, data: Any) -> Any:
    origin = typing.get_origin(tp)
    args = typing.get_args(tp)
    if tp is Any or data is None:
        return data
    if origin in _UNION_TYPES:
        non_none = [a for a in args if a is not type(None)]
        if len(non_none) == 1:
            return from_data(non_none[0], data)
        for a in non_none:                       # first type that decodes without error
            try:
                return from_data(a, data)
            except (TypeError, ValueError, KeyError):
                continue
        raise TypeError(f"{data!r} matches none of {tp}")
    if origin in (list, tuple, set, frozenset):
        if origin is tuple and len(args) == 2 and args[1] is Ellipsis:
            return tuple(from_data(args[0], x) for x in data)
        if origin is tuple and args:
            return tuple(from_data(a, x) for a, x in zip(args, data))
        inner = args[0] if args else Any
        return origin(from_data(inner, x) for x in data)
    if origin is dict:
        kt, vt = args if args else (Any, Any)
        return {from_data(kt, k): from_data(vt, v) for k, v in data.items()}
    if isinstance(tp, type) and dataclasses.is_dataclass(tp):
        return decode(tp, data)
    if isinstance(tp, type) and issubclass(tp, enum.Enum):
        return tp(data)
    if tp is float and isinstance(data, int):
        return float(data)
    return data

def decode(cls: Type[T], data: dict) -> T:
    hints = typing.get_type_hints(cls)
    kwargs = {}
    for f in dataclasses.fields(cls):
        if not f.init:
            continue
        key = f.metadata.get("json_key", f.name)
        if key in data:
            kwargs[f.name] = from_data(hints[f.name], data[key])
    unknown = set(data) - {f.metadata.get("json_key", f.name) for f in dataclasses.fields(cls)}
    if unknown and getattr(cls, "__strict_keys__", False):
        raise ValueError(f"{cls.__name__}: unknown keys {sorted(unknown)}")
    return cls(**kwargs)

import enum
import json
from dataclasses import asdict, dataclass

class Role(str, enum.Enum):            # 3.11+: enum.StrEnum
    USER = "user"
    ASSISTANT = "assistant"

@dataclass
class Turn:
    role: Role
    content: str

t = Turn(Role.USER, "hi")
print(json.dumps(asdict(t)))           # {"role": "user", "content": "hi"}: str mixin serializes as its value
print(Turn(Role("assistant"), "ok"))   # decode by value

import datetime as dt
from dataclasses import dataclass, field

def utcnow() -> dt.datetime:
    return dt.datetime.now(dt.timezone.utc)

@dataclass
class RunRecord:
    run_id: str
    started_at: dt.datetime = field(default_factory=utcnow)

    def to_dict(self) -> dict:
        return {"run_id": self.run_id, "started_at": self.started_at.isoformat()}

    @classmethod
    def from_dict(cls, d: dict) -> "RunRecord":
        ts = d["started_at"].replace("Z", "+00:00")     # fromisoformat accepts "Z" only on 3.11+
        return cls(run_id=d["run_id"], started_at=dt.datetime.fromisoformat(ts))

import base64
from dataclasses import dataclass, field

import numpy as np

@dataclass
class EmbeddedChunk:
    chunk_id: str
    text: str
    embedding: np.ndarray = field(repr=False, compare=False)

    def to_dict(self, *, compact: bool = True) -> dict:
        vec = np.asarray(self.embedding, dtype=np.float32)
        emb = base64.b64encode(vec.tobytes()).decode() if compact else vec.tolist()
        return {"chunk_id": self.chunk_id, "text": self.text, "embedding": emb, "dim": int(vec.shape[0])}

    @classmethod
    def from_dict(cls, d: dict) -> "EmbeddedChunk":
        e = d["embedding"]
        vec = (np.frombuffer(base64.b64decode(e), dtype=np.float32) if isinstance(e, str)
               else np.asarray(e, dtype=np.float32))
        return cls(d["chunk_id"], d["text"], vec)

from dataclasses import dataclass, field, fields

@dataclass
class GatewayMeta:
    request_id: str = field(metadata={"json_key": "x-request-id"})
    latency_ms: float = field(metadata={"json_key": "latencyMs"})
    class_: str = field(default="standard", metadata={"json_key": "class"})

def wire_to_kwargs(cls, d: dict) -> dict:
    return {f.name: d[f.metadata.get("json_key", f.name)]
            for f in fields(cls) if f.init and f.metadata.get("json_key", f.name) in d}

meta = GatewayMeta(**wire_to_kwargs(GatewayMeta, {"x-request-id": "r1", "latencyMs": 12.5, "class": "batch"}))

import os
from dataclasses import dataclass, field, fields
from typing import Any

@dataclass(frozen=True)
class TrainConfig:
    model_name: str
    lr: float = 3e-4
    batch_size: int = 32
    epochs: int = 3
    warmup_ratio: float = 0.03
    tags: tuple = ()

    def __post_init__(self) -> None:
        if self.lr <= 0 or self.batch_size <= 0:
            raise ValueError("lr and batch_size must be positive")

    @classmethod
    def from_mapping(cls, raw: dict[str, Any]) -> "TrainConfig":
        names = {f.name for f in fields(cls)}
        unknown = set(raw) - names
        if unknown:
            raise ValueError(f"unknown config keys: {sorted(unknown)}")   # catches typos like 'learning_rate'
        conv: dict[str, Any] = {}
        for f in fields(cls):
            if f.name in raw:
                v = raw[f.name]
                conv[f.name] = tuple(v) if f.type in (tuple, "tuple") else v
        return cls(**conv)

    @classmethod
    def from_env(cls, prefix: str = "TRAIN_") -> "TrainConfig":
        casts = {int: int, float: float, str: str, "int": int, "float": float, "str": str}
        raw = {}
        for f in fields(cls):
            env = os.environ.get(prefix + f.name.upper())
            if env is not None:
                raw[f.name] = casts.get(f.type, str)(env) if f.type not in (tuple, "tuple") else tuple(env.split(","))
        return cls.from_mapping(raw)

# YAML (PyYAML): always safe_load
# import yaml
# cfg = TrainConfig.from_mapping(yaml.safe_load(open("train.yaml")))

import csv
import json
from dataclasses import asdict, dataclass, fields

@dataclass
class EvalRow:
    example_id: str
    model: str
    score: float
    latency_ms: float

rows = [EvalRow("ex1", "gpt-4o-mini", 0.8, 420.0), EvalRow("ex2", "gpt-4o-mini", 0.6, 380.0)]

with open("eval.jsonl", "w", encoding="utf-8") as fh:
    for r in rows:
        fh.write(json.dumps(asdict(r)) + "\n")

with open("eval.csv", "w", newline="", encoding="utf-8") as fh:
    w = csv.DictWriter(fh, fieldnames=[f.name for f in fields(EvalRow)])
    w.writeheader()
    w.writerows(asdict(r) for r in rows)

# import pandas as pd; df = pd.DataFrame([asdict(r) for r in rows])   # pandas also accepts dataclass lists directly

from dataclasses import dataclass

@dataclass(frozen=True)
class SamplingConfig:
    temperature: float = 0.7
    top_p: float = 1.0
    max_tokens: int = 512
    n: int = 1

    def __post_init__(self) -> None:
        errors = []
        if not 0.0 <= self.temperature <= 2.0:
            errors.append(f"temperature={self.temperature} not in [0, 2]")
        if not 0.0 < self.top_p <= 1.0:
            errors.append(f"top_p={self.top_p} not in (0, 1]")
        if self.max_tokens < 1:
            errors.append("max_tokens must be >= 1")
        if self.n > 1 and self.temperature == 0.0:
            errors.append("n > 1 with temperature 0 returns identical samples")
        if errors:
            raise ValueError("invalid SamplingConfig: " + "; ".join(errors))

import dataclasses
import typing

def check_simple_types(obj) -> None:
    """Shallow isinstance check for non-generic annotations (int, str, float, bool, classes)."""
    hints = typing.get_type_hints(type(obj))
    for f in dataclasses.fields(obj):
        tp = hints[f.name]
        if isinstance(tp, type) and not isinstance(getattr(obj, f.name), tp):
            if tp is float and isinstance(getattr(obj, f.name), int):
                continue
            raise TypeError(f"{type(obj).__name__}.{f.name}: expected {tp.__name__}, "
                            f"got {type(getattr(obj, f.name)).__name__}")

import json
from dataclasses import asdict

def test_completion_round_trip():
    raw = {"id": "c1", "text": "hi", "usage": {"prompt_tokens": 3, "completion_tokens": 1}, "stopReason": "end"}
    c = Completion.from_dict(raw)
    assert c.usage is not None and c.usage.prompt_tokens == 3
    again = Completion.from_dict(json.loads(json.dumps(export(c))))
    assert again == c
    assert asdict(c)["stop_reason"] == "end"     # asdict uses Python names, not wire names