tessl/pypi-atproto

Comprehensive Python SDK for the AT Protocol, providing client interfaces, authentication, and real-time streaming for decentralized social networks.

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Core Functionality

Name: tessl/pypi-atproto
Author: tessl

Fundamental AT Protocol data structures and utilities including Content Identifiers (CID), Content Addressable Archives (CAR), AT Protocol URIs, Namespaced IDs (NSID), and DID document handling. These components provide the foundation for content addressing, resource identification, and decentralized identity management in AT Protocol.

Capabilities

Content Identifiers (CID)

Content Identifiers provide cryptographic addressing for IPLD data structures, enabling content-based addressing and verification.

class CID:
    """
    Content Identifier for IPLD data structures.
    
    Attributes:
        version (int): CID version (0 or 1)
        codec (int): Content type codec identifier
        hash (Multihash): Cryptographic hash of the content
    """
    version: int
    codec: int
    hash: Multihash
    
    @classmethod
    def decode(cls, value: Union[str, bytes]) -> 'CID':
        """
        Decode CID from string or bytes.
        
        Args:
            value (Union[str, bytes]): CID in string or binary format
            
        Returns:
            CID: Decoded CID object
            
        Raises:
            ValueError: If CID format is invalid
        """
    
    def encode(self) -> str:
        """
        Encode CID to string representation.
        
        Returns:
            str: Base58 encoded CID string
        """
    
    def __str__(self) -> str:
        """String representation of the CID."""
    
    def __eq__(self, other) -> bool:
        """Check CID equality."""

# Type alias for CID values
CIDType = Union[str, CID]

Usage example:

from atproto import CID

# Decode CID from string
cid_str = "bafybeigdyrzt5n4k53bvolvbtng2jrcaykswh75f3cw5vyz5l2qx2d6vby"
cid = CID.decode(cid_str)

print(f"Version: {cid.version}")
print(f"Codec: {cid.codec}")
print(f"Encoded: {cid.encode()}")

# CIDs support equality comparison
cid2 = CID.decode(cid_str)
assert cid == cid2

Content Addressable Archives (CAR)

CAR files package IPLD blocks with their CIDs for efficient storage and transmission of related content.

class CAR:
    """
    Content Addressable aRchives (CAR) file handler.
    
    Attributes:
        root (CID): Root CID of the CAR file
        blocks (Dict[CID, bytes]): Dictionary mapping CIDs to block data
    """
    root: CID
    blocks: Dict[CID, bytes]
    
    def __init__(self, root: CID, blocks: Dict[CID, bytes]):
        """
        Initialize CAR with root CID and blocks.
        
        Args:
            root (CID): Root CID
            blocks (Dict[CID, bytes]): Block data mapped by CID
        """
    
    @classmethod
    def from_bytes(cls, data: bytes) -> 'CAR':
        """
        Decode CAR from bytes.
        
        Args:
            data (bytes): CAR file data
            
        Returns:
            CAR: Decoded CAR object
            
        Raises:
            ValueError: If CAR format is invalid
        """
    
    def to_bytes(self) -> bytes:
        """
        Encode CAR to bytes.
        
        Returns:
            bytes: Encoded CAR file data
        """
    
    def get_block(self, cid: CID) -> Optional[bytes]:
        """
        Get block data by CID.
        
        Args:
            cid (CID): Content identifier
            
        Returns:
            Optional[bytes]: Block data or None if not found
        """

Usage example:

from atproto import CAR, CID

# Load CAR file
with open('example.car', 'rb') as f:
    car_data = f.read()

# Parse CAR file
car = CAR.from_bytes(car_data)

print(f"Root CID: {car.root}")
print(f"Number of blocks: {len(car.blocks)}")

# Access specific block
block_data = car.get_block(car.root)
if block_data:
    print(f"Root block size: {len(block_data)} bytes")

# Create new CAR file
new_blocks = {cid: data for cid, data in car.blocks.items()}
new_car = CAR(car.root, new_blocks)
car_bytes = new_car.to_bytes()

