···
+
AT Protocol Lexicon Type Models
+
Combined implementation of all AT Protocol Lexicon data types including:
+
- Primitive types (boolean, integer, string, null)
+
- Complex types (array, object, params)
+
- Reference types (ref, union, token)
+
- Special types (record, query, procedure, subscription)
+
- Binary types (bytes, CID links)
+
from datetime import datetime
+
from pydantic import field_validator, field_serializer
+
from cid.cid import CIDv1, make_cid
+
from .base import DataModel
+
from .exceptions import ValidationError, SerializationError, InvalidCIDError
+
class BooleanModel(DataModel):
+
"""Model for AT Protocol boolean type."""
+
default: bool | None = None
+
"""Default value if not provided"""
+
const: bool | None = None
+
"""Fixed constant value if specified"""
+
def __init__(self, **data: Any) -> None:
+
Initialize boolean model with validation.
+
**data: Input data containing boolean value
+
ValueError: If value doesn't match const or is not boolean
+
super().__init__(**data)
+
if self.const is not None and self.value != self.const:
+
raise ValueError(f"Boolean value must be {self.const}")
+
@field_validator("value", mode="before")
+
def validateBoolean(cls, v: Any) -> bool:
+
Validate and convert input to boolean.
+
Validated boolean value
+
ValueError: If value cannot be converted to boolean
+
if isinstance(v, bool):
+
if v.lower() in ("true", "1"):
+
if v.lower() in ("false", "0"):
+
raise ValueError("Value must be a boolean")
+
class IntegerModel(DataModel):
+
"""Model for AT Protocol integer type."""
+
minimum: int | None = None
+
"""Minimum acceptable value"""
+
maximum: int | None = None
+
"""Maximum acceptable value"""
+
enum: list[int] | None = None
+
"""Closed set of allowed values"""
+
default: int | None = None
+
"""Default value if not provided"""
+
const: int | None = None
+
"""Fixed constant value if specified"""
+
def __init__(self, **data: Any) -> None:
+
Initialize integer model with validation.
+
**data: Input data containing integer value
+
ValueError: If value violates constraints
+
super().__init__(**data)
+
if self.const is not None and self.value != self.const:
+
raise ValueError(f"Integer value must be {self.const}")
+
@field_validator("value", mode="before")
+
def validateInteger(cls, v: Any) -> int:
+
Validate and convert input to integer.
+
Validated integer value
+
ValueError: If value violates constraints
+
if not isinstance(v, int):
+
except (TypeError, ValueError):
+
raise ValueError("Value must be an integer")
+
if cls.enum and v not in cls.enum:
+
raise ValueError(f"Value must be one of {cls.enum}")
+
if cls.minimum is not None and v < cls.minimum:
+
raise ValueError(f"Value must be >= {cls.minimum}")
+
if cls.maximum is not None and v > cls.maximum:
+
raise ValueError(f"Value must be <= {cls.maximum}")
+
# String Types (from string.py)
+
class StringModel(DataModel):
+
"""Model for AT Protocol string type."""
+
format: str | None = None
+
"""String format restriction"""
+
maxLength: int | None = None
+
"""Maximum length in UTF-8 bytes"""
+
minLength: int | None = None
+
"""Minimum length in UTF-8 bytes"""
+
knownValues: list[str] | None = None
+
"""Suggested/common values"""
+
enum: list[str] | None = None
+
"""Closed set of allowed values"""
+
default: str | None = None
+
"""Default value if not provided"""
+
const: str | None = None
+
"""Fixed constant value if specified"""
+
def __init__(self, **data: Any) -> None:
+
Initialize string model with validation.
+
**data: Input data containing string value
+
ValueError: If value violates constraints
+
super().__init__(**data)
+
if self.const is not None and self.value != self.const:
+
raise ValueError(f"String value must be {self.const}")
+
@field_validator("value", mode="before")
+
def validateString(cls, v: Any) -> str:
+
Validate and convert input to string.
+
ValueError: If value violates constraints
+
if not isinstance(v, str):
+
if cls.minLength is not None and len(v.encode()) < cls.minLength:
+
raise ValueError(f"String must be at least {cls.minLength} bytes")
+
if cls.maxLength is not None and len(v.encode()) > cls.maxLength:
+
raise ValueError(f"String must be at most {cls.maxLength} bytes")
+
if cls.enum and v not in cls.enum:
+
raise ValueError(f"Value must be one of {cls.enum}")
+
def _validateFormat(cls, v: str) -> None:
+
"""Validate string format based on specified format type."""
+
if cls.format == "datetime":
+
cls._validateDatetime(v)
+
elif cls.format == "uri":
+
elif cls.format == "did":
+
elif cls.format == "handle":
+
elif cls.format == "at-identifier":
+
cls._validateAtIdentifier(v)
+
elif cls.format == "at-uri":
+
elif cls.format == "cid":
+
elif cls.format == "nsid":
+
elif cls.format == "tid":
+
elif cls.format == "record-key":
+
cls._validateRecordKey(v)
+
elif cls.format == "language":
+
cls._validateLanguage(v)
+
def _validateDid(cls, v: str) -> None:
+
"""Validate DID format"""
+
raise ValueError("DID too long")
+
if not re.match(r"^did:[a-z]+:[a-zA-Z0-9._:%-]*[a-zA-Z0-9._-]$", v):
+
raise ValueError("Invalid DID format")
+
def _validateHandle(cls, v: str) -> None:
+
"""Validate handle format"""
+
r"^([a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?$",
+
raise ValueError("Handle contains invalid characters")
+
raise ValueError("Handle too long")
+
def _validateAtIdentifier(cls, v: str) -> None:
+
"""Validate at-identifier format (DID or handle)"""
+
if v.startswith("did:"):
+
except ValueError as e:
+
raise ValueError(f"Invalid at-identifier: {e}")
+
def _validateAtUri(cls, v: str) -> None:
+
"""Validate AT-URI format"""
+
if not v.startswith("at://"):
+
raise ValueError("AT-URI must start with 'at://'")
+
raise ValueError("AT-URI too long")
+
raise ValueError("AT-URI cannot have trailing slash")
+
parts = v[5:].split("/")
+
raise ValueError("AT-URI must have authority")
+
if authority.startswith("did:"):
+
if len(authority) > 2048:
+
raise ValueError("DID too long")
+
if ":" not in authority[4:]:
+
raise ValueError("Invalid DID format")
+
if not re.match(r"^[a-z0-9.-]+$", authority):
+
raise ValueError("Invalid handle characters")
+
if len(authority) > 253:
+
raise ValueError("Handle too long")
+
raise ValueError("AT-URI path too deep")
+
if not re.match(r"^[a-zA-Z0-9.-]+$", collection):
+
raise ValueError("Invalid collection NSID")
+
raise ValueError("Record key cannot be empty")
+
if not re.match(r"^[a-zA-Z0-9._:%-~]+$", rkey):
+
raise ValueError("Invalid record key characters")
+
def _validateCid(cls, v: str) -> None:
+
"""Validate CID string format"""
+
raise ValueError("CID too long")
+
if not re.match(r"^[a-zA-Z]+$", v):
+
raise ValueError("CID contains invalid characters")
+
def _validateNsid(cls, v: str) -> None:
+
"""Validate NSID format"""
+
raise ValueError("NSID too long")
+
r"^[a-zA-Z]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(\.[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)+(\.[a-zA-Z]([a-zA-Z0-9]{0,62})?)$",
+
raise ValueError("NSID contains invalid characters")
+
def _validateTid(cls, v: str) -> None:
+
"""Validate TID format"""
+
raise ValueError("TID too long")
+
r"^[234567abcdefghij][234567abcdefghijklmnopqrstuvwxyz]{12}$", v
+
raise ValueError("TID contains invalid characters")
+
def _validateRecordKey(cls, v: str) -> None:
+
"""Validate record-key format"""
+
raise ValueError("Record key too long")
+
if v == "." or v == "..":
+
raise ValueError(f"Record key is {v}, which is not allowed")
+
if not re.match(r"^[a-zA-Z0-9._:%-~]+$", v):
+
raise ValueError("Record key contains invalid characters")
+
def _validateLanguage(cls, v: str) -> None:
+
"""Validate BCP 47 language tag"""
+
if not re.match(r"^[a-zA-Z]{1,8}(-[a-zA-Z0-9]{1,8})*$", v):
+
raise ValueError("Invalid language tag format")
+
cls._validateDatetime(v)
+
elif cls.format == "uri":
+
elif cls.format == "did":
+
elif cls.format == "handle":
+
elif cls.format == "at-identifier":
+
cls._validateAtIdentifier(v)
+
elif cls.format == "at-uri":
+
elif cls.format == "cid":
+
elif cls.format == "nsid":
+
elif cls.format == "tid":
+
elif cls.format == "record-key":
+
cls._validateRecordKey(v)
+
elif cls.format == "language":
+
cls._validateLanguage(v)
+
def _validateDatetime(cls, v: str) -> None:
+
"""Validate RFC 3339 datetime format"""
+
datetime.fromisoformat(v.replace("Z", "+00:00"))
+
raise ValueError("Invalid datetime format")
+
def _validateUri(cls, v: str) -> None:
+
"""Validate URI format"""
+
raise ValueError("URI too long")
+
if not re.match(r"^[a-zA-Z][a-zA-Z0-9+.-]*:.+", v):
+
raise ValueError("Invalid URI format")
+
# ... (other validation methods remain the same)
+
# Binary Types (from binary.py)
+
class BytesModel(DataModel):
+
"""Model for AT Protocol bytes type."""
+
minLength: int | None = None
+
"""Minimum size in bytes"""
+
maxLength: int | None = None
+
"""Maximum size in bytes"""
+
def __init__(self, **data: Any) -> None:
+
Initialize bytes model with validation.
+
**data: Input data containing bytes value and constraints
+
ValidationError: If length constraints are violated
+
super().__init__(**data)
+
@field_validator("value")
+
def validateLength(cls, v: bytes, info: Any) -> bytes:
+
Validate bytes length against constraints.
+
v: Bytes value to validate
+
info: Validation info containing field values
+
ValidationError: If length constraints are violated
+
minLen = info.data.get("minLength")
+
maxLen = info.data.get("maxLength")
+
if minLen is not None and len(v) < minLen:
+
message=f"Bytes length {len(v)} is less than minimum {minLen}",
+
if maxLen is not None and len(v) > maxLen:
+
field="value", message=f"Bytes length {len(v)} exceeds maximum {maxLen}"
+
@field_serializer("value")
+
def serializeBytes(self, v: bytes) -> dict[str, str]:
+
Serialize bytes to JSON format with base64 encoding.
+
Dictionary with base64 encoded bytes
+
SerializationError: If encoding fails
+
return {"$bytes": base64.b64encode(v).decode()}
+
raise SerializationError("value", f"Failed to encode bytes: {e}")
+
class CidLinkModel(DataModel):
+
"""Model for AT Protocol CID link type."""
+
"""CID reference to linked content"""
+
def __init__(self, **data: Any) -> None:
+
Initialize CID link model with validation.
+
**data: Input data containing CID link
+
InvalidCIDError: If CID is invalid
+
if isinstance(data.get("link"), str):
+
data["link"] = make_cid(data["link"])
+
except ValueError as e:
+
raise InvalidCIDError(f"Invalid CID: {e}")
+
super().__init__(**data)
+
@field_serializer("link")
+
def serializeCid(self, v: CIDv1) -> dict[str, str]:
+
Serialize CID to JSON format.
+
Dictionary with string CID representation
+
return {"$link": str(v)}
+
# Complex Types (from complex.py)
+
class ArrayModel(DataModel):
+
"""Model for AT Protocol array type."""
+
"""Schema definition for array elements"""
+
minLength: int | None = None
+
"""Minimum number of elements"""
+
maxLength: int | None = None
+
"""Maximum number of elements"""
+
def __init__(self, **data: Any) -> None:
+
Initialize array model with validation.
+
**data: Input data containing array values
+
ValueError: If array violates constraints
+
super().__init__(**data)
+
@field_validator("value", mode="before")
+
def validateArray(cls, v: Any) -> list[Any]:
+
Validate array structure and elements.
+
ValueError: If array violates constraints
+
if not isinstance(v, list):
+
raise ValueError("Value must be an array")
+
if cls.minLength is not None and len(v) < cls.minLength:
+
raise ValueError(f"Array must have at least {cls.minLength} items")
+
if cls.maxLength is not None and len(v) > cls.maxLength:
+
raise ValueError(f"Array must have at most {cls.maxLength} items")
+
class ObjectModel(DataModel):
+
"""Model for AT Protocol object type."""
+
properties: dict[str, Any]
+
"""Map of property names to their schema definitions"""
+
required: list[str] | None = None
+
"""List of required property names"""
+
nullable: list[str] | None = None
+
"""List of properties that can be null"""
+
"""Object property values"""
+
def __init__(self, **data: Any) -> None:
+
Initialize object model with validation.
+
**data: Input data containing object properties
+
ValueError: If object violates constraints
+
super().__init__(**data)
+
@field_validator("value", mode="before")
+
def validateObject(cls, v: Any) -> dict[str, Any]:
+
Validate object structure and properties.
+
ValueError: If object violates constraints
+
if not isinstance(v, dict):
+
raise ValueError("Value must be an object")
+
for field in cls.required:
+
raise ValueError(f"Missing required field: {field}")
+
for field, value in v.items():
+
if field not in cls.nullable and value is None:
+
raise ValueError(f"Field {field} cannot be null")
+
class ParamsModel(DataModel):
+
"""Model for AT Protocol params type."""
+
required: list[str] | None = None
+
"""List of required parameter names"""
+
properties: dict[str, Any]
+
"""Map of parameter names to their schema definitions"""
+
def __init__(self, **data: Any) -> None:
+
Initialize params model with validation.
+
**data: Input data containing parameter values
+
ValueError: If parameters violate constraints
+
super().__init__(**data)
+
@field_validator("value", mode="before")
+
def validateParams(cls, v: Any) -> dict[str, Any]:
+
Validate parameters structure and values.
+
Validated parameters dictionary
+
ValueError: If parameters violate constraints
+
if not isinstance(v, dict):
+
raise ValueError("Value must be a dictionary of parameters")
+
for param in cls.required:
+
if param not in validated:
+
raise ValueError(f"Missing required parameter: {param}")
+
for param, value in validated.items():
+
if param in cls.properties:
+
propType = cls.properties[param].get("type")
+
if propType == "boolean" and not isinstance(value, bool):
+
raise ValueError(f"Parameter {param} must be boolean")
+
elif propType == "integer" and not isinstance(value, int):
+
raise ValueError(f"Parameter {param} must be integer")
+
elif propType == "string" and not isinstance(value, str):
+
raise ValueError(f"Parameter {param} must be string")
+
elif propType == "array":
+
if not isinstance(value, list):
+
raise ValueError(f"Parameter {param} must be array")
+
if "items" in cls.properties[param]:
+
itemType = cls.properties[param]["items"].get("type")
+
if itemType == "boolean" and not isinstance(item, bool):
+
f"Array item in {param} must be boolean"
+
elif itemType == "integer" and not isinstance(item, int):
+
f"Array item in {param} must be integer"
+
elif itemType == "string" and not isinstance(item, str):
+
f"Array item in {param} must be string"
+
elif itemType == "unknown" and not isinstance(item, dict):
+
f"Array item in {param} must be object"
+
elif propType == "unknown" and not isinstance(value, dict):
+
raise ValueError(f"Parameter {param} must be object")
dwn.poxiao-labs.work