Compare changes

Choose any two refs to compare.

-96
src/atpasser/handle/__init__.py
···
-
import dns.resolver, requests
-
-
from atpasser import did
-
-
-
class Handle:
-
"""
-
A class representing a Handle.
-
-
-
Attributes:
-
handle (str): The Handle.
-
"""
-
-
def __init__(self, handle: str) -> None:
-
"""
-
Initalizes an Handle object.
-
-
Parameters:
-
handle (str): The Handle.
-
"""
-
-
if len(handle) > 253:
-
raise ValueError("handle is more than 253 chars")
-
-
labels = handle.lower().split(".")
-
-
if len(labels) < 2:
-
raise ValueError("are you tld?")
-
-
if labels[0] == "" or labels[-1] == "":
-
raise ValueError("proceeding or tariling ascii periods")
-
-
for label in labels:
-
if len(label) not in range(1, 64):
-
raise ValueError("two periods or segment longer than 63 char")
-
charset = set(label)
-
validset = set("abcdefghijklmnopqrstuvwxyz0123456789-")
-
if not charset.issubset(validset):
-
raise ValueError("invalid char used in segment")
-
if label.startswith("-") or label.endswith("-"):
-
raise ValueError("segments starts or ends with hyphen")
-
-
tld = labels[-1]
-
if tld[0] in "0123456789":
-
raise ValueError("tld starts with digit")
-
-
self.handle = handle
-
-
def __str__(self) -> str:
-
"""
-
-
Convert the TID to a string by given the URI.
-
"""
-
return self.handle
-
-
def __eq__(self, value: object, /) -> bool:
-
"""
-
-
Check if the 2 values are exactly the same.
-
"""
-
-
if isinstance(value, Handle):
-
-
return str(self) == str(value)
-
else:
-
-
return False
-
-
def toTID(self):
-
"""
-
Convert the handle to TID.
-
-
Returns:
-
An DID object, or `None` if the handle is invalid.
-
"""
-
try:
-
answers = dns.resolver.resolve("_atproto." + self.handle, "TXT")
-
except:
-
answers = []
-
for answer in answers:
-
if str(answer).startswith('"did='):
-
try:
-
uri = str(answer)[5:-1]
-
return did.DID(uri)
-
except:
-
pass # cannot resolve via dns
-
response = requests.get(f"https://{self.handle}/.well-known/atproto-did")
-
if response.status_code // 100 != 2:
-
return None
-
if response.headers.get("Content-Type") != "text/plain":
-
pass # Pass for now, because some sites like neocities, breaks this rule
-
try:
-
return did.DID(response.text)
-
except:
-
return None
+61 -17
src/atpasser/uri/handle.py
···
"""
if len(handle) > 253:
-
raise InvalidHandleError(handle, "exceeds maximum length", f"Handle length {len(handle)} exceeds maximum allowed length of 253 characters")
+
raise InvalidHandleError(
+
handle,
+
"exceeds maximum length",
+
f"Handle length {len(handle)} exceeds maximum allowed length of 253 characters",
+
)
labels = handle.lower().split(".")
if len(labels) < 2:
-
raise InvalidHandleError(handle, "invalid format", "Handle must contain at least one dot separator, e.g., 'example.com'")
+
raise InvalidHandleError(
+
handle,
+
"invalid format",
+
"Handle must contain at least one dot separator, e.g., 'example.com'",
+
)
if labels[0] == "" or labels[-1] == "":
-
raise InvalidHandleError(handle, "invalid format", "Handle cannot start or end with a dot")
+
raise InvalidHandleError(
+
handle, "invalid format", "Handle cannot start or end with a dot"
+
)
for i, label in enumerate(labels):
if len(label) not in range(1, 64):
-
raise InvalidHandleError(handle, "segment length error", f"Handle segment {i+1} length {len(label)} is not in the 1-63 character range")
+
raise InvalidHandleError(
+
handle,
+
"segment length error",
+
f"Handle segment {i+1} length {len(label)} is not in the 1-63 character range",
+
)
charset = set(label)
validset = set("abcdefghijklmnopqrstuvwxyz0123456789-")
if not charset.issubset(validset):
invalid_chars = charset - validset
-
raise InvalidHandleError(handle, "contains invalid characters", f"Handle segment {i+1} contains invalid characters: {', '.join(invalid_chars)}")
+
raise InvalidHandleError(
+
handle,
+
"contains invalid characters",
+
f"Handle segment {i+1} contains invalid characters: {', '.join(invalid_chars)}",
+
)
if label.startswith("-") or label.endswith("-"):
-
raise InvalidHandleError(handle, "invalid format", f"Handle segment {i+1} cannot start or end with a hyphen")
+
raise InvalidHandleError(
+
handle,
+
"invalid format",
+
f"Handle segment {i+1} cannot start or end with a hyphen",
+
)
tld = labels[-1]
if tld[0] in "0123456789":
-
raise InvalidHandleError(handle, "invalid format", "Handle's top-level domain cannot start with a digit")
+
raise InvalidHandleError(
+
handle,
+
"invalid format",
+
"Handle's top-level domain cannot start with a digit",
+
)
self.handle = handle
···
answers = dns.resolver.resolve("_atproto." + self.handle, "TXT")
except Exception as e:
answers = []
-
+
for answer in answers:
answer_str = str(answer)
if answer_str.startswith('"did='):
···
# Continue trying other records or methods
continue
except Exception as e:
-
raise ResolutionError(self.handle, "DNS resolution", f"Error parsing DNS TXT record: {str(e)}")
-
+
raise ResolutionError(
+
self.handle,
+
"DNS resolution",
+
f"Error parsing DNS TXT record: {str(e)}",
+
)
+
# If DNS resolution fails, try HTTP method
try:
-
response = requests.get(f"https://{self.handle}/.well-known/atproto-did")
+
response = requests.get(
+
f"https://{self.handle}/.well-known/atproto-did"
+
)
if response.status_code // 100 != 2:
return None
-
+
# Some websites may return incorrect Content-Type, so here we only warn without throwing an exception
content_type = response.headers.get("Content-Type")
if content_type != "text/plain" and content_type:
# Log warning but don't block processing
pass
-
+
try:
return DID(response.text.strip())
except InvalidDIDError:
return None
except requests.RequestException as e:
-
raise ResolutionError(self.handle, "HTTP request", f"Error requesting well-known endpoint: {str(e)}")
+
raise ResolutionError(
+
self.handle,
+
"HTTP request",
+
f"Error requesting well-known endpoint: {str(e)}",
+
)
except Exception as e:
-
raise ResolutionError(self.handle, "HTTP parsing", f"Error parsing HTTP response: {str(e)}")
-
+
raise ResolutionError(
+
self.handle,
+
"HTTP parsing",
+
f"Error parsing HTTP response: {str(e)}",
+
)
+
except Exception as e:
if isinstance(e, ResolutionError):
raise
-
raise ResolutionError(self.handle, "resolution", f"Unknown error occurred while resolving Handle: {str(e)}")
+
raise ResolutionError(
+
self.handle,
+
"resolution",
+
f"Unknown error occurred while resolving Handle: {str(e)}",
+
)
+82 -20
src/atpasser/uri/nsid.py
···
if "#" in nsid:
parts = nsid.split("#", 1)
if len(parts) != 2:
-
raise InvalidNSIDError(nsid, "invalid format", "NSID fragment format is incorrect")
+
raise InvalidNSIDError(
+
nsid, "invalid format", "NSID fragment format is incorrect"
+
)
nsidWithoutFragment, fragment = parts
else:
nsidWithoutFragment, fragment = nsid, None
···
if not set([x for x in nsidWithoutFragment]).issubset(
set([chr(i) for i in range(0x80)])
):
-
raise InvalidNSIDError(nsid, "contains invalid characters", "NSID must only contain ASCII characters")
+
raise InvalidNSIDError(
+
nsid,
+
"contains invalid characters",
+
"NSID must only contain ASCII characters",
+
)
# Check length
if len(nsidWithoutFragment) > 317:
-
raise InvalidNSIDError(nsid, "exceeds maximum length", f"NSID length {len(nsidWithoutFragment)} exceeds maximum allowed length of 317 characters")
+
raise InvalidNSIDError(
+
nsid,
+
"exceeds maximum length",
+
f"NSID length {len(nsidWithoutFragment)} exceeds maximum allowed length of 317 characters",
+
)
segments = nsidWithoutFragment.split(".")
# Check for leading or trailing dots
if nsidWithoutFragment.startswith(".") or nsidWithoutFragment.endswith("."):
-
raise InvalidNSIDError(nsid, "invalid format", "NSID cannot start or end with a dot")
+
raise InvalidNSIDError(
+
nsid, "invalid format", "NSID cannot start or end with a dot"
+
)
# Check segment count
if len(segments) < 3:
-
raise InvalidNSIDError(nsid, "invalid format", f"NSID must contain at least 3 segments, currently has {len(segments)}")
+
raise InvalidNSIDError(
+
nsid,
+
"invalid format",
+
f"NSID must contain at least 3 segments, currently has {len(segments)}",
+
)
domainAuthority = [segment.lower() for segment in segments[0:-1]]
# Check domain authority length
if len(".".join(domainAuthority)) > 253:
-
raise InvalidNSIDError(nsid, "domain authority length exceeds limit", "Domain authority part length exceeds 253 characters")
+
raise InvalidNSIDError(
+
nsid,
+
"domain authority length exceeds limit",
+
"Domain authority part length exceeds 253 characters",
+
)
# Check each domain segment
for i, segment in enumerate(domainAuthority):
if len(segment) > 63 or segment == "":
-
raise InvalidNSIDError(nsid, "segment length error", f"Domain authority segment {i+1} length is not in the 1-63 character range")
+
raise InvalidNSIDError(
+
nsid,
+
"segment length error",
+
f"Domain authority segment {i+1} length is not in the 1-63 character range",
+
)
if not set(segment).issubset(set("abcdefghijklmnopqrstuvwxyz0123456789-")):
-
invalid_chars = set(segment) - set("abcdefghijklmnopqrstuvwxyz0123456789-")
-
raise InvalidNSIDError(nsid, "contains invalid characters", f"Domain authority segment {i+1} contains invalid characters: {', '.join(invalid_chars)}")
+
invalid_chars = set(segment) - set(
+
"abcdefghijklmnopqrstuvwxyz0123456789-"
+
)
+
raise InvalidNSIDError(
+
nsid,
+
"contains invalid characters",
+
f"Domain authority segment {i+1} contains invalid characters: {', '.join(invalid_chars)}",
+
)
if segment.startswith("-") or segment.endswith("-"):
-
raise InvalidNSIDError(nsid, "invalid format", f"Domain authority segment {i+1} cannot start or end with a hyphen")
-
+
raise InvalidNSIDError(
+
nsid,
+
"invalid format",
+
f"Domain authority segment {i+1} cannot start or end with a hyphen",
+
)
+
# Check if top-level domain starts with a digit
if segments[0][0] in "0123456789":
-
raise InvalidNSIDError(nsid, "invalid format", "NSID's top-level domain cannot start with a digit")
+
raise InvalidNSIDError(
+
nsid,
+
"invalid format",
+
"NSID's top-level domain cannot start with a digit",
+
)
self.domainAuthority = domainAuthority
self.domainAuthorityAsText = ".".join(domainAuthority)
···
# Check name
if name == "" or len(name) > 63:
-
raise InvalidNSIDError(nsid, "name length error", "NSID name cannot be empty and length cannot exceed 63 characters")
+
raise InvalidNSIDError(
+
nsid,
+
"name length error",
+
"NSID name cannot be empty and length cannot exceed 63 characters",
+
)
if not set(name).issubset(
set("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789")
):
-
invalid_chars = set(name) - set("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789")
-
raise InvalidNSIDError(nsid, "contains invalid characters", f"NSID name contains invalid characters: {', '.join(invalid_chars)}")
+
invalid_chars = set(name) - set(
+
"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"
+
)
+
raise InvalidNSIDError(
+
nsid,
+
"contains invalid characters",
+
f"NSID name contains invalid characters: {', '.join(invalid_chars)}",
+
)
if name[0] in "0123456789":
-
raise InvalidNSIDError(nsid, "invalid format", "NSID name cannot start with a digit")
+
raise InvalidNSIDError(
+
nsid, "invalid format", "NSID name cannot start with a digit"
+
)
self.name = name
# Check fragment
if fragment != None:
if fragment == "" or len(fragment) > 63:
-
raise InvalidNSIDError(nsid, "fragment length error", "NSID fragment cannot be empty and length cannot exceed 63 characters")
+
raise InvalidNSIDError(
+
nsid,
+
"fragment length error",
+
"NSID fragment cannot be empty and length cannot exceed 63 characters",
+
)
if not set(fragment).issubset(
set("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789")
):
-
invalid_chars = set(fragment) - set("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789")
-
raise InvalidNSIDError(nsid, "contains invalid characters", f"NSID fragment contains invalid characters: {', '.join(invalid_chars)}")
+
invalid_chars = set(fragment) - set(
+
"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"
+
)
+
raise InvalidNSIDError(
+
nsid,
+
"contains invalid characters",
+
f"NSID fragment contains invalid characters: {', '.join(invalid_chars)}",
+
)
if fragment[0] in "0123456789":
-
raise InvalidNSIDError(nsid, "invalid format", "NSID fragment cannot start with a digit")
+
raise InvalidNSIDError(
+
nsid, "invalid format", "NSID fragment cannot start with a digit"
+
)
self.fragment = fragment
-136
src/atpasser/uri/tid.py
···
-
import datetime, random
-
-
-
class TID:
-
"""A class representing a TID (Time-based Identifier) in the AT Protocol.
-
-
TIDs are time-based identifiers used for ordering and uniquely identifying
-
records in the AT Protocol. They combine a microsecond-precision timestamp
-
with a clock identifier to ensure uniqueness even when multiple records
-
are created in the same microsecond. TIDs are sortable and provide both
-
temporal ordering and uniqueness guarantees.
-
-
Attributes:
-
timestamp (datetime.datetime): The timestamp component of the TID.
-
clockIdentifier (int): Clock identifier (0-1023) for disambiguation.
-
"""
-
-
def __init__(
-
self, time: datetime.datetime | None = None, clockIdentifier: int | None = None
-
) -> None:
-
"""Initializes a TID object with timestamp and clock identifier.
-
-
Creates a new TID with the specified timestamp and clock identifier.
-
If no timestamp is provided, uses the current time. If no clock identifier
-
is provided, generates a random value between 0-1023.
-
-
Args:
-
time (datetime.datetime, optional): The timestamp for the TID.
-
Defaults to current time if not provided.
-
clockIdentifier (int, optional): Clock identifier (0-1023) for
-
disambiguation. Defaults to random value if not provided.
-
"""
-
if time == None:
-
self.timestamp = datetime.datetime.now()
-
else:
-
self.timestamp = time
-
if clockIdentifier == None:
-
self.clockIdentifier = random.randrange(0, 1024)
-
else:
-
self.clockIdentifier = clockIdentifier
-
-
def __int__(self):
-
"""Convert the TID to its integer representation.
-
-
Combines the timestamp (in microseconds) and clock identifier into
-
a single 64-bit integer where the high bits represent the timestamp
-
and the low 10 bits represent the clock identifier.
-
-
Returns:
-
int: The integer representation of the TID.
-
"""
-
timestamp = int(self.timestamp.timestamp() * 1000000)
-
return timestamp * 1024 + self.clockIdentifier
-
-
def __str__(self):
-
"""Convert the TID to a base32-sortable string representation.
-
-
Encodes the TID as a base32 string using a custom character set that
-
maintains lexicographical sort order corresponding to temporal order.
-
This format is commonly used in the AT Protocol for compact,
-
sortable identifiers.
-
-
Returns:
-
str: The base32 string representation of the TID.
-
"""
-
integer = int(self)
-
binary = f"{integer:065b}"
-
return "".join(
-
[
-
"234567abcdefghijklmnopqrstuvwxyz"[int(binary[i : i + 5], base=2)]
-
for i in range(0, len(binary), 5)
-
]
-
)
-
-
def __eq__(self, value: object, /) -> bool:
-
"""Check if two TID objects represent the same identifier.
-
-
Args:
-
value (object): The object to compare with.
-
-
Returns:
-
bool: True if the objects represent the same TID, False otherwise.
-
"""
-
if isinstance(value, TID):
-
return int(self) == int(value)
-
else:
-
return False
-
-
-
def importTIDfromInteger(value: int | None = None):
-
"""Create a TID object from an integer representation.
-
-
Converts a 64-bit integer back into a TID object by extracting the
-
timestamp and clock identifier components. If no value is provided,
-
creates a TID for the current time.
-
-
Args:
-
value (int, optional): The integer value to convert to a TID.
-
Defaults to creating a TID for the current time if not provided.
-
-
Returns:
-
TID: The TID object created from the integer value.
-
"""
-
if value == None:
-
value = int(TID())
-
clockIdentifier = value % 1024
-
timestamp = (value >> 10) / 1000000
-
return TID(datetime.datetime.fromtimestamp(timestamp), clockIdentifier)
-
-
-
def importTIDfromBase32(value: str | None = None):
-
"""Create a TID object from a base32 string representation.
-
-
Converts a base32-encoded TID string back into a TID object by decoding
-
the string to its integer representation and then extracting the timestamp
-
and clock identifier components. If no value is provided, creates a TID
-
for the current time.
-
-
Args:
-
value (str, optional): The base32 string to convert to a TID.
-
Defaults to creating a TID for the current time if not provided.
-
-
Returns:
-
TID: The TID object created from the base32 string.
-
"""
-
if value == None:
-
value = str(TID())
-
b32s = "234567abcdefghijklmnopqrstuvwxyz"
-
return importTIDfromInteger(
-
sum(
-
[
-
b32s.find(value[i]) * (32 ** (len(value) - i - 1))
-
for i in range(len(value))
-
]
-
)
-
)
+3
src/atpasser/uri/__init__.py
···
"JSONPath parsing failed",
f"Failed to parse JSONPath fragment '{fragment}': {str(e)}",
)
+
else:
+
self.fragment = None
+
self.fragmentAsText = None
if query != None:
try:
-182
src/atpasser/data/decoder.py
···
-
"""
-
JSON decoder for ATProto data model.
-
-
This module provides a JSON decoder that handles ATProto-specific data types,
-
including bytes, CID links, and typed objects.
-
"""
-
-
import base64
-
import json
-
from typing import Any, Callable, Dict, Optional
-
from cid import CIDv0, CIDv1, make_cid
-
-
-
class JsonDecoder(json.JSONDecoder):
-
"""A JSON decoder that supports ATProto data types.
-
-
This decoder extends the standard JSON decoder to handle ATProto-specific
-
data types, including bytes, CID links, and typed objects.
-
-
Attributes:
-
type_hook_registry: Registry for type-specific hooks.
-
encoding: The encoding to use for string deserialization.
-
"""
-
-
def __init__(
-
self,
-
*,
-
object_hook: Optional[Callable[[Dict[str, Any]], Any]] = None,
-
type_hook_registry: Optional[Any] = None,
-
type_processor_registry: Optional[Any] = None,
-
encoding: str = "utf-8",
-
**kwargs: Any,
-
) -> None:
-
"""Initialize the JSON decoder.
-
-
Args:
-
object_hook: Optional function to call with each decoded object.
-
type_hook_registry: Registry for type-specific hooks.
-
type_processor_registry: Registry for type-specific processors.
-
encoding: The encoding to use for string deserialization.
-
**kwargs: Additional keyword arguments to pass to the parent class.
-
"""
-
# Use the type processor registry if provided, otherwise use the type hook registry
-
if type_processor_registry is not None:
-
type_hook_registry = type_processor_registry.to_hook_registry()
-
elif type_hook_registry is None:
-
from .hooks import get_global_registry
-
-
type_hook_registry = get_global_registry()
-
-
# Create a combined object hook that calls both the custom hook and our hook
-
combined_hook = self._create_combined_hook(object_hook, type_hook_registry)
-
-
super().__init__(object_hook=combined_hook, **kwargs)
-
self.type_hook_registry = type_hook_registry
-
self.type_processor_registry = type_processor_registry
-
self.encoding = encoding
-
-
def _create_combined_hook(
-
self,
-
custom_hook: Optional[Callable[[Dict[str, Any]], Any]],
-
type_hook_registry: Optional[Any],
-
) -> Callable[[Dict[str, Any]], Any]:
-
"""Create a combined object hook function.
-
-
Args:
-
custom_hook: Optional custom object hook function.
-
type_hook_registry: Registry for type-specific hooks.
-
-
Returns:
-
A combined object hook function.
-
"""
-
-
def combined_hook(obj: Dict[str, Any]) -> Any:
-
# First, apply our ATProto-specific decoding
-
decoded_obj = self._atproto_object_hook(obj)
-
-
# Then, apply the custom hook if provided
-
if custom_hook is not None:
-
decoded_obj = custom_hook(decoded_obj)
-
-
return decoded_obj
-
-
return combined_hook
-
-
def _atproto_object_hook(self, obj: Dict[str, Any]) -> Any:
-
"""Handle ATProto-specific object decoding.
-
-
Args:
-
obj: The object to decode.
-
-
Returns:
-
The decoded object.
-
"""
-
# Handle $bytes key (RFC-4648 base64 decoding)
-
if "$bytes" in obj:
-
if len(obj) != 1:
-
# If there are other keys, this is invalid
-
raise ValueError(f"Invalid $bytes object: {obj}")
-
return base64.b64decode(obj["$bytes"].encode(self.encoding))
-
-
# Handle $link key (CID parsing)
-
elif "$link" in obj:
-
if len(obj) != 1:
-
# If there are other keys, this is invalid
-
raise ValueError(f"Invalid $link object: {obj}")
-
return make_cid(obj["$link"])
-
-
# Handle $type key (typed objects)
-
elif "$type" in obj:
-
type_value = obj["$type"]
-
remaining_obj = {k: v for k, v in obj.items() if k != "$type"}
-
-
# Check if there's a registered type handler
-
if self.type_hook_registry is not None:
-
handler = self.type_hook_registry.get_handler(type_value)
-
if handler is not None:
-
return handler(remaining_obj)
-
-
# If no handler is registered, return a typed object
-
return TypedObject(type_value, remaining_obj)
-
-
# Handle nested objects recursively
-
elif isinstance(obj, dict):
-
return {
-
k: self._atproto_object_hook(v) if isinstance(v, dict) else v
-
for k, v in obj.items()
-
}
-
-
return obj
-
-
-
class TypedObject:
-
"""A typed object in the ATProto data model.
-
-
This class represents an object with a $type field in the ATProto data model.
-
-
Attributes:
-
type: The type of the object.
-
data: The data associated with the object.
-
"""
-
-
def __init__(self, type_name: str, data: Dict[str, Any]) -> None:
-
"""Initialize a typed object.
-
-
Args:
-
type_name: The type of the object.
-
data: The data associated with the object.
-
"""
-
self.type_name = type_name
-
self.data = data
-
-
def __repr__(self) -> str:
-
"""Return a string representation of the typed object.
-
-
Returns:
-
A string representation of the typed object.
-
"""
-
return f"TypedObject(type_name={self.type_name!r}, data={self.data!r})"
-
-
def __eq__(self, other: Any) -> bool:
-
"""Check if two typed objects are equal.
-
-
Args:
-
other: The object to compare with.
-
-
Returns:
-
True if the objects are equal, False otherwise.
-
"""
-
if not isinstance(other, TypedObject):
-
return False
-
return self.type_name == other.type_name and self.data == other.data
-
-
def __atproto_json_encode__(self) -> Dict[str, Any]:
-
"""Encode the typed object to a JSON-serializable format.
-
-
Returns:
-
A JSON-serializable representation of the typed object.
-
"""
-
result = {"$type": self.type_name}
-
result.update(self.data)
-
return result
-82
src/atpasser/data/encoder.py
···
-
"""
-
JSON encoder for ATProto data model.
-
-
This module provides a JSON encoder that handles ATProto-specific data types,
-
including bytes, CID links, and typed objects.
-
"""
-
-
import base64
-
import json
-
from typing import Any, Optional
-
from cid import CIDv0, CIDv1
-
-
-
class JsonEncoder(json.JSONEncoder):
-
"""A JSON encoder that supports ATProto data types.
-
-
This encoder extends the standard JSON encoder to handle ATProto-specific
-
data types, including bytes, CID links, and typed objects.
-
-
Attributes:
-
encoding (str): The encoding to use for string serialization.
-
type_processor_registry: Registry for type-specific processors.
-
"""
-
-
def __init__(
-
self,
-
*,
-
encoding: str = "utf-8",
-
type_processor_registry: Optional[Any] = None,
-
**kwargs: Any,
-
) -> None:
-
"""Initialize the JSON encoder.
-
-
Args:
-
encoding: The encoding to use for string serialization.
-
type_processor_registry: Registry for type-specific processors.
-
**kwargs: Additional keyword arguments to pass to the parent class.
-
"""
-
super().__init__(**kwargs)
-
self.encoding = encoding
-
self.type_processor_registry = type_processor_registry
-
-
def default(self, o: Any) -> Any:
-
"""Convert an object to a serializable format.
-
-
Args:
-
o: The object to serialize.
-
-
Returns:
-
A serializable representation of the object.
-
-
Raises:
-
TypeError: If the object is not serializable.
-
"""
-
if isinstance(o, bytes):
-
# Handle bytes using RFC-4648 base64 encoding
-
return {"$bytes": base64.b64encode(o).decode(self.encoding)}
-
elif isinstance(o, (CIDv0, CIDv1)):
-
# Handle CID objects
-
return {"$link": str(o)}
-
elif hasattr(o, "__atproto_json_encode__"):
-
# Handle objects with custom ATProto encoding
-
return o.__atproto_json_encode__()
-
elif self.type_processor_registry is not None:
-
# Try to find a type processor for this object
-
obj_type_name = type(o).__name__
-
encoder = self.type_processor_registry.get_encoder(obj_type_name)
-
if encoder is not None:
-
result = encoder(o)
-
# Add $type field if not already present
-
if isinstance(result, dict) and "$type" not in result:
-
result["$type"] = obj_type_name
-
return result
-
elif isinstance(o, dict):
-
# Handle dictionaries recursively
-
return {k: self.default(v) for k, v in o.items()}
-
elif isinstance(o, (list, tuple)):
-
# Handle lists and tuples recursively
-
return [self.default(item) for item in o]
-
else:
-
# Use the parent class for other types
-
return super().default(o)
-227
src/atpasser/data/hooks.py
···
-
"""
-
Type hook system for ATProto JSON decoder.
-
-
This module provides a decorator-based system for registering custom type handlers
-
for objects with $type keys in the ATProto data model.
-
"""
-
-
import functools
-
from typing import Any, Callable, Dict, Optional, TypeVar, Union
-
-
# Type variable for the decorated function
-
F = TypeVar("F", bound=Callable[..., Any])
-
-
-
class TypeHookRegistry:
-
"""Registry for type-specific hooks in the ATProto JSON decoder.
-
-
This class maintains a registry of type-specific hooks that can be used
-
to customize the decoding of objects with $type keys in the ATProto data model.
-
-
Attributes:
-
_handlers: Dictionary mapping type names to handler functions.
-
"""
-
-
def __init__(self) -> None:
-
"""Initialize the type hook registry."""
-
self._handlers: Dict[str, Callable[[Dict[str, Any]], Any]] = {}
-
-
def register(self, type_name: str) -> Callable[[F], F]:
-
"""Register a type handler function.
-
-
This method can be used as a decorator to register a function as a handler
-
for a specific type.
-
-
Args:
-
type_name: The name of the type to handle.
-
-
Returns:
-
A decorator function that registers the decorated function as a handler.
-
-
Example:
-
>>> registry = TypeHookRegistry()
-
>>>
-
>>> @registry.register("app.bsky.feed.post")
-
... def handle_post(data: Dict[str, Any]) -> Any:
-
... return Post(**data)
-
"""
-
-
def decorator(func: F) -> F:
-
self._handlers[type_name] = func
-
return func
-
-
return decorator
-
-
def register_handler(
-
self, type_name: str, handler: Callable[[Dict[str, Any]], Any]
-
) -> None:
-
"""Register a type handler function directly.
-
-
Args:
-
type_name: The name of the type to handle.
-
handler: The function to call when decoding objects of this type.
-
-
Example:
-
>>> registry = TypeHookRegistry()
-
>>>
-
>>> def handle_post(data: Dict[str, Any]) -> Any:
-
... return Post(**data)
-
>>>
-
>>> registry.register_handler("app.bsky.feed.post", handle_post)
-
"""
-
self._handlers[type_name] = handler
-
-
def unregister(self, type_name: str) -> None:
-
"""Unregister a type handler function.
-
-
Args:
-
type_name: The name of the type to unregister.
-
"""
-
if type_name in self._handlers:
-
del self._handlers[type_name]
-
-
def get_handler(self, type_name: str) -> Optional[Callable[[Dict[str, Any]], Any]]:
-
"""Get the handler function for a specific type.
-
-
Args:
-
type_name: The name of the type to get the handler for.
-
-
Returns:
-
The handler function for the specified type, or None if no handler
-
is registered.
-
"""
-
return self._handlers.get(type_name)
-
-
def has_handler(self, type_name: str) -> bool:
-
"""Check if a handler is registered for a specific type.
-
-
Args:
-
type_name: The name of the type to check.
-
-
Returns:
-
True if a handler is registered for the specified type, False otherwise.
-
"""
-
return type_name in self._handlers
-
-
def clear(self) -> None:
-
"""Clear all registered handlers."""
-
self._handlers.clear()
-
-
def get_registered_types(self) -> set:
-
"""Get the set of all registered type names.
-
-
Returns:
-
A set of all registered type names.
-
"""
-
return set(self._handlers.keys())
-
-
-
# Global registry instance
-
_global_registry = TypeHookRegistry()
-
-
-
def type_handler(type_name: str) -> Callable[[F], F]:
-
"""Register a global type handler function.
-
-
This decorator registers a function as a global handler for a specific type
-
in the ATProto data model.
-
-
Args:
-
type_name: The name of the type to handle.
-
-
Returns:
-
A decorator function that registers the decorated function as a handler.
-
-
Example:
-
>>> @type_handler("app.bsky.feed.post")
-
... def handle_post(data: Dict[str, Any]) -> Any:
-
... return Post(**data)
-
"""
-
return _global_registry.register(type_name)
-
-
-
def get_global_registry() -> TypeHookRegistry:
-
"""Get the global type hook registry.
-
-
Returns:
-
The global TypeHookRegistry instance.
-
"""
-
return _global_registry
-
-
-
def register_type_handler(
-
type_name: str, handler: Callable[[Dict[str, Any]], Any]
-
) -> None:
-
"""Register a global type handler function directly.
-
-
Args:
-
type_name: The name of the type to handle.
-
handler: The function to call when decoding objects of this type.
-
-
Example:
-
>>> def handle_post(data: Dict[str, Any]) -> Any:
-
... return Post(**data)
-
>>>
-
>>> register_type_handler("app.bsky.feed.post", handle_post)
-
"""
-
_global_registry.register_handler(type_name, handler)
-
-
-
def unregister_type_handler(type_name: str) -> None:
-
"""Unregister a global type handler function.
-
-
Args:
-
type_name: The name of the type to unregister.
-
"""
-
_global_registry.unregister(type_name)
-
-
-
def get_type_handler(type_name: str) -> Optional[Callable[[Dict[str, Any]], Any]]:
-
"""Get the global handler function for a specific type.
-
-
Args:
-
type_name: The name of the type to get the handler for.
-
-
Returns:
-
The handler function for the specified type, or None if no handler
-
is registered.
-
"""
-
return _global_registry.get_handler(type_name)
-
-
-
def has_type_handler(type_name: str) -> bool:
-
"""Check if a global handler is registered for a specific type.
-
-
Args:
-
type_name: The name of the type to check.
-
-
Returns:
-
True if a handler is registered for the specified type, False otherwise.
-
"""
-
return _global_registry.has_handler(type_name)
-
-
-
def clear_type_handlers() -> None:
-
"""Clear all globally registered handlers."""
-
_global_registry.clear()
-
-
-
def get_registered_types() -> set:
-
"""Get the set of all globally registered type names.
-
-
Returns:
-
A set of all registered type names.
-
"""
-
return _global_registry.get_registered_types()
-
-
-
def create_registry() -> TypeHookRegistry:
-
"""Create a new type hook registry.
-
-
This function creates a new, independent registry that can be used
-
instead of the global registry.
-
-
Returns:
-
A new TypeHookRegistry instance.
-
"""
-
return TypeHookRegistry()
-510
src/atpasser/data/types.py
···
-
"""
-
Type processor system for ATProto JSON decoder.
-
-
This module provides an advanced type processor system that allows users to
-
register custom type converters for objects with $type keys in the ATProto data model.
-
"""
-
-
import inspect
-
from typing import Any, Callable, Dict, List, Optional, Type, TypeVar, Union
-
from .hooks import TypeHookRegistry
-
-
# Type variable for the decorated class
-
T = TypeVar("T")
-
-
-
class TypeProcessor:
-
"""A type processor for ATProto JSON objects.
-
-
This class represents a processor for a specific type in the ATProto data model.
-
It contains information about how to convert JSON data to Python objects and
-
vice versa.
-
-
Attributes:
-
type_name: The name of the type this processor handles.
-
decoder: The function to decode JSON data to a Python object.
-
encoder: The function to encode a Python object to JSON data.
-
priority: The priority of this processor (higher values = higher priority).
-
"""
-
-
def __init__(
-
self,
-
type_name: str,
-
decoder: Optional[Callable[[Dict[str, Any]], Any]] = None,
-
encoder: Optional[Callable[[Any], Dict[str, Any]]] = None,
-
priority: int = 0,
-
) -> None:
-
"""Initialize a type processor.
-
-
Args:
-
type_name: The name of the type this processor handles.
-
decoder: The function to decode JSON data to a Python object.
-
encoder: The function to encode a Python object to JSON data.
-
priority: The priority of this processor (higher values = higher priority).
-
"""
-
self.type_name = type_name
-
self.decoder = decoder
-
self.encoder = encoder
-
self.priority = priority
-
-
def decode(self, data: Dict[str, Any]) -> Any:
-
"""Decode JSON data to a Python object.
-
-
Args:
-
data: The JSON data to decode.
-
-
Returns:
-
The decoded Python object.
-
-
Raises:
-
ValueError: If no decoder is registered.
-
"""
-
if self.decoder is None:
-
raise ValueError(f"No decoder registered for type {self.type_name}")
-
return self.decoder(data)
-
-
def encode(self, obj: Any) -> Dict[str, Any]:
-
"""Encode a Python object to JSON data.
-
-
Args:
-
obj: The Python object to encode.
-
-
Returns:
-
The encoded JSON data.
-
-
Raises:
-
ValueError: If no encoder is registered.
-
"""
-
if self.encoder is None:
-
raise ValueError(f"No encoder registered for type {self.type_name}")
-
return self.encoder(obj)
-
-
-
class TypeProcessorRegistry:
-
"""Registry for type processors in the ATProto JSON decoder.
-
-
This class maintains a registry of type processors that can be used
-
to customize the encoding and decoding of objects with $type keys in
-
the ATProto data model.
-
-
Attributes:
-
_processors: Dictionary mapping type names to processor lists.
-
"""
-
-
def __init__(self) -> None:
-
"""Initialize the type processor registry."""
-
self._processors: Dict[str, List[TypeProcessor]] = {}
-
-
def register_processor(self, processor: TypeProcessor) -> None:
-
"""Register a type processor.
-
-
Args:
-
processor: The type processor to register.
-
"""
-
if processor.type_name not in self._processors:
-
self._processors[processor.type_name] = []
-
-
self._processors[processor.type_name].append(processor)
-
# Sort processors by priority (descending)
-
self._processors[processor.type_name].sort(
-
key=lambda p: p.priority, reverse=True
-
)
-
-
def register(
-
self, type_name: str, priority: int = 0
-
) -> Callable[[Callable[[Dict[str, Any]], Any]], Callable[[Dict[str, Any]], Any]]:
-
"""Register a type decoder function.
-
-
This method can be used as a decorator to register a function as a decoder
-
for a specific type.
-
-
Args:
-
type_name: The name of the type to handle.
-
priority: The priority of this processor (higher values = higher priority).
-
-
Returns:
-
A decorator function that registers the decorated function as a decoder.
-
-
Example:
-
>>> registry = TypeProcessorRegistry()
-
>>>
-
>>> @registry.register("app.bsky.feed.post", priority=10)
-
... def decode_post(data: Dict[str, Any]) -> Any:
-
... return Post(**data)
-
"""
-
-
def decorator(
-
func: Callable[[Dict[str, Any]], Any],
-
) -> Callable[[Dict[str, Any]], Any]:
-
processor = TypeProcessor(type_name, decoder=func, priority=priority)
-
self.register_processor(processor)
-
return func
-
-
return decorator
-
-
def register_encoder(
-
self, type_name: str, priority: int = 0
-
) -> Callable[[Callable[[Any], Dict[str, Any]]], Callable[[Any], Dict[str, Any]]]:
-
"""Register a type encoder function.
-
-
This method can be used as a decorator to register a function as an encoder
-
for a specific type.
-
-
Args:
-
type_name: The name of the type to handle.
-
priority: The priority of this processor (higher values = higher priority).
-
-
Returns:
-
A decorator function that registers the decorated function as an encoder.
-
-
Example:
-
>>> registry = TypeProcessorRegistry()
-
>>>
-
>>> @registry.register_encoder("app.bsky.feed.post", priority=10)
-
... def encode_post(post: Post) -> Dict[str, Any]:
-
... return {"text": post.text, "createdAt": post.created_at}
-
"""
-
-
def decorator(
-
func: Callable[[Any], Dict[str, Any]],
-
) -> Callable[[Any], Dict[str, Any]]:
-
# Check if a processor for this type already exists
-
if type_name in self._processors:
-
for processor in self._processors[type_name]:
-
if processor.decoder is not None:
-
# Update the existing processor with the encoder
-
processor.encoder = func
-
break
-
else:
-
# No decoder found, create a new processor
-
processor = TypeProcessor(
-
type_name, encoder=func, priority=priority
-
)
-
self.register_processor(processor)
-
else:
-
# No processor exists, create a new one
-
processor = TypeProcessor(type_name, encoder=func, priority=priority)
-
self.register_processor(processor)
-
-
return func
-
-
return decorator
-
-
def register_class(
-
self, type_name: str, priority: int = 0
-
) -> Callable[[Type[T]], Type[T]]:
-
"""Register a class for both encoding and decoding.
-
-
This method can be used as a decorator to register a class for both
-
encoding and decoding of a specific type.
-
-
The class must have a class method `from_json` that takes a dictionary
-
and returns an instance of the class, and an instance method `to_json`
-
that returns a dictionary.
-
-
Args:
-
type_name: The name of the type to handle.
-
priority: The priority of this processor (higher values = higher priority).
-
-
Returns:
-
A decorator function that registers the decorated class.
-
-
Example:
-
>>> registry = TypeProcessorRegistry()
-
>>>
-
>>> @registry.register_class("app.bsky.feed.post", priority=10)
-
... class Post:
-
... def __init__(self, text: str, created_at: str) -> None:
-
... self.text = text
-
... self.created_at = created_at
-
...
-
... @classmethod
-
... def from_json(cls, data: Dict[str, Any]) -> "Post":
-
... return cls(data["text"], data["createdAt"])
-
...
-
... def to_json(self) -> Dict[str, Any]:
-
... return {"text": self.text, "createdAt": self.created_at}
-
"""
-
-
def decorator(cls: Type[T]) -> Type[T]:
-
# Create decoder from class method
-
if hasattr(cls, "from_json"):
-
decoder = lambda data: getattr(cls, "from_json")(data)
-
else:
-
# Try to create a decoder from the constructor
-
init_signature = inspect.signature(cls.__init__)
-
if init_signature.parameters:
-
# Create a decoder that passes the data as keyword arguments
-
decoder = lambda data: cls(**data)
-
else:
-
raise ValueError(
-
f"Class {cls.__name__} has no from_json method or compatible __init__"
-
)
-
-
# Create encoder from instance method
-
if hasattr(cls, "to_json"):
-
encoder = lambda obj: obj.to_json()
-
else:
-
raise ValueError(f"Class {cls.__name__} has no to_json method")
-
-
# Register the processor
-
processor = TypeProcessor(
-
type_name, decoder=decoder, encoder=encoder, priority=priority
-
)
-
self.register_processor(processor)
-
-
return cls
-
-
return decorator
-
-
def unregister(self, type_name: str, priority: Optional[int] = None) -> None:
-
"""Unregister type processors.
-
-
Args:
-
type_name: The name of the type to unregister.
-
priority: If specified, only unregister processors with this priority.
-
"""
-
if type_name in self._processors:
-
if priority is not None:
-
# Remove processors with the specified priority
-
self._processors[type_name] = [
-
p for p in self._processors[type_name] if p.priority != priority
-
]
-
else:
-
# Remove all processors for this type
-
del self._processors[type_name]
-
-
def get_decoder(self, type_name: str) -> Optional[Callable[[Dict[str, Any]], Any]]:
-
"""Get the decoder function for a specific type.
-
-
Args:
-
type_name: The name of the type to get the decoder for.
-
-
Returns:
-
The decoder function for the specified type, or None if no decoder
-
is registered.
-
"""
-
if type_name in self._processors and self._processors[type_name]:
-
# Return the decoder of the highest priority processor
-
return self._processors[type_name][0].decoder
-
return None
-
-
def get_encoder(self, type_name: str) -> Optional[Callable[[Any], Dict[str, Any]]]:
-
"""Get the encoder function for a specific type.
-
-
Args:
-
type_name: The name of the type to get the encoder for.
-
-
Returns:
-
The encoder function for the specified type, or None if no encoder
-
is registered.
-
"""
-
if type_name in self._processors and self._processors[type_name]:
-
# Return the encoder of the highest priority processor
-
return self._processors[type_name][0].encoder
-
return None
-
-
def has_processor(self, type_name: str) -> bool:
-
"""Check if a processor is registered for a specific type.
-
-
Args:
-
type_name: The name of the type to check.
-
-
Returns:
-
True if a processor is registered for the specified type, False otherwise.
-
"""
-
return type_name in self._processors and bool(self._processors[type_name])
-
-
def clear(self) -> None:
-
"""Clear all registered processors."""
-
self._processors.clear()
-
-
def get_registered_types(self) -> set:
-
"""Get the set of all registered type names.
-
-
Returns:
-
A set of all registered type names.
-
"""
-
return set(self._processors.keys())
-
-
def to_hook_registry(self) -> TypeHookRegistry:
-
"""Convert this processor registry to a hook registry.
-
-
This method creates a TypeHookRegistry that uses the decoders from
-
this processor registry.
-
-
Returns:
-
A TypeHookRegistry with the same decoders as this processor registry.
-
"""
-
hook_registry = TypeHookRegistry()
-
-
for type_name, processors in self._processors.items():
-
if processors and processors[0].decoder is not None:
-
hook_registry.register_handler(type_name, processors[0].decoder)
-
-
return hook_registry
-
-
-
# Global registry instance
-
_global_processor_registry = TypeProcessorRegistry()
-
-
-
def register_type(
-
type_name: str, priority: int = 0
-
) -> Callable[[Callable[[Dict[str, Any]], Any]], Callable[[Dict[str, Any]], Any]]:
-
"""Register a global type decoder function.
-
-
This decorator registers a function as a global decoder for a specific type
-
in the ATProto data model.
-
-
Args:
-
type_name: The name of the type to handle.
-
priority: The priority of this processor (higher values = higher priority).
-
-
Returns:
-
A decorator function that registers the decorated function as a decoder.
-
-
Example:
-
>>> @register_type("app.bsky.feed.post", priority=10)
-
... def decode_post(data: Dict[str, Any]) -> Any:
-
... return Post(**data)
-
"""
-
return _global_processor_registry.register(type_name, priority)
-
-
-
def get_global_processor_registry() -> TypeProcessorRegistry:
-
"""Get the global type processor registry.
-
-
Returns:
-
The global TypeProcessorRegistry instance.
-
"""
-
return _global_processor_registry
-
-
-
def register_type_encoder(
-
type_name: str, priority: int = 0
-
) -> Callable[[Callable[[Any], Dict[str, Any]]], Callable[[Any], Dict[str, Any]]]:
-
"""Register a global type encoder function.
-
-
This decorator registers a function as a global encoder for a specific type
-
in the ATProto data model.
-
-
Args:
-
type_name: The name of the type to handle.
-
priority: The priority of this processor (higher values = higher priority).
-
-
Returns:
-
A decorator function that registers the decorated function as an encoder.
-
-
Example:
-
>>> @register_type_encoder("app.bsky.feed.post", priority=10)
-
... def encode_post(post: Post) -> Dict[str, Any]:
-
... return {"text": post.text, "createdAt": post.created_at}
-
"""
-
return _global_processor_registry.register_encoder(type_name, priority)
-
-
-
def register_type_class(
-
type_name: str, priority: int = 0
-
) -> Callable[[Type[T]], Type[T]]:
-
"""Register a class for both global encoding and decoding.
-
-
This decorator registers a class for both encoding and decoding of a specific type
-
in the ATProto data model.
-
-
Args:
-
type_name: The name of the type to handle.
-
priority: The priority of this processor (higher values = higher priority).
-
-
Returns:
-
A decorator function that registers the decorated class.
-
-
Example:
-
>>> @register_type_class("app.bsky.feed.post", priority=10)
-
... class Post:
-
... def __init__(self, text: str, created_at: str) -> None:
-
... self.text = text
-
... self.created_at = created_at
-
...
-
... @classmethod
-
... def from_json(cls, data: Dict[str, Any]) -> "Post":
-
... return cls(data["text"], data["createdAt"])
-
...
-
... def to_json(self) -> Dict[str, Any]:
-
... return {"text": self.text, "createdAt": self.created_at}
-
"""
-
return _global_processor_registry.register_class(type_name, priority)
-
-
-
def unregister_type(type_name: str, priority: Optional[int] = None) -> None:
-
"""Unregister global type processors.
-
-
Args:
-
type_name: The name of the type to unregister.
-
priority: If specified, only unregister processors with this priority.
-
"""
-
_global_processor_registry.unregister(type_name, priority)
-
-
-
def get_type_decoder(type_name: str) -> Optional[Callable[[Dict[str, Any]], Any]]:
-
"""Get the global decoder function for a specific type.
-
-
Args:
-
type_name: The name of the type to get the decoder for.
-
-
Returns:
-
The decoder function for the specified type, or None if no decoder
-
is registered.
-
"""
-
return _global_processor_registry.get_decoder(type_name)
-
-
-
def get_type_encoder(type_name: str) -> Optional[Callable[[Any], Dict[str, Any]]]:
-
"""Get the global encoder function for a specific type.
-
-
Args:
-
type_name: The name of the type to get the encoder for.
-
-
Returns:
-
The encoder function for the specified type, or None if no encoder
-
is registered.
-
"""
-
return _global_processor_registry.get_encoder(type_name)
-
-
-
def has_type_processor(type_name: str) -> bool:
-
"""Check if a global processor is registered for a specific type.
-
-
Args:
-
type_name: The name of the type to check.
-
-
Returns:
-
True if a processor is registered for the specified type, False otherwise.
-
"""
-
return _global_processor_registry.has_processor(type_name)
-
-
-
def clear_type_processors() -> None:
-
"""Clear all globally registered processors."""
-
_global_processor_registry.clear()
-
-
-
def get_registered_types() -> set:
-
"""Get the set of all globally registered type names.
-
-
Returns:
-
A set of all registered type names.
-
"""
-
return _global_processor_registry.get_registered_types()
-
-
-
def create_processor_registry() -> TypeProcessorRegistry:
-
"""Create a new type processor registry.
-
-
This function creates a new, independent registry that can be used
-
instead of the global registry.
-
-
Returns:
-
A new TypeProcessorRegistry instance.
-
"""
-
return TypeProcessorRegistry()
-346
src/atpasser/data/wrapper.py
···
-
"""
-
JSON wrapper functions for ATProto data model.
-
-
This module provides wrapper functions that mirror the standard json module
-
but with support for ATProto-specific data types.
-
"""
-
-
import json
-
import io
-
from typing import Any, Callable, Dict, Optional, TextIO, Union
-
from .encoder import JsonEncoder
-
from .decoder import JsonDecoder
-
from .hooks import TypeHookRegistry
-
from .types import TypeProcessorRegistry
-
-
-
def dump(
-
obj: Any,
-
fp: TextIO,
-
*,
-
skipkeys: bool = False,
-
ensure_ascii: bool = True,
-
check_circular: bool = True,
-
allow_nan: bool = True,
-
cls: Optional[type[JsonEncoder]] = None,
-
indent: Optional[Union[int, str]] = None,
-
separators: Optional[tuple[str, str]] = None,
-
default: Optional[Callable[[Any], Any]] = None,
-
sort_keys: bool = False,
-
encoding: str = "utf-8",
-
type_processor_registry: Optional[TypeProcessorRegistry] = None,
-
**kwargs: Any,
-
) -> None:
-
"""Serialize obj as a JSON formatted stream to fp.
-
-
This function is similar to json.dump() but supports ATProto-specific
-
data types, including bytes, CID links, and typed objects.
-
-
Args:
-
obj: The object to serialize.
-
fp: A file-like object with a write() method.
-
skipkeys: If True, dict keys that are not basic types (str, int, float,
-
bool, None) will be skipped instead of raising a TypeError.
-
ensure_ascii: If True, the output is guaranteed to have all incoming
-
non-ASCII characters escaped. If False, these characters will be
-
output as-is.
-
check_circular: If True, circular references will be checked and
-
a CircularReferenceError will be raised if one is found.
-
allow_nan: If True, NaN, Infinity, and -Infinity will be encoded as
-
such. This behavior is not JSON specification compliant, but it
-
is consistent with most JavaScript based encoders and decoders.
-
Otherwise, it will raise a ValueError.
-
cls: A custom JSONEncoder subclass. If not specified, JsonEncoder is used.
-
indent: If indent is a non-negative integer or string, then JSON array
-
elements and object members will be pretty-printed with that indent
-
level. An indent level of 0, negative, or "" will only insert newlines.
-
None (the default) selects the most compact representation.
-
separators: If specified, separators should be an (item_separator, key_separator)
-
tuple. The default is (', ', ': ') if indent is None and (',', ': ') otherwise.
-
To get the most compact JSON representation, you should specify (',', ':')
-
to eliminate whitespace.
-
default: If specified, default should be a function that gets called for
-
objects that can't otherwise be serialized. It should return a JSON
-
encodable version of the object or raise a TypeError.
-
sort_keys: If sort_keys is True, then the output of dictionaries will be
-
sorted by key.
-
encoding: The encoding to use for string serialization.
-
type_processor_registry: Registry for type-specific processors.
-
**kwargs: Additional keyword arguments to pass to the JSON encoder.
-
"""
-
if cls is None:
-
cls = JsonEncoder
-
-
# Use the global type processor registry if none is provided
-
if type_processor_registry is None:
-
from .types import get_global_processor_registry
-
-
type_processor_registry = get_global_processor_registry()
-
-
# Create an encoder instance with the specified encoding and type processor registry
-
encoder = cls(
-
encoding=encoding, type_processor_registry=type_processor_registry, **kwargs
-
)
-
-
# Use the standard json.dump with our custom encoder
-
json.dump(
-
obj,
-
fp,
-
skipkeys=skipkeys,
-
ensure_ascii=ensure_ascii,
-
check_circular=check_circular,
-
allow_nan=allow_nan,
-
cls=cls,
-
indent=indent,
-
separators=separators,
-
default=default,
-
sort_keys=sort_keys,
-
**kwargs,
-
)
-
-
-
def dumps(
-
obj: Any,
-
*,
-
skipkeys: bool = False,
-
ensure_ascii: bool = True,
-
check_circular: bool = True,
-
allow_nan: bool = True,
-
cls: Optional[type[JsonEncoder]] = None,
-
indent: Optional[Union[int, str]] = None,
-
separators: Optional[tuple[str, str]] = None,
-
default: Optional[Callable[[Any], Any]] = None,
-
sort_keys: bool = False,
-
encoding: str = "utf-8",
-
type_processor_registry: Optional[TypeProcessorRegistry] = None,
-
**kwargs: Any,
-
) -> str:
-
"""Serialize obj to a JSON formatted string.
-
-
This function is similar to json.dumps() but supports ATProto-specific
-
data types, including bytes, CID links, and typed objects.
-
-
Args:
-
obj: The object to serialize.
-
skipkeys: If True, dict keys that are not basic types (str, int, float,
-
bool, None) will be skipped instead of raising a TypeError.
-
ensure_ascii: If True, the output is guaranteed to have all incoming
-
non-ASCII characters escaped. If False, these characters will be
-
output as-is.
-
check_circular: If True, circular references will be checked and
-
a CircularReferenceError will be raised if one is found.
-
allow_nan: If True, NaN, Infinity, and -Infinity will be encoded as
-
such. This behavior is not JSON specification compliant, but it
-
is consistent with most JavaScript based encoders and decoders.
-
Otherwise, it will raise a ValueError.
-
cls: A custom JSONEncoder subclass. If not specified, JsonEncoder is used.
-
indent: If indent is a non-negative integer or string, then JSON array
-
elements and object members will be pretty-printed with that indent
-
level. An indent level of 0, negative, or "" will only insert newlines.
-
None (the default) selects the most compact representation.
-
separators: If specified, separators should be an (item_separator, key_separator)
-
tuple. The default is (', ', ': ') if indent is None and (',', ': ') otherwise.
-
To get the most compact JSON representation, you should specify (',', ':')
-
to eliminate whitespace.
-
default: If specified, default should be a function that gets called for
-
objects that can't otherwise be serialized. It should return a JSON
-
encodable version of the object or raise a TypeError.
-
sort_keys: If sort_keys is True, then the output of dictionaries will be
-
sorted by key.
-
encoding: The encoding to use for string serialization.
-
type_processor_registry: Registry for type-specific processors.
-
**kwargs: Additional keyword arguments to pass to the JSON encoder.
-
-
Returns:
-
A JSON formatted string.
-
"""
-
if cls is None:
-
cls = JsonEncoder
-
-
# Create an encoder instance with the specified encoding and type processor registry
-
encoder = cls(
-
encoding=encoding, type_processor_registry=type_processor_registry, **kwargs
-
)
-
-
# Use the standard json.dumps with our custom encoder
-
return json.dumps(
-
obj,
-
skipkeys=skipkeys,
-
ensure_ascii=ensure_ascii,
-
check_circular=check_circular,
-
allow_nan=allow_nan,
-
cls=cls,
-
indent=indent,
-
separators=separators,
-
default=default,
-
sort_keys=sort_keys,
-
**kwargs,
-
)
-
-
-
def load(
-
fp: TextIO,
-
*,
-
cls: Optional[type[JsonDecoder]] = None,
-
object_hook: Optional[Callable[[Dict[str, Any]], Any]] = None,
-
parse_float: Optional[Callable[[str], Any]] = None,
-
parse_int: Optional[Callable[[str], Any]] = None,
-
parse_constant: Optional[Callable[[str], Any]] = None,
-
object_pairs_hook: Optional[Callable[[list[tuple[str, Any]]], Any]] = None,
-
type_hook_registry: Optional[TypeHookRegistry] = None,
-
type_processor_registry: Optional[TypeProcessorRegistry] = None,
-
encoding: str = "utf-8",
-
**kwargs: Any,
-
) -> Any:
-
"""Deserialize fp (a .read()-supporting text file or binary file containing
-
a JSON document) to a Python object.
-
-
This function is similar to json.load() but supports ATProto-specific
-
data types, including bytes, CID links, and typed objects.
-
-
Args:
-
fp: A .read()-supporting text file or binary file containing a JSON document.
-
cls: A custom JSONDecoder subclass. If not specified, JsonDecoder is used.
-
object_hook: Optional function that will be called with the result of
-
every JSON object decoded and its return value will be used in place
-
of the given dict.
-
parse_float: Optional function that will be called with the string of
-
every JSON float to be decoded. By default, this is equivalent to
-
float(num_str). This can be used to use another datatype or parser
-
for JSON floats (e.g. decimal.Decimal).
-
parse_int: Optional function that will be called with the string of
-
every JSON int to be decoded. By default, this is equivalent to
-
int(num_str). This can be used to use another datatype or parser
-
for JSON integers (e.g. float).
-
parse_constant: Optional function that will be called with the string of
-
every JSON constant to be decoded. By default, this is equivalent to
-
constant_mapping[constant_str]. This can be used to use another
-
datatype or parser for JSON constants (e.g. decimal.Decimal).
-
object_pairs_hook: Optional function that will be called with the result
-
of every JSON object decoded with an ordered list of pairs. The return
-
value of object_pairs_hook will be used instead of the dict. This
-
feature can be used to implement custom decoders. If object_hook is
-
also defined, the object_pairs_hook takes priority.
-
type_hook_registry: Registry for type-specific hooks.
-
type_processor_registry: Registry for type-specific processors.
-
encoding: The encoding to use for string deserialization.
-
**kwargs: Additional keyword arguments to pass to the JSON decoder.
-
-
Returns:
-
A Python object.
-
"""
-
if cls is None:
-
cls = JsonDecoder
-
-
# Use the global type hook registry if none is provided
-
if type_hook_registry is None and type_processor_registry is None:
-
from .hooks import get_global_registry
-
-
type_hook_registry = get_global_registry()
-
elif type_processor_registry is not None:
-
# Convert the type processor registry to a hook registry
-
type_hook_registry = type_processor_registry.to_hook_registry()
-
-
# Create a decoder instance with the specified parameters
-
decoder = cls(
-
object_hook=object_hook,
-
type_hook_registry=type_hook_registry,
-
encoding=encoding,
-
**kwargs,
-
)
-
-
# Use the standard json.load with our custom decoder
-
return json.load(
-
fp,
-
cls=cls,
-
object_hook=object_hook,
-
parse_float=parse_float,
-
parse_int=parse_int,
-
parse_constant=parse_constant,
-
object_pairs_hook=object_pairs_hook,
-
**kwargs,
-
)
-
-
-
def loads(
-
s: Union[str, bytes],
-
*,
-
cls: Optional[type[JsonDecoder]] = None,
-
object_hook: Optional[Callable[[Dict[str, Any]], Any]] = None,
-
parse_float: Optional[Callable[[str], Any]] = None,
-
parse_int: Optional[Callable[[str], Any]] = None,
-
parse_constant: Optional[Callable[[str], Any]] = None,
-
object_pairs_hook: Optional[Callable[[list[tuple[str, Any]]], Any]] = None,
-
type_hook_registry: Optional[TypeHookRegistry] = None,
-
type_processor_registry: Optional[TypeProcessorRegistry] = None,
-
encoding: str = "utf-8",
-
**kwargs: Any,
-
) -> Any:
-
"""Deserialize s (a str, bytes or bytearray instance containing a JSON document)
-
to a Python object.
-
-
This function is similar to json.loads() but supports ATProto-specific
-
data types, including bytes, CID links, and typed objects.
-
-
Args:
-
s: A str, bytes or bytearray instance containing a JSON document.
-
cls: A custom JSONDecoder subclass. If not specified, JsonDecoder is used.
-
object_hook: Optional function that will be called with the result of
-
every JSON object decoded and its return value will be used in place
-
of the given dict.
-
parse_float: Optional function that will be called with the string of
-
every JSON float to be decoded. By default, this is equivalent to
-
float(num_str). This can be used to use another datatype or parser
-
for JSON floats (e.g. decimal.Decimal).
-
parse_int: Optional function that will be called with the string of
-
every JSON int to be decoded. By default, this is equivalent to
-
int(num_str). This can be used to use another datatype or parser
-
for JSON integers (e.g. float).
-
parse_constant: Optional function that will be called with the string of
-
every JSON constant to be decoded. By default, this is equivalent to
-
constant_mapping[constant_str]. This can be used to use another
-
datatype or parser for JSON constants (e.g. decimal.Decimal).
-
object_pairs_hook: Optional function that will be called with the result
-
of every JSON object decoded with an ordered list of pairs. The return
-
value of object_pairs_hook will be used instead of the dict. This
-
feature can be used to implement custom decoders. If object_hook is
-
also defined, the object_pairs_hook takes priority.
-
type_hook_registry: Registry for type-specific hooks.
-
type_processor_registry: Registry for type-specific processors.
-
encoding: The encoding to use for string deserialization.
-
**kwargs: Additional keyword arguments to pass to the JSON decoder.
-
-
Returns:
-
A Python object.
-
"""
-
if cls is None:
-
cls = JsonDecoder
-
-
# Use the global type hook registry if none is provided
-
if type_hook_registry is None and type_processor_registry is None:
-
from .hooks import get_global_registry
-
-
type_hook_registry = get_global_registry()
-
elif type_processor_registry is not None:
-
# Convert the type processor registry to a hook registry
-
type_hook_registry = type_processor_registry.to_hook_registry()
-
-
# Create a decoder instance with the specified parameters
-
decoder = cls(
-
object_hook=object_hook,
-
type_hook_registry=type_hook_registry,
-
encoding=encoding,
-
**kwargs,
-
)
-
-
# Use the standard json.loads with our custom decoder
-
return json.loads(
-
s,
-
cls=cls,
-
object_hook=object_hook,
-
parse_float=parse_float,
-
parse_int=parse_int,
-
parse_constant=parse_constant,
-
object_pairs_hook=object_pairs_hook,
-
**kwargs,
-
)
+6 -1
.gitignore
···
__marimo__/
# Streamlit
-
.streamlit/secrets.toml
+
.streamlit/secrets.toml
+
+
#####
+
+
# Added by DWN - temp dir
+
tmp/
+10 -2
README.md
···
-
# ATPasser โ•ฌ<!
+
# ATPasser!
A simple library for the [Authenticated Transfer Protocol](https://atproto.com/specs/atp) (AT Protocol or atproto for short).
···
---
-
[See the roadmap](docs/roadmap.md)
+
## Other ATProto libraries
+
+
[There's an ATProto SDK already (and used by lots of projects) by MarshalX,](https://github.com/MarshalX/atproto) and why do this exists?
+
+
The first reason is that I'm recovering the now-closed [Tietiequan](https://tangled.org/@dwn.dwnfonts.cc/bluesky-circle) app and found that some API has changed so I have to rewrite it via vanilla JS.
+
+
The second reason is that I'm a newbie in ATProto, wanting to know how ATProto is, and how this can be represented in Python.
+
+
The architecture will be small, only containing the data model and the client.
---
-16
src/atpasser/blob/__init__.py
···
-
import cid
-
import multihash, hashlib
-
-
-
def generateCID(file):
-
hasher = hashlib.new("sha-256")
-
while True:
-
chunk = file.read(8192)
-
if not chunk:
-
break
-
hasher.update(chunk)
-
-
digest = hasher.digest
-
mh = multihash.encode(digest, "sha-256")
-
-
return cid.CIDv1(codec="raw", multihash=mh)
-76
src/atpasser/data/_data.py
···
-
import base64
-
from cid import CIDv0, CIDv1, cid, make_cid
-
import json
-
-
-
class Data:
-
"""
-
A class representing data with "$type" key.
-
-
Attributes:
-
type (str): The type of the data.
-
json (str): Original object in JSON format.
-
"""
-
-
def __init__(self, dataType: str, json: str = "{}") -> None:
-
"""
-
Initalizes data object.
-
-
Parameters:
-
type (str): The type of the data.
-
json (str): Original object in JSON format.
-
"""
-
self.type = dataType
-
self.json = json
-
-
def data(self):
-
"""
-
Loads data as a Python-friendly format.
-
-
Returns:
-
dict: Converted data from JSON object.
-
"""
-
return json.loads(self.json, object_hook=dataHook)
-
-
-
def dataHook(data: dict):
-
"""
-
Treated as `JSONDecoder`'s `object_hook`
-
-
Parameters:
-
data: data in format that `JSONDecoder` like ;)
-
"""
-
if "$bytes" in data:
-
return base64.b64decode(data["$bytes"])
-
elif "$link" in data:
-
return make_cid(data["$link"])
-
elif "$type" in data:
-
dataType = data["$type"]
-
del data["$type"]
-
return Data(dataType, json.dumps(data))
-
else:
-
return data
-
-
-
def _convertDataToFakeJSON(data):
-
if isinstance(data, bytes):
-
return {"$bytes": base64.b64encode(data)}
-
elif isinstance(data, (CIDv0, CIDv1)):
-
return {"link": data.encode()}
-
elif isinstance(data, dict):
-
for item in data:
-
data[item] = _convertDataToFakeJSON(data[item])
-
elif isinstance(data, (tuple, list, set)):
-
return [_convertDataToFakeJSON(item) for item in data]
-
else:
-
return data
-
-
-
class DataEncoder(json.JSONEncoder):
-
"""
-
A superset of `JSONEncoder` to support ATProto data.
-
"""
-
-
def default(self, o):
-
result = _convertDataToFakeJSON(o)
-
return super().default(result)
-61
src/atpasser/data/_wrapper.py
···
-
from json import loads
-
from typing import Callable, Any
-
from ._data import *
-
import functools
-
-
# Pyright did the whole job. Thank it.
-
-
-
class DataDecoder(json.JSONDecoder):
-
"""
-
A superset of `JSONDecoder` to support ATProto data.
-
"""
-
-
def __init__(
-
self,
-
*,
-
object_hook: Callable[[dict[str, Any]], Any] | None = dataHook,
-
parse_float: Callable[[str], Any] | None = None,
-
parse_int: Callable[[str], Any] | None = None,
-
parse_constant: Callable[[str], Any] | None = None,
-
strict: bool = True,
-
object_pairs_hook: Callable[[list[tuple[str, Any]]], Any] | None = None,
-
) -> None:
-
super().__init__(
-
object_hook=object_hook,
-
parse_float=parse_float,
-
parse_int=parse_int,
-
parse_constant=parse_constant,
-
strict=strict,
-
object_pairs_hook=object_pairs_hook,
-
)
-
-
-
# Screw it. I have to make 4 `json`-like functions.
-
-
-
def _dataDecoratorForDump(func):
-
@functools.wraps(func)
-
def wrapper(obj, *args, **kwargs):
-
kwargs.setdefault("cls", DataEncoder)
-
return func(obj, *args, **kwargs)
-
-
return wrapper
-
-
-
def _dataDecoratorForLoad(func):
-
@functools.wraps(func)
-
def wrapper(obj, *args, **kwargs):
-
kwargs.setdefault("cls", DataDecoder)
-
return func(obj, *args, **kwargs)
-
-
return wrapper
-
-
-
dump = _dataDecoratorForDump(json.dump)
-
dumps = _dataDecoratorForDump(json.dumps)
-
load = _dataDecoratorForLoad(json.load)
-
loads = _dataDecoratorForLoad(json.loads)
-
"""
-
Wrapper of the JSON functions to support ATProto data.
-
"""
-137
src/atpasser/data/cbor.py
···
-
from datetime import tzinfo
-
import typing
-
import cbor2
-
import cid
-
-
from .data import dataHook, Data
-
-
-
def tagHook(decoder: cbor2.CBORDecoder, tag: cbor2.CBORTag, shareable_index=None):
-
"""
-
A simple tag hook for CID support.
-
"""
-
return cid.from_bytes(tag.value) if tag.tag == 42 else tag
-
-
-
class CBOREncoder(cbor2.CBOREncoder):
-
"""
-
Wrapper of cbor2.CBOREncoder.
-
"""
-
-
def __init__(
-
self,
-
fp: typing.IO[bytes],
-
datetime_as_timestamp: bool = False,
-
timezone: tzinfo | None = None,
-
value_sharing: bool = False,
-
default: (
-
typing.Callable[[cbor2.CBOREncoder, typing.Any], typing.Any] | None
-
) = None,
-
canonical: bool = False,
-
date_as_datetime: bool = False,
-
string_referencing: bool = False,
-
indefinite_containers: bool = False,
-
):
-
super().__init__(
-
fp,
-
datetime_as_timestamp,
-
timezone,
-
value_sharing,
-
default,
-
canonical,
-
date_as_datetime,
-
string_referencing,
-
indefinite_containers,
-
)
-
-
@cbor2.shareable_encoder
-
def cidOrDataEncoder(self: cbor2.CBOREncoder, value: cid.CIDv0 | cid.CIDv1 | Data):
-
"""
-
Encode CID or Data to CBOR Tag.
-
"""
-
if isinstance(value, (cid.CIDv0, cid.CIDv1)):
-
self.encode(cbor2.CBORTag(42, value.encode()))
-
elif isinstance(value, Data):
-
self.encode(value.data())
-
-
-
def _cborObjectHook(decoder: cbor2.CBORDecoder, value):
-
return dataHook(value)
-
-
-
class CBORDecoder(cbor2.CBORDecoder):
-
"""
-
Wrapper of cbor2.CBORDecoder.
-
"""
-
-
def __init__(
-
self,
-
fp: typing.IO[bytes],
-
tag_hook: (
-
typing.Callable[[cbor2.CBORDecoder, cbor2.CBORTag], typing.Any] | None
-
) = tagHook,
-
object_hook: (
-
typing.Callable[
-
[cbor2.CBORDecoder, dict[typing.Any, typing.Any]], typing.Any
-
]
-
| None
-
) = _cborObjectHook,
-
str_errors: typing.Literal["strict", "error", "replace"] = "strict",
-
):
-
super().__init__(fp, tag_hook, object_hook, str_errors)
-
-
-
# Make things for CBOR again.
-
-
from io import BytesIO
-
-
-
def dumps(
-
obj: object,
-
datetime_as_timestamp: bool = False,
-
timezone: tzinfo | None = None,
-
value_sharing: bool = False,
-
default: typing.Callable[[cbor2.CBOREncoder, typing.Any], typing.Any] | None = None,
-
canonical: bool = False,
-
date_as_datetime: bool = False,
-
string_referencing: bool = False,
-
indefinite_containers: bool = False,
-
) -> bytes:
-
with BytesIO() as fp:
-
CBOREncoder(
-
fp,
-
datetime_as_timestamp=datetime_as_timestamp,
-
timezone=timezone,
-
value_sharing=value_sharing,
-
default=default,
-
canonical=canonical,
-
date_as_datetime=date_as_datetime,
-
string_referencing=string_referencing,
-
indefinite_containers=indefinite_containers,
-
).encode(obj)
-
return fp.getvalue()
-
-
-
def dump(
-
obj: object,
-
fp: typing.IO[bytes],
-
datetime_as_timestamp: bool = False,
-
timezone: tzinfo | None = None,
-
value_sharing: bool = False,
-
default: typing.Callable[[cbor2.CBOREncoder, typing.Any], typing.Any] | None = None,
-
canonical: bool = False,
-
date_as_datetime: bool = False,
-
string_referencing: bool = False,
-
indefinite_containers: bool = False,
-
) -> None:
-
CBOREncoder(
-
fp,
-
datetime_as_timestamp=datetime_as_timestamp,
-
timezone=timezone,
-
value_sharing=value_sharing,
-
default=default,
-
canonical=canonical,
-
date_as_datetime=date_as_datetime,
-
string_referencing=string_referencing,
-
indefinite_containers=indefinite_containers,
-
).encode(obj)
-179
tests/_strings.py
···
-
from atpasser import did, handle, nsid, rKey, uri
-
-
-
testStrings, testMethods = {}, {}
-
-
-
testStrings[
-
"did"
-
-
] = """did:plc:z72i7hdynmk6r22z27h6tvur
-
-
did:web:blueskyweb.xyz
-
-
did:method:val:two
-
-
did:m:v
-
-
did:method::::val
-
-
did:method:-:_:.
-
-
did:key:zQ3shZc2QzApp2oymGvQbzP8eKheVshBHbU4ZYjeXqwSKEn6N
-
-
did:METHOD:val
-
-
did:m123:val
-
-
DID:method:val
-
did:method:
-
-
did:method:val/two
-
-
did:method:val?two
-
-
did:method:val#two"""
-
-
testMethods["did"] = did.DID
-
-
-
testStrings[
-
"handle"
-
-
] = """jay.bsky.social
-
-
8.cn
-
-
name.t--t
-
-
XX.LCS.MIT.EDU
-
a.co
-
-
xn--notarealidn.com
-
-
xn--fiqa61au8b7zsevnm8ak20mc4a87e.xn--fiqs8s
-
-
xn--ls8h.test
-
example.t
-
-
jo@hn.test
-
-
๐Ÿ’ฉ.tes
-
t
-
john..test
-
-
xn--bcher-.tld
-
-
john.0
-
-
cn.8
-
-
www.maseล‚kowski.pl.com
-
-
org
-
-
name.org.
-
-
2gzyxa5ihm7nsggfxnu52rck2vv4rvmdlkiu3zzui5du4xyclen53wid.onion
-
laptop.local
-
-
blah.arpa"""
-
-
testMethods["handle"] = handle.Handle
-
-
-
testStrings[
-
"nsid"
-
-
] = """com.example.fooBar
-
-
net.users.bob.ping
-
-
a-0.b-1.c
-
-
a.b.c
-
-
com.example.fooBarV2
-
-
cn.8.lex.stuff
-
-
com.exa๐Ÿ’ฉple.thin
-
com.example
-
-
com.example.3"""
-
-
testMethods["nsid"] = nsid.NSID
-
-
-
testStrings[
-
-
"rkey"
-
-
] = """3jui7kd54zh2y
-
self
-
example.com
-
-
~1.2-3_
-
-
dHJ1ZQ
-
pre:fix
-
-
_
-
-
alpha/beta
-
.
-
..
-
-
#extra
-
-
@handle
-
-
any space
-
-
any+space
-
-
number[3]
-
-
number(3)
-
-
"quote"
-
-
dHJ1ZQ=="""
-
-
testMethods["rkey"] = rKey.RKey
-
-
-
testStrings[
-
"uri"
-
-
] = """at://foo.com/com.example.foo/123
-
-
at://foo.com/example/123
-
-
at://computer
-
-
at://example.com:3000
-
-
at://foo.com/
-
-
at://user:pass@foo.com"""
-
-
testMethods["uri"] = uri.URI
-
-
-
for item in testMethods:
-
-
print(f"START TEST {item}")
-
-
for value in testStrings[item].splitlines():
-
-
print(f"Value: {value}")
-
-
try:
-
-
print(f"str(): {str(testMethods[item](value))}")
-
-
except Exception as e:
-
-
print(f"ร— {e}")
-
+41
src/atpasser/model/typed.py
···
+
from typing import Any
+
from pydantic import field_serializer
+
from .base import DataModel
+
+
class TypedDataModel(DataModel):
+
"""
+
Model for AT Protocol data with type information.
+
+
Includes support for $type field that specifies Lexicon schema.
+
"""
+
+
type: str | None = None
+
"""Lexicon schema type identifier"""
+
+
def __init__(self, **data: Any) -> None:
+
"""
+
Initialize typed data model with automatic $type handling.
+
+
Args:
+
**data: Data including optional $type field
+
"""
+
# Extract $type if present
+
dataType = data.pop("$type", None)
+
if dataType:
+
data["type"] = dataType
+
super().__init__(**data)
+
+
@field_serializer("type")
+
def serializeType(self, v: str | None) -> dict[str, str] | None:
+
"""
+
Serialize type field to $type object.
+
+
Args:
+
v: Type value to serialize
+
+
Returns:
+
$type object if type is not None
+
"""
+
if v is not None:
+
return {"$type": v}
+
return None
+36
src/atpasser/model/exceptions.py
···
+
class AtprotoModelError(Exception):
+
"""Base exception for all AT Protocol model errors"""
+
pass
+
+
class ValidationError(AtprotoModelError):
+
"""Raised when data validation fails"""
+
def __init__(self, field: str, message: str):
+
self.field = field
+
self.message = message
+
super().__init__(f"Validation error for field '{field}': {message}")
+
+
class SerializationError(AtprotoModelError):
+
"""Raised when data serialization fails"""
+
def __init__(self, field: str, message: str):
+
self.field = field
+
self.message = message
+
super().__init__(f"Serialization error for field '{field}': {message}")
+
+
class DeserializationError(AtprotoModelError):
+
"""Raised when data deserialization fails"""
+
def __init__(self, field: str, message: str):
+
self.field = field
+
self.message = message
+
super().__init__(f"Deserialization error for field '{field}': {message}")
+
+
class InvalidCIDError(AtprotoModelError):
+
"""Raised when CID validation fails"""
+
pass
+
+
class InvalidBlobError(AtprotoModelError):
+
"""Raised when blob validation fails"""
+
pass
+
+
class TypeMismatchError(AtprotoModelError):
+
"""Raised when type validation fails"""
+
pass
+3
src/atpasser/model/__init__.py
···
ProcedureModel,
SubscriptionModel
)
+
from .converter import LexiconConverter
__all__ = [
"DataModel",
···
"QueryModel",
"ProcedureModel",
"SubscriptionModel",
+
# Converter
+
"LexiconConverter",
# Exceptions
"AtprotoModelError",
"ValidationError",
-5
src/atpasser/model/base.py
···
import base64
-
import re
-
from datetime import datetime
from typing import Any
-
from collections.abc import Mapping
from cid.cid import CIDv1, make_cid
from pydantic import BaseModel, field_serializer, field_validator, ConfigDict
-
from pydantic.fields import FieldInfo
from .exceptions import (
-
ValidationError,
SerializationError,
DeserializationError,
InvalidCIDError
+5 -5
src/atpasser/model/blob.py
···
from typing import Any
from pydantic import field_validator, ConfigDict
from .base import DataModel
-
from .exceptions import ValidationError, InvalidBlobError
+
from .exceptions import ValidationError
class BlobModel(DataModel):
"""
···
Validated size
Raises:
-
ValueError: If size is not positive
+
ValidationError: If size is not positive
"""
if v <= 0:
-
raise ValueError("Blob size must be positive and non-zero")
+
raise ValidationError(field="size", message="must be positive and non-zero")
return v
@field_validator("mimeType")
···
Validated MIME type
Raises:
-
ValueError: If MIME type is empty
+
ValidationError: If MIME type is empty
"""
if not v:
-
raise ValueError("MIME type cannot be empty")
+
raise ValidationError(field="mimeType", message="cannot be empty")
return v
+1
pyproject.toml
···
[tool.poetry.group.dev.dependencies]
pytest = "^8.2.0"
pytest-cov = "^5.0.0"
+
pygithub = "^2.8.1"
[build-system]
+7
src/atpasser/model/converter.py
···
UnknownModel, RecordModel, QueryModel,
ProcedureModel, SubscriptionModel
)
+
from .types.binary import BytesModel, CidLinkModel
+
from .blob import BlobModel
class LexiconConverter:
"""
···
"integer": IntegerModel,
"string": StringModel,
+
# Binary types
+
"bytes": BytesModel,
+
"cid-link": CidLinkModel,
+
"blob": BlobModel,
+
# Complex types
"array": ArrayModel,
"object": ObjectModel,
+214
src/atpasser/model/types/complex.py
···
+
from typing import Any
+
from pydantic import field_validator
+
from ..base import DataModel
+
+
class ArrayModel(DataModel):
+
"""
+
Model for AT Protocol array type.
+
+
Represents an array of elements with support for item schema definition,
+
minimum/maximum length constraints as specified in Lexicon.
+
"""
+
+
items: Any
+
"""Schema definition for array elements"""
+
+
minLength: int | None = None
+
"""Minimum number of elements"""
+
+
maxLength: int | None = None
+
"""Maximum number of elements"""
+
+
value: list[Any]
+
"""Array values"""
+
+
def __init__(self, **data: Any) -> None:
+
"""
+
Initialize array model with validation.
+
+
Args:
+
**data: Input data containing array values
+
+
Raises:
+
ValueError: If array violates constraints
+
"""
+
super().__init__(**data)
+
+
@field_validator("value", mode="before")
+
def validate_array(cls, v: Any) -> list[Any]:
+
"""
+
Validate array structure and elements.
+
+
Args:
+
v: Value to validate
+
+
Returns:
+
Validated array
+
+
Raises:
+
ValueError: If array violates constraints
+
"""
+
if not isinstance(v, list):
+
raise ValueError("Value must be an array")
+
+
# Validate length constraints
+
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")
+
+
return v
+
+
class ObjectModel(DataModel):
+
"""
+
Model for AT Protocol object type.
+
+
Represents a generic object schema with properties definitions,
+
required fields and nullable fields as specified in Lexicon.
+
"""
+
+
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"""
+
+
value: dict[str, Any]
+
"""Object property values"""
+
+
def __init__(self, **data: Any) -> None:
+
"""
+
Initialize object model with validation.
+
+
Args:
+
**data: Input data containing object properties
+
+
Raises:
+
ValueError: If object violates constraints
+
"""
+
super().__init__(**data)
+
+
@field_validator("value", mode="before")
+
def validate_object(cls, v: Any) -> dict[str, Any]:
+
"""
+
Validate object structure and properties.
+
+
Args:
+
v: Value to validate
+
+
Returns:
+
Validated object
+
+
Raises:
+
ValueError: If object violates constraints
+
"""
+
if not isinstance(v, dict):
+
raise ValueError("Value must be an object")
+
+
# Validate required fields
+
if cls.required:
+
for field in cls.required:
+
if field not in v:
+
raise ValueError(f"Missing required field: {field}")
+
+
# Validate nullable fields
+
if cls.nullable:
+
for field, value in v.items():
+
if field not in cls.nullable and value is None:
+
raise ValueError(f"Field {field} cannot be null")
+
+
return v
+
+
class ParamsModel(DataModel):
+
"""
+
Model for AT Protocol params type.
+
+
Specialized for HTTP query parameters with support for boolean,
+
integer, string and unknown types as specified in Lexicon.
+
"""
+
+
required: list[str] | None = None
+
"""List of required parameter names"""
+
+
properties: dict[str, Any]
+
"""Map of parameter names to their schema definitions"""
+
+
value: dict[str, Any]
+
"""Parameter values
+
+
Supported types:
+
- boolean
+
- integer
+
- string
+
- array (of boolean/integer/string/unknown)
+
- unknown (object)
+
"""
+
+
def __init__(self, **data: Any) -> None:
+
"""
+
Initialize params model with validation.
+
+
Args:
+
**data: Input data containing parameter values
+
+
Raises:
+
ValueError: If parameters violate constraints
+
"""
+
super().__init__(**data)
+
+
@field_validator("value", mode="before")
+
def validate_params(cls, v: Any) -> dict[str, Any]:
+
"""
+
Validate parameters structure and values.
+
+
Args:
+
v: Value to validate
+
+
Returns:
+
Validated parameters
+
+
Raises:
+
ValueError: If parameters violate constraints
+
"""
+
if not isinstance(v, dict):
+
raise ValueError("Value must be a dictionary of parameters")
+
+
# Validate required parameters
+
if cls.required:
+
for param in cls.required:
+
if param not in v:
+
raise ValueError(f"Missing required parameter: {param}")
+
+
# Validate parameter types
+
for param, value in v.items():
+
if param in cls.properties:
+
prop_type = cls.properties[param].get("type")
+
if prop_type == "boolean" and not isinstance(value, bool):
+
raise ValueError(f"Parameter {param} must be boolean")
+
elif prop_type == "integer" and not isinstance(value, int):
+
raise ValueError(f"Parameter {param} must be integer")
+
elif prop_type == "string" and not isinstance(value, str):
+
raise ValueError(f"Parameter {param} must be string")
+
elif prop_type == "array":
+
if not isinstance(value, list):
+
raise ValueError(f"Parameter {param} must be array")
+
# Validate array items if schema is specified
+
if "items" in cls.properties[param]:
+
item_type = cls.properties[param]["items"].get("type")
+
for item in value:
+
if item_type == "boolean" and not isinstance(item, bool):
+
raise ValueError(f"Array item in {param} must be boolean")
+
elif item_type == "integer" and not isinstance(item, int):
+
raise ValueError(f"Array item in {param} must be integer")
+
elif item_type == "string" and not isinstance(item, str):
+
raise ValueError(f"Array item in {param} must be string")
+
elif item_type == "unknown" and not isinstance(item, dict):
+
raise ValueError(f"Array item in {param} must be object")
+
elif prop_type == "unknown" and not isinstance(value, dict):
+
raise ValueError(f"Parameter {param} must be object")
+
+
return v
+172
src/atpasser/model/types/primitive.py
···
+
from typing import Any
+
from pydantic import field_validator
+
from ..base import DataModel
+
+
class NullModel(DataModel):
+
"""
+
Model for AT Protocol null type.
+
+
Represents a null value in AT Protocol data model. This model ensures proper
+
serialization and validation of null values according to Lexicon specification.
+
"""
+
+
value: None = None
+
"""Always None for null type"""
+
+
def __init__(self, **data: Any) -> None:
+
"""
+
Initialize null model with validation.
+
+
Args:
+
**data: Input data (must be empty or contain only None values)
+
+
Raises:
+
ValueError: If non-null value is provided
+
"""
+
if data and any(v is not None for v in data.values()):
+
raise ValueError("NullModel only accepts None values")
+
super().__init__(**data)
+
+
@field_validator("*", mode="before")
+
def validate_null(cls, v: Any) -> None:
+
"""
+
Validate that value is null.
+
+
Args:
+
v: Value to validate
+
+
Returns:
+
None if validation succeeds
+
+
Raises:
+
ValueError: If value is not null
+
"""
+
if v is not None:
+
raise ValueError("NullModel only accepts None values")
+
return None
+
+
class BooleanModel(DataModel):
+
"""
+
Model for AT Protocol boolean type.
+
+
Represents a boolean value in AT Protocol data model with support for
+
default values and constants as specified in Lexicon.
+
"""
+
+
value: bool
+
"""Boolean value"""
+
+
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.
+
+
Args:
+
**data: Input data containing boolean value
+
+
Raises:
+
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 validate_boolean(cls, v: Any) -> bool:
+
"""
+
Validate and convert input to boolean.
+
+
Args:
+
v: Value to validate
+
+
Returns:
+
Validated boolean value
+
+
Raises:
+
ValueError: If value cannot be converted to boolean
+
"""
+
if isinstance(v, bool):
+
return v
+
if isinstance(v, str):
+
if v.lower() in ("true", "1"):
+
return True
+
if v.lower() in ("false", "0"):
+
return False
+
raise ValueError("Value must be a boolean")
+
+
class IntegerModel(DataModel):
+
"""
+
Model for AT Protocol integer type.
+
+
Represents a signed integer number with support for minimum/maximum values,
+
enumeration sets, default values and constraints as specified in Lexicon.
+
"""
+
+
value: int
+
"""Integer value"""
+
+
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.
+
+
Args:
+
**data: Input data containing integer value
+
+
Raises:
+
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 validate_integer(cls, v: Any) -> int:
+
"""
+
Validate and convert input to integer.
+
+
Args:
+
v: Value to validate
+
+
Returns:
+
Validated integer value
+
+
Raises:
+
ValueError: If value violates constraints
+
"""
+
if not isinstance(v, int):
+
try:
+
v = int(v)
+
except (TypeError, ValueError):
+
raise ValueError("Value must be an integer")
+
+
# Validate against instance attributes
+
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}")
+
+
return v
+131
src/atpasser/model/types/reference.py
···
+
from typing import Any
+
from pydantic import field_validator
+
from ..base import DataModel
+
+
class TokenModel(DataModel):
+
"""
+
Model for AT Protocol token type.
+
+
Represents empty data values which exist only to be referenced by name.
+
Tokens encode as string data with the string being the fully-qualified
+
reference to the token itself (NSID followed by optional fragment).
+
"""
+
+
name: str
+
"""Token name/identifier"""
+
+
description: str | None = None
+
"""Description clarifying the meaning of the token"""
+
+
def __init__(self, **data: Any) -> None:
+
"""
+
Initialize token model.
+
+
Args:
+
**data: Input data containing token name
+
"""
+
super().__init__(**data)
+
+
@field_validator("name")
+
def validate_name(cls, v: str) -> str:
+
"""
+
Validate token name format.
+
+
Args:
+
v: Name to validate
+
+
Returns:
+
Validated name
+
+
Raises:
+
ValueError: If name contains whitespace
+
"""
+
if any(c.isspace() for c in v):
+
raise ValueError("Token name must not contain whitespace")
+
return v
+
+
class RefModel(DataModel):
+
"""
+
Model for AT Protocol ref type.
+
+
Represents a reference to another schema definition, either globally
+
(using NSID) or locally (using #-delimited name).
+
"""
+
+
ref: str
+
"""Reference to schema definition (NSID or #name)"""
+
+
description: str | None = None
+
"""Description of the reference"""
+
+
def __init__(self, **data: Any) -> None:
+
"""
+
Initialize reference model.
+
+
Args:
+
**data: Input data containing reference
+
"""
+
super().__init__(**data)
+
+
@field_validator("ref")
+
def validate_ref(cls, v: str) -> str:
+
"""
+
Validate reference format.
+
+
Args:
+
v: Reference to validate
+
+
Returns:
+
Validated reference
+
+
Raises:
+
ValueError: If reference is empty or invalid
+
"""
+
if not v:
+
raise ValueError("Reference cannot be empty")
+
return v
+
+
class UnionModel(DataModel):
+
"""
+
Model for AT Protocol union type.
+
+
Represents that multiple possible types could be present at a location.
+
The references follow the same syntax as `ref`, allowing references to
+
both global or local schema definitions.
+
"""
+
+
refs: list[str]
+
"""References to schema definitions"""
+
+
closed: bool = False
+
"""Indicates if union is open (can be extended) or closed"""
+
+
description: str | None = None
+
"""Description of the union"""
+
+
def __init__(self, **data: Any) -> None:
+
"""
+
Initialize union model.
+
+
Args:
+
**data: Input data containing union references
+
"""
+
super().__init__(**data)
+
+
@field_validator("refs")
+
def validate_refs(cls, v: list[str]) -> list[str]:
+
"""
+
Validate union references.
+
+
Args:
+
v: References to validate
+
+
Returns:
+
Validated references
+
+
Raises:
+
ValueError: If references list is empty for closed union
+
"""
+
if cls.closed and not v:
+
raise ValueError("Closed union must have at least one reference")
+
return v
+323
src/atpasser/model/types/special.py
···
+
from typing import Any
+
from pydantic import field_validator
+
from ..base import DataModel
+
+
class UnknownModel(DataModel):
+
"""
+
Model for AT Protocol unknown type.
+
+
Indicates that any data object could appear at this location,
+
with no specific validation. The top-level data must be an object.
+
"""
+
+
description: str | None = None
+
"""Description of the unknown type usage"""
+
+
def __init__(self, **data: Any) -> None:
+
"""
+
Initialize unknown model.
+
+
Args:
+
**data: Input data containing unknown object
+
"""
+
super().__init__(**data)
+
+
@field_validator("*", mode="before")
+
def validate_unknown(cls, v: Any) -> Any:
+
"""
+
Validate unknown data is an object.
+
+
Args:
+
v: Value to validate
+
+
Returns:
+
Validated value
+
+
Raises:
+
ValueError: If value is not an object
+
"""
+
if not isinstance(v, dict):
+
raise ValueError("Unknown type must be an object")
+
return v
+
+
class RecordModel(DataModel):
+
"""
+
Model for AT Protocol record type.
+
+
Describes an object that can be stored in a repository record.
+
Records must include a $type field indicating their schema.
+
"""
+
+
key: str
+
"""Specifies the Record Key type"""
+
+
record: dict[str, Any]
+
"""Schema definition with type 'object'"""
+
+
type: str
+
"""Lexicon schema type identifier"""
+
+
def __init__(self, **data: Any) -> None:
+
"""
+
Initialize record model with validation.
+
+
Args:
+
**data: Input data containing record values
+
+
Raises:
+
ValueError: If record is missing required fields
+
"""
+
# Extract $type if present
+
data_type = data.pop("$type", None)
+
if data_type:
+
data["type"] = data_type
+
super().__init__(**data)
+
+
@field_validator("type")
+
def validate_type(cls, v: str) -> str:
+
"""
+
Validate record type field.
+
+
Args:
+
v: Type value to validate
+
+
Returns:
+
Validated type
+
+
Raises:
+
ValueError: If type is empty
+
"""
+
if not v:
+
raise ValueError("Record must have a type")
+
return v
+
+
@field_validator("record", mode="before")
+
def validate_record(cls, v: Any) -> dict[str, Any]:
+
"""
+
Validate record structure.
+
+
Args:
+
v: Record value to validate
+
+
Returns:
+
Validated record
+
+
Raises:
+
ValueError: If record is not an object
+
"""
+
if not isinstance(v, dict):
+
raise ValueError("Record must be an object")
+
return v
+
+
class QueryModel(DataModel):
+
"""
+
Model for AT Protocol query type.
+
+
Describes an XRPC Query endpoint (HTTP GET) with support for
+
parameters, output schema and error responses.
+
"""
+
+
parameters: dict[str, Any] | None = None
+
"""HTTP query parameters schema"""
+
+
output: dict[str, Any] | None = None
+
"""HTTP response body schema"""
+
+
errors: list[dict[str, str]] | None = None
+
"""Possible error responses"""
+
+
def __init__(self, **data: Any) -> None:
+
"""
+
Initialize query model with validation.
+
+
Args:
+
**data: Input data containing query definition
+
"""
+
super().__init__(**data)
+
+
@field_validator("output")
+
def validate_output(cls, v: dict[str, Any] | None) -> dict[str, Any] | None:
+
"""
+
Validate output schema.
+
+
Args:
+
v: Output schema to validate
+
+
Returns:
+
Validated output schema
+
+
Raises:
+
ValueError: If output schema is invalid
+
"""
+
if v and "encoding" not in v:
+
raise ValueError("Output must specify encoding")
+
return v
+
+
@field_validator("errors")
+
def validate_errors(cls, v: list[dict[str, str]] | None) -> list[dict[str, str]] | None:
+
"""
+
Validate error definitions.
+
+
Args:
+
v: Error definitions to validate
+
+
Returns:
+
Validated error definitions
+
+
Raises:
+
ValueError: If any error definition is invalid
+
"""
+
if v:
+
for error in v:
+
if "name" not in error:
+
raise ValueError("Error must have a name")
+
return v
+
+
class ProcedureModel(DataModel):
+
"""
+
Model for AT Protocol procedure type.
+
+
Describes an XRPC Procedure endpoint (HTTP POST) with support for
+
parameters, input/output schemas and error responses.
+
"""
+
+
parameters: dict[str, Any] | None = None
+
"""HTTP query parameters schema"""
+
+
input: dict[str, Any] | None = None
+
"""HTTP request body schema"""
+
+
output: dict[str, Any] | None = None
+
"""HTTP response body schema"""
+
+
errors: list[dict[str, str]] | None = None
+
"""Possible error responses"""
+
+
def __init__(self, **data: Any) -> None:
+
"""
+
Initialize procedure model with validation.
+
+
Args:
+
**data: Input data containing procedure definition
+
"""
+
super().__init__(**data)
+
+
@field_validator("input")
+
def validate_input(cls, v: dict[str, Any] | None) -> dict[str, Any] | None:
+
"""
+
Validate input schema.
+
+
Args:
+
v: Input schema to validate
+
+
Returns:
+
Validated input schema
+
+
Raises:
+
ValueError: If input schema is invalid
+
"""
+
if v and "encoding" not in v:
+
raise ValueError("Input must specify encoding")
+
return v
+
+
@field_validator("output")
+
def validate_output(cls, v: dict[str, Any] | None) -> dict[str, Any] | None:
+
"""
+
Validate output schema.
+
+
Args:
+
v: Output schema to validate
+
+
Returns:
+
Validated output schema
+
+
Raises:
+
ValueError: If output schema is invalid
+
"""
+
if v and "encoding" not in v:
+
raise ValueError("Output must specify encoding")
+
return v
+
+
@field_validator("errors")
+
def validate_errors(cls, v: list[dict[str, str]] | None) -> list[dict[str, str]] | None:
+
"""
+
Validate error definitions.
+
+
Args:
+
v: Error definitions to validate
+
+
Returns:
+
Validated error definitions
+
+
Raises:
+
ValueError: If any error definition is invalid
+
"""
+
if v:
+
for error in v:
+
if "name" not in error:
+
raise ValueError("Error must have a name")
+
return v
+
+
class SubscriptionModel(DataModel):
+
"""
+
Model for AT Protocol subscription type.
+
+
Describes an Event Stream (WebSocket) with support for parameters,
+
message schemas and error responses.
+
"""
+
+
parameters: dict[str, Any] | None = None
+
"""HTTP query parameters schema"""
+
+
message: dict[str, Any] | None = None
+
"""Specifies what messages can be"""
+
+
errors: list[dict[str, str]] | None = None
+
"""Possible error responses"""
+
+
def __init__(self, **data: Any) -> None:
+
"""
+
Initialize subscription model with validation.
+
+
Args:
+
**data: Input data containing subscription definition
+
"""
+
super().__init__(**data)
+
+
@field_validator("message")
+
def validate_message(cls, v: dict[str, Any] | None) -> dict[str, Any] | None:
+
"""
+
Validate message schema.
+
+
Args:
+
v: Message schema to validate
+
+
Returns:
+
Validated message schema
+
+
Raises:
+
ValueError: If message schema is invalid
+
"""
+
if v and "schema" not in v:
+
raise ValueError("Message must specify schema")
+
return v
+
+
@field_validator("errors")
+
def validate_errors(cls, v: list[dict[str, str]] | None) -> list[dict[str, str]] | None:
+
"""
+
Validate error definitions.
+
+
Args:
+
v: Error definitions to validate
+
+
Returns:
+
Validated error definitions
+
+
Raises:
+
ValueError: If any error definition is invalid
+
"""
+
if v:
+
for error in v:
+
if "name" not in error:
+
raise ValueError("Error must have a name")
+
return v
+249
src/atpasser/model/types/string.py
···
+
from typing import Any
+
import re
+
from datetime import datetime
+
from pydantic import field_validator
+
from ..base import DataModel
+
+
class StringModel(DataModel):
+
"""
+
Model for AT Protocol string type.
+
+
Represents a Unicode string with support for format restrictions, length limits,
+
known values, enumeration sets, default values and constants as specified in Lexicon.
+
"""
+
+
value: str
+
"""String value"""
+
+
format: str | None = None
+
"""String format restriction (e.g. 'datetime', 'uri')"""
+
+
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 (not enforced)"""
+
+
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.
+
+
Args:
+
**data: Input data containing string value
+
+
Raises:
+
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 validate_string(cls, v: Any) -> str:
+
"""
+
Validate and convert input to string.
+
+
Args:
+
v: Value to validate
+
+
Returns:
+
Validated string value
+
+
Raises:
+
ValueError: If value violates constraints
+
"""
+
if not isinstance(v, str):
+
v = str(v)
+
+
# Validate length constraints
+
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")
+
+
# Validate enum
+
if cls.enum and v not in cls.enum:
+
raise ValueError(f"Value must be one of {cls.enum}")
+
+
# Validate format if specified
+
if cls.format:
+
if cls.format == "datetime":
+
cls._validate_datetime(v)
+
elif cls.format == "uri":
+
cls._validate_uri(v)
+
elif cls.format == "did":
+
cls._validate_did(v)
+
elif cls.format == "handle":
+
cls._validate_handle(v)
+
elif cls.format == "at-identifier":
+
cls._validate_at_identifier(v)
+
elif cls.format == "at-uri":
+
cls._validate_at_uri(v)
+
elif cls.format == "cid":
+
cls._validate_cid(v)
+
elif cls.format == "nsid":
+
cls._validate_nsid(v)
+
elif cls.format == "tid":
+
cls._validate_tid(v)
+
elif cls.format == "record-key":
+
cls._validate_record_key(v)
+
elif cls.format == "language":
+
cls._validate_language(v)
+
+
return v
+
+
@classmethod
+
def _validate_datetime(cls, v: str) -> None:
+
"""Validate RFC 3339 datetime format"""
+
try:
+
datetime.fromisoformat(v.replace("Z", "+00:00"))
+
except ValueError:
+
raise ValueError("Invalid datetime format, must be RFC 3339")
+
+
@classmethod
+
def _validate_uri(cls, v: str) -> None:
+
"""Validate URI format"""
+
if len(v) > 8192: # 8KB max
+
raise ValueError("URI too long, max 8KB")
+
if not re.match(r"^[a-zA-Z][a-zA-Z0-9+.-]*:.+", v):
+
raise ValueError("Invalid URI format")
+
+
@classmethod
+
def _validate_did(cls, v: str) -> None:
+
"""Validate DID format"""
+
if len(v) > 2048:
+
raise ValueError("DID too long, max 2048 chars")
+
if not re.match(r"^did:[a-z]+:[a-zA-Z0-9._:%-]*[a-zA-Z0-9._-]$", v):
+
raise ValueError("Invalid URI format")
+
+
@classmethod
+
def _validate_handle(cls, v: str) -> None:
+
"""Validate handle format"""
+
if not re.match(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])?$", v):
+
raise ValueError("Handle contains invalid characters")
+
if len(v) > 253:
+
raise ValueError("Handle too long, max 253 chars")
+
+
@classmethod
+
def _validate_at_identifier(cls, v: str) -> None:
+
"""Validate at-identifier format (DID or handle)"""
+
try:
+
if v.startswith("did:"):
+
cls._validate_did(v)
+
else:
+
cls._validate_handle(v)
+
except ValueError as e:
+
raise ValueError(f"Invalid at-identifier: {e}")
+
+
@classmethod
+
def _validate_at_uri(cls, v: str) -> None:
+
"""
+
Validate AT-URI format according to AT Protocol specification.
+
+
Args:
+
v: AT-URI string to validate
+
+
Raises:
+
ValueError: If URI violates any of these rules:
+
- Must start with 'at://'
+
- Max length 8KB
+
- No trailing slash
+
- Authority must be valid DID or handle
+
- Path segments must follow NSID/RKEY rules if present
+
"""
+
if not v.startswith("at://"):
+
raise ValueError("AT-URI must start with 'at://'")
+
if len(v) > 8192: # 8KB
+
raise ValueError("AT-URI too long, max 8KB")
+
if v.endswith('/'):
+
raise ValueError("AT-URI cannot have trailing slash")
+
+
# Split into parts
+
parts = v[5:].split('/') # Skip 'at://'
+
authority = parts[0]
+
+
# Validate authority (DID or handle)
+
if not authority:
+
raise ValueError("AT-URI must have authority")
+
+
if authority.startswith('did:'):
+
# Basic DID format check - actual DID validation is done elsewhere
+
if len(authority) > 2048:
+
raise ValueError("DID too long")
+
if ':' not in authority[4:]:
+
raise ValueError("Invalid DID format")
+
else:
+
# Handle validation
+
if not re.match(r'^[a-z0-9.-]+$', authority):
+
raise ValueError("Invalid handle characters")
+
if len(authority) > 253:
+
raise ValueError("Handle too long")
+
+
# Validate path segments if present
+
if len(parts) > 1:
+
if len(parts) > 3:
+
raise ValueError("AT-URI path too deep")
+
+
collection = parts[1]
+
if not re.match(r'^[a-zA-Z0-9.-]+$', collection):
+
raise ValueError("Invalid collection NSID")
+
+
if len(parts) > 2:
+
rkey = parts[2]
+
if not rkey:
+
raise ValueError("Record key cannot be empty")
+
if not re.match(r'^[a-zA-Z0-9._:%-~]+$', rkey):
+
raise ValueError("Invalid record key characters")
+
+
@classmethod
+
def _validate_cid(cls, v: str) -> None:
+
"""Validate CID string format"""
+
if len(v) > 100:
+
raise ValueError("CID too long, max 100 chars")
+
if not re.match(r"^[a-zA-Z0-9]+$", v):
+
raise ValueError("CID contains invalid characters")
+
+
@classmethod
+
def _validate_nsid(cls, v: str) -> None:
+
"""Validate NSID format"""
+
if len(v) > 317:
+
raise ValueError("NSID too long, max 317 chars")
+
if not re.match(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})?)$", v):
+
raise ValueError("NSID contains invalid characters")
+
+
@classmethod
+
def _validate_tid(cls, v: str) -> None:
+
"""Validate TID format"""
+
if len(v) > 13:
+
raise ValueError("TID too long, max 13 chars")
+
if not re.match(r"^[234567abcdefghij][234567abcdefghijklmnopqrstuvwxyz]{12}$", v):
+
raise ValueError("TID contains invalid characters")
+
+
@classmethod
+
def _validate_record_key(cls, v: str) -> None:
+
"""Validate record-key format"""
+
if len(v) > 512:
+
raise ValueError("Record key too long, max 512 chars")
+
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")
+
+
@classmethod
+
def _validate_language(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")