"""Shared wire schema. Semantic storage validation belongs to the worker.""" import json import math STRING_FIELDS = ("raw_id", "device_id", "caller_num", "svc", "lac", "ci") INTEGER_FIELDS = ("rat", "epoch_start", "epoch_end", "up_bytes", "down_bytes") REQUIRED_FIELDS = STRING_FIELDS + INTEGER_FIELDS class PayloadError(ValueError): def __init__(self, reason, detail, field=None): super().__init__(detail) self.reason = reason self.field = field def parse_payload(data): def reject_constant(value): raise ValueError(f"Non-finite number: {value}") def finite_float(value): number = float(value) if not math.isfinite(number): raise ValueError(f"Non-finite number: {value}") return number try: payload = json.loads(data.decode("utf-8"), parse_constant=reject_constant, parse_float=finite_float) except (UnicodeError, ValueError, RecursionError) as exc: raise PayloadError("invalid_json", str(exc)) from exc if not isinstance(payload, dict): raise PayloadError("invalid_payload", "Expected a JSON object") return payload def validate_structure(payload): for field in REQUIRED_FIELDS: if payload.get(field) is None: raise PayloadError("missing_required_field", f"Field {field} is required", field) for field in STRING_FIELDS: if not isinstance(payload[field], str) or not payload[field].strip(): raise PayloadError("invalid_payload", f"Field {field} must be a nonempty string", field) try: payload[field].encode("utf-8") except UnicodeError as exc: raise PayloadError("invalid_payload", f"Field {field} contains invalid Unicode", field) from exc for field in INTEGER_FIELDS: if type(payload[field]) is not int: raise PayloadError("invalid_payload", f"Field {field} must be an integer", field) if payload.get("src_ip") is not None and not isinstance(payload["src_ip"], str): raise PayloadError("invalid_payload", "Field src_ip must be a string or null", "src_ip")