0
# Core Functionality
1
2
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.
3
4
## Capabilities
5
6
### Content Identifiers (CID)
7
8
Content Identifiers provide cryptographic addressing for IPLD data structures, enabling content-based addressing and verification.
9
10
```python { .api }
11
class CID:
12
"""
13
Content Identifier for IPLD data structures.
14
15
Attributes:
16
version (int): CID version (0 or 1)
17
codec (int): Content type codec identifier
18
hash (Multihash): Cryptographic hash of the content
19
"""
20
version: int
21
codec: int
22
hash: Multihash
23
24
@classmethod
25
def decode(cls, value: Union[str, bytes]) -> 'CID':
26
"""
27
Decode CID from string or bytes.
28
29
Args:
30
value (Union[str, bytes]): CID in string or binary format
31
32
Returns:
33
CID: Decoded CID object
34
35
Raises:
36
ValueError: If CID format is invalid
37
"""
38
39
def encode(self) -> str:
40
"""
41
Encode CID to string representation.
42
43
Returns:
44
str: Base58 encoded CID string
45
"""
46
47
def __str__(self) -> str:
48
"""String representation of the CID."""
49
50
def __eq__(self, other) -> bool:
51
"""Check CID equality."""
52
53
# Type alias for CID values
54
CIDType = Union[str, CID]
55
```
56
57
Usage example:
58
59
```python
60
from atproto import CID
61
62
# Decode CID from string
63
cid_str = "bafybeigdyrzt5n4k53bvolvbtng2jrcaykswh75f3cw5vyz5l2qx2d6vby"
64
cid = CID.decode(cid_str)
65
66
print(f"Version: {cid.version}")
67
print(f"Codec: {cid.codec}")
68
print(f"Encoded: {cid.encode()}")
69
70
# CIDs support equality comparison
71
cid2 = CID.decode(cid_str)
72
assert cid == cid2
73
```
74
75
### Content Addressable Archives (CAR)
76
77
CAR files package IPLD blocks with their CIDs for efficient storage and transmission of related content.
78
79
```python { .api }
80
class CAR:
81
"""
82
Content Addressable aRchives (CAR) file handler.
83
84
Attributes:
85
root (CID): Root CID of the CAR file
86
blocks (Dict[CID, bytes]): Dictionary mapping CIDs to block data
87
"""
88
root: CID
89
blocks: Dict[CID, bytes]
90
91
def __init__(self, root: CID, blocks: Dict[CID, bytes]):
92
"""
93
Initialize CAR with root CID and blocks.
94
95
Args:
96
root (CID): Root CID
97
blocks (Dict[CID, bytes]): Block data mapped by CID
98
"""
99
100
@classmethod
101
def from_bytes(cls, data: bytes) -> 'CAR':
102
"""
103
Decode CAR from bytes.
104
105
Args:
106
data (bytes): CAR file data
107
108
Returns:
109
CAR: Decoded CAR object
110
111
Raises:
112
ValueError: If CAR format is invalid
113
"""
114
115
def to_bytes(self) -> bytes:
116
"""
117
Encode CAR to bytes.
118
119
Returns:
120
bytes: Encoded CAR file data
121
"""
122
123
def get_block(self, cid: CID) -> Optional[bytes]:
124
"""
125
Get block data by CID.
126
127
Args:
128
cid (CID): Content identifier
129
130
Returns:
131
Optional[bytes]: Block data or None if not found
132
"""
133
```
134
135
Usage example:
136
137
```python
138
from atproto import CAR, CID
139
140
# Load CAR file
141
with open('example.car', 'rb') as f:
142
car_data = f.read()
143
144
# Parse CAR file
145
car = CAR.from_bytes(car_data)
146
147
print(f"Root CID: {car.root}")
148
print(f"Number of blocks: {len(car.blocks)}")
149
150
# Access specific block
151
block_data = car.get_block(car.root)
152
if block_data:
153
print(f"Root block size: {len(block_data)} bytes")
154
155
# Create new CAR file
156
new_blocks = {cid: data for cid, data in car.blocks.items()}
157
new_car = CAR(car.root, new_blocks)
158
car_bytes = new_car.to_bytes()
159
```
160
161
### AT Protocol URIs
162
163
AT Protocol URIs provide a standardized way to address resources in the AT Protocol network using the `at://` scheme.
164
165
```python { .api }
166
class AtUri:
167
"""
168
ATProtocol URI scheme handler for at:// URIs.
169
170
Handles both DID and handle-based URIs with path and query parameters.
171
"""
172
def __init__(self,
173
host: str,
174
pathname: str = '',
175
hash_: str = '',
176
search_params: Optional[List[Tuple[str, Any]]] = None):
177
"""
178
Initialize AT Protocol URI.
179
180
Args:
181
host (str): DID or handle
182
pathname (str): Resource path
183
hash_ (str): Fragment identifier
184
search_params (List[Tuple[str, Any]], optional): Query parameters
185
"""
186
187
@classmethod
188
def from_str(cls, uri: str) -> 'AtUri':
189
"""
190
Parse AT Protocol URI from string.
191
192
Args:
193
uri (str): URI string (e.g., "at://alice.bsky.social/app.bsky.feed.post/123")
194
195
Returns:
196
AtUri: Parsed URI object
197
198
Raises:
199
ValueError: If URI format is invalid
200
"""
201
202
@property
203
def host(self) -> str:
204
"""Get the host (DID or handle)."""
205
206
@property
207
def pathname(self) -> str:
208
"""Get the path component."""
209
210
@property
211
def collection(self) -> Optional[str]:
212
"""Extract collection name from path."""
213
214
@property
215
def rkey(self) -> Optional[str]:
216
"""Extract record key from path."""
217
218
def __str__(self) -> str:
219
"""String representation of the URI."""
220
```
221
222
Usage example:
223
224
```python
225
from atproto import AtUri
226
227
# Parse URI from string
228
uri_str = "at://alice.bsky.social/app.bsky.feed.post/3k2l4m5n6"
229
uri = AtUri.from_str(uri_str)
230
231
print(f"Host: {uri.host}")
232
print(f"Collection: {uri.collection}")
233
print(f"Record key: {uri.rkey}")
234
235
# Create URI programmatically
236
post_uri = AtUri(
237
host="did:plc:alice123",
238
pathname="/app.bsky.feed.post/abc123"
239
)
240
print(f"URI: {post_uri}")
241
242
# Extract components
243
if uri.collection == "app.bsky.feed.post":
244
print(f"This is a post with rkey: {uri.rkey}")
245
```
246
247
### Namespaced IDs (NSID)
248
249
NSIDs provide a hierarchical naming system for AT Protocol lexicons and schema definitions.
250
251
```python { .api }
252
class NSID:
253
"""
254
NameSpaced IDs for ATProto lexicons.
255
256
Attributes:
257
segments (List[str]): Dot-separated segments of the NSID
258
"""
259
segments: List[str]
260
261
@classmethod
262
def from_str(cls, nsid: str) -> 'NSID':
263
"""
264
Parse NSID from string.
265
266
Args:
267
nsid (str): NSID string (e.g., "app.bsky.feed.post")
268
269
Returns:
270
NSID: Parsed NSID object
271
272
Raises:
273
ValueError: If NSID format is invalid
274
"""
275
276
@property
277
def authority(self) -> str:
278
"""Get the authority segment (first part)."""
279
280
@property
281
def name(self) -> str:
282
"""Get the name segment (last part)."""
283
284
def __str__(self) -> str:
285
"""String representation of the NSID."""
286
287
def validate_nsid(nsid: str) -> bool:
288
"""
289
Validate NSID string format.
290
291
Args:
292
nsid (str): NSID string to validate
293
294
Returns:
295
bool: True if valid NSID format
296
"""
297
```
298
299
Usage example:
300
301
```python
302
from atproto import NSID, validate_nsid
303
304
# Parse NSID
305
nsid_str = "app.bsky.feed.post"
306
nsid = NSID.from_str(nsid_str)
307
308
print(f"Authority: {nsid.authority}") # "app"
309
print(f"Name: {nsid.name}") # "post"
310
print(f"Segments: {nsid.segments}") # ["app", "bsky", "feed", "post"]
311
312
# Validate NSID format
313
valid_nsids = [
314
"app.bsky.feed.post",
315
"com.example.my.record",
316
"io.github.user.schema"
317
]
318
319
for nsid_str in valid_nsids:
320
if validate_nsid(nsid_str):
321
print(f"Valid NSID: {nsid_str}")
322
```
323
324
### DID Documents
325
326
DID documents provide decentralized identity information including service endpoints and cryptographic keys.
327
328
```python { .api }
329
class DidDocument:
330
"""
331
Decentralized Identifier document representation.
332
333
Contains identity information, service endpoints, and verification methods.
334
"""
335
@classmethod
336
def from_dict(cls, data: dict) -> 'DidDocument':
337
"""
338
Create DID document from dictionary.
339
340
Args:
341
data (dict): DID document data
342
343
Returns:
344
DidDocument: Parsed DID document
345
346
Raises:
347
ValueError: If document format is invalid
348
"""
349
350
def get_pds_endpoint(self) -> Optional[str]:
351
"""
352
Extract Personal Data Server endpoint from document.
353
354
Returns:
355
Optional[str]: PDS endpoint URL or None if not found
356
"""
357
358
def get_service_endpoint(self, service_id: str) -> Optional[str]:
359
"""
360
Get service endpoint by ID.
361
362
Args:
363
service_id (str): Service identifier
364
365
Returns:
366
Optional[str]: Service endpoint URL
367
"""
368
369
@property
370
def id(self) -> str:
371
"""Get the DID identifier."""
372
373
@property
374
def verification_methods(self) -> List[Dict[str, Any]]:
375
"""Get verification methods."""
376
377
@property
378
def services(self) -> List[Dict[str, Any]]:
379
"""Get service endpoints."""
380
381
def is_valid_did_doc(did_doc: Any) -> bool:
382
"""
383
Validate DID document structure.
384
385
Args:
386
did_doc (Any): Document to validate
387
388
Returns:
389
bool: True if valid DID document
390
"""
391
```
392
393
Usage example:
394
395
```python
396
from atproto import DidDocument, is_valid_did_doc
397
398
# Parse DID document from JSON
399
did_doc_data = {
400
"id": "did:plc:alice123",
401
"verificationMethod": [...],
402
"service": [
403
{
404
"id": "#atproto_pds",
405
"type": "AtprotoPersonalDataServer",
406
"serviceEndpoint": "https://alice.pds.example.com"
407
}
408
]
409
}
410
411
# Validate and parse
412
if is_valid_did_doc(did_doc_data):
413
doc = DidDocument.from_dict(did_doc_data)
414
415
print(f"DID: {doc.id}")
416
417
# Extract PDS endpoint
418
pds_endpoint = doc.get_pds_endpoint()
419
if pds_endpoint:
420
print(f"PDS: {pds_endpoint}")
421
422
# Get specific service
423
pds_service = doc.get_service_endpoint("#atproto_pds")
424
if pds_service:
425
print(f"PDS Service: {pds_service}")
426
```
427
428
## Validation Utilities
429
430
```python { .api }
431
def is_valid_cid(cid: Union[str, bytes]) -> bool:
432
"""
433
Validate CID format.
434
435
Args:
436
cid (Union[str, bytes]): CID to validate
437
438
Returns:
439
bool: True if valid CID
440
"""
441
442
def is_valid_at_uri(uri: str) -> bool:
443
"""
444
Validate AT Protocol URI format.
445
446
Args:
447
uri (str): URI to validate
448
449
Returns:
450
bool: True if valid AT URI
451
"""
452
453
def is_valid_nsid(nsid: str) -> bool:
454
"""
455
Validate NSID format.
456
457
Args:
458
nsid (str): NSID to validate
459
460
Returns:
461
bool: True if valid NSID
462
"""
463
464
def is_valid_did(did: str) -> bool:
465
"""
466
Validate DID format.
467
468
Args:
469
did (str): DID to validate
470
471
Returns:
472
bool: True if valid DID
473
"""
474
```
475
476
These validation utilities help ensure data integrity when working with AT Protocol identifiers and documents:
477
478
```python
479
from atproto import is_valid_cid, is_valid_at_uri, is_valid_nsid, is_valid_did
480
481
# Validate various AT Protocol identifiers
482
identifiers = {
483
"CID": "bafybeigdyrzt5n4k53bvolvbtng2jrcaykswh75f3cw5vyz5l2qx2d6vby",
484
"AT URI": "at://alice.bsky.social/app.bsky.feed.post/123",
485
"NSID": "app.bsky.feed.post",
486
"DID": "did:plc:alice123456789"
487
}
488
489
validators = {
490
"CID": is_valid_cid,
491
"AT URI": is_valid_at_uri,
492
"NSID": is_valid_nsid,
493
"DID": is_valid_did
494
}
495
496
for type_name, identifier in identifiers.items():
497
validator = validators[type_name]
498
is_valid = validator(identifier)
499
print(f"{type_name} '{identifier}': {'valid' if is_valid else 'invalid'}")
500
```