AT Protocol URIs

AT Protocol URIs provide a standardized way to address resources in the AT Protocol network using the at:// scheme.

class AtUri:
    """
    ATProtocol URI scheme handler for at:// URIs.
    
    Handles both DID and handle-based URIs with path and query parameters.
    """
    def __init__(self, 
                 host: str, 
                 pathname: str = '', 
                 hash_: str = '', 
                 search_params: Optional[List[Tuple[str, Any]]] = None):
        """
        Initialize AT Protocol URI.
        
        Args:
            host (str): DID or handle
            pathname (str): Resource path
            hash_ (str): Fragment identifier
            search_params (List[Tuple[str, Any]], optional): Query parameters
        """
    
    @classmethod
    def from_str(cls, uri: str) -> 'AtUri':
        """
        Parse AT Protocol URI from string.
        
        Args:
            uri (str): URI string (e.g., "at://alice.bsky.social/app.bsky.feed.post/123")
            
        Returns:
            AtUri: Parsed URI object
            
        Raises:
            ValueError: If URI format is invalid
        """
    
    @property
    def host(self) -> str:
        """Get the host (DID or handle)."""
    
    @property
    def pathname(self) -> str:
        """Get the path component."""
    
    @property
    def collection(self) -> Optional[str]:
        """Extract collection name from path."""
    
    @property
    def rkey(self) -> Optional[str]:
        """Extract record key from path."""
    
    def __str__(self) -> str:
        """String representation of the URI."""

Usage example:

from atproto import AtUri

# Parse URI from string
uri_str = "at://alice.bsky.social/app.bsky.feed.post/3k2l4m5n6"
uri = AtUri.from_str(uri_str)

print(f"Host: {uri.host}")
print(f"Collection: {uri.collection}")
print(f"Record key: {uri.rkey}")

# Create URI programmatically
post_uri = AtUri(
    host="did:plc:alice123", 
    pathname="/app.bsky.feed.post/abc123"
)
print(f"URI: {post_uri}")

# Extract components
if uri.collection == "app.bsky.feed.post":
    print(f"This is a post with rkey: {uri.rkey}")

Namespaced IDs (NSID)

NSIDs provide a hierarchical naming system for AT Protocol lexicons and schema definitions.

class NSID:
    """
    NameSpaced IDs for ATProto lexicons.
    
    Attributes:
        segments (List[str]): Dot-separated segments of the NSID
    """
    segments: List[str]
    
    @classmethod
    def from_str(cls, nsid: str) -> 'NSID':
        """
        Parse NSID from string.
        
        Args:
            nsid (str): NSID string (e.g., "app.bsky.feed.post")
            
        Returns:
            NSID: Parsed NSID object
            
        Raises:
            ValueError: If NSID format is invalid
        """
    
    @property
    def authority(self) -> str:
        """Get the authority segment (first part)."""
    
    @property 
    def name(self) -> str:
        """Get the name segment (last part)."""
    
    def __str__(self) -> str:
        """String representation of the NSID."""

def validate_nsid(nsid: str) -> bool:
    """
    Validate NSID string format.
    
    Args:
        nsid (str): NSID string to validate
        
    Returns:
        bool: True if valid NSID format
    """

Usage example:

from atproto import NSID, validate_nsid

# Parse NSID
nsid_str = "app.bsky.feed.post"
nsid = NSID.from_str(nsid_str)

print(f"Authority: {nsid.authority}")  # "app"
print(f"Name: {nsid.name}")          # "post"
print(f"Segments: {nsid.segments}")   # ["app", "bsky", "feed", "post"]

# Validate NSID format
valid_nsids = [
    "app.bsky.feed.post",
    "com.example.my.record",
    "io.github.user.schema"
]

for nsid_str in valid_nsids:
    if validate_nsid(nsid_str):
        print(f"Valid NSID: {nsid_str}")

DID Documents

DID documents provide decentralized identity information including service endpoints and cryptographic keys.

class DidDocument:
    """
    Decentralized Identifier document representation.
    
    Contains identity information, service endpoints, and verification methods.
    """
    @classmethod
    def from_dict(cls, data: dict) -> 'DidDocument':
        """
        Create DID document from dictionary.
        
        Args:
            data (dict): DID document data
            
        Returns:
            DidDocument: Parsed DID document
            
        Raises:
            ValueError: If document format is invalid
        """
    
    def get_pds_endpoint(self) -> Optional[str]:
        """
        Extract Personal Data Server endpoint from document.
        
        Returns:
            Optional[str]: PDS endpoint URL or None if not found
        """
    
    def get_service_endpoint(self, service_id: str) -> Optional[str]:
        """
        Get service endpoint by ID.
        
        Args:
            service_id (str): Service identifier
            
        Returns:
            Optional[str]: Service endpoint URL
        """
    
    @property
    def id(self) -> str:
        """Get the DID identifier."""
    
    @property
    def verification_methods(self) -> List[Dict[str, Any]]:
        """Get verification methods."""
    
    @property
    def services(self) -> List[Dict[str, Any]]:
        """Get service endpoints."""

def is_valid_did_doc(did_doc: Any) -> bool:
    """
    Validate DID document structure.
    
    Args:
        did_doc (Any): Document to validate
        
    Returns:
        bool: True if valid DID document
    """

Usage example:

from atproto import DidDocument, is_valid_did_doc

# Parse DID document from JSON
did_doc_data = {
    "id": "did:plc:alice123",
    "verificationMethod": [...],
    "service": [
        {
            "id": "#atproto_pds",
            "type": "AtprotoPersonalDataServer", 
            "serviceEndpoint": "https://alice.pds.example.com"
        }
    ]
}

# Validate and parse
if is_valid_did_doc(did_doc_data):
    doc = DidDocument.from_dict(did_doc_data)
    
    print(f"DID: {doc.id}")
    
    # Extract PDS endpoint
    pds_endpoint = doc.get_pds_endpoint()
    if pds_endpoint:
        print(f"PDS: {pds_endpoint}")
    
    # Get specific service
    pds_service = doc.get_service_endpoint("#atproto_pds")
    if pds_service:
        print(f"PDS Service: {pds_service}")

Validation Utilities

def is_valid_cid(cid: Union[str, bytes]) -> bool:
    """
    Validate CID format.
    
    Args:
        cid (Union[str, bytes]): CID to validate
        
    Returns:
        bool: True if valid CID
    """

def is_valid_at_uri(uri: str) -> bool:
    """
    Validate AT Protocol URI format.
    
    Args:
        uri (str): URI to validate
        
    Returns:
        bool: True if valid AT URI
    """

def is_valid_nsid(nsid: str) -> bool:
    """
    Validate NSID format.
    
    Args:
        nsid (str): NSID to validate
        
    Returns:
        bool: True if valid NSID
    """

def is_valid_did(did: str) -> bool:
    """
    Validate DID format.
    
    Args:
        did (str): DID to validate
        
    Returns:
        bool: True if valid DID
    """

These validation utilities help ensure data integrity when working with AT Protocol identifiers and documents:

from atproto import is_valid_cid, is_valid_at_uri, is_valid_nsid, is_valid_did

# Validate various AT Protocol identifiers
identifiers = {
    "CID": "bafybeigdyrzt5n4k53bvolvbtng2jrcaykswh75f3cw5vyz5l2qx2d6vby",
    "AT URI": "at://alice.bsky.social/app.bsky.feed.post/123",
    "NSID": "app.bsky.feed.post",
    "DID": "did:plc:alice123456789"
}

validators = {
    "CID": is_valid_cid,
    "AT URI": is_valid_at_uri,
    "NSID": is_valid_nsid,
    "DID": is_valid_did
}

for type_name, identifier in identifiers.items():
    validator = validators[type_name]
    is_valid = validator(identifier)
    print(f"{type_name} '{identifier}': {'valid' if is_valid else 'invalid'}")

Install with Tessl CLI