-16

src/atpasser/blob/__init__.py

··· 1 1 - import cid 2 2 - import multihash, hashlib 3 3 - 4 4 - 5 5 - def generateCID(file): 6 6 - hasher = hashlib.new("sha-256") 7 7 - while True: 8 8 - chunk = file.read(8192) 9 9 - if not chunk: 10 10 - break 11 11 - hasher.update(chunk) 12 12 - 13 13 - digest = hasher.digest 14 14 - mh = multihash.encode(digest, "sha-256") 15 15 - 16 16 - return cid.CIDv1(codec="raw", multihash=mh)

-1

src/atpasser/data/__init__.py

··· 1 1 - from ._wrapper import *

-76

src/atpasser/data/_data.py

··· 1 1 - import base64 2 2 - from cid import CIDv0, CIDv1, cid, make_cid 3 3 - import json 4 4 - 5 5 - 6 6 - class Data: 7 7 - """ 8 8 - A class representing data with "$type" key. 9 9 - 10 10 - Attributes: 11 11 - type (str): The type of the data. 12 12 - json (str): Original object in JSON format. 13 13 - """ 14 14 - 15 15 - def __init__(self, dataType: str, json: str = "{}") -> None: 16 16 - """ 17 17 - Initalizes data object. 18 18 - 19 19 - Parameters: 20 20 - type (str): The type of the data. 21 21 - json (str): Original object in JSON format. 22 22 - """ 23 23 - self.type = dataType 24 24 - self.json = json 25 25 - 26 26 - def data(self): 27 27 - """ 28 28 - Loads data as a Python-friendly format. 29 29 - 30 30 - Returns: 31 31 - dict: Converted data from JSON object. 32 32 - """ 33 33 - return json.loads(self.json, object_hook=dataHook) 34 34 - 35 35 - 36 36 - def dataHook(data: dict): 37 37 - """ 38 38 - Treated as `JSONDecoder`'s `object_hook` 39 39 - 40 40 - Parameters: 41 41 - data: data in format that `JSONDecoder` like ;) 42 42 - """ 43 43 - if "$bytes" in data: 44 44 - return base64.b64decode(data["$bytes"]) 45 45 - elif "$link" in data: 46 46 - return make_cid(data["$link"]) 47 47 - elif "$type" in data: 48 48 - dataType = data["$type"] 49 49 - del data["$type"] 50 50 - return Data(dataType, json.dumps(data)) 51 51 - else: 52 52 - return data 53 53 - 54 54 - 55 55 - def _convertDataToFakeJSON(data): 56 56 - if isinstance(data, bytes): 57 57 - return {"$bytes": base64.b64encode(data)} 58 58 - elif isinstance(data, (CIDv0, CIDv1)): 59 59 - return {"link": data.encode()} 60 60 - elif isinstance(data, dict): 61 61 - for item in data: 62 62 - data[item] = _convertDataToFakeJSON(data[item]) 63 63 - elif isinstance(data, (tuple, list, set)): 64 64 - return [_convertDataToFakeJSON(item) for item in data] 65 65 - else: 66 66 - return data 67 67 - 68 68 - 69 69 - class DataEncoder(json.JSONEncoder): 70 70 - """ 71 71 - A superset of `JSONEncoder` to support ATProto data. 72 72 - """ 73 73 - 74 74 - def default(self, o): 75 75 - result = _convertDataToFakeJSON(o) 76 76 - return super().default(result)

-61

src/atpasser/data/_wrapper.py

··· 1 1 - from json import loads 2 2 - from typing import Callable, Any 3 3 - from ._data import * 4 4 - import functools 5 5 - 6 6 - # Pyright did the whole job. Thank it. 7 7 - 8 8 - 9 9 - class DataDecoder(json.JSONDecoder): 10 10 - """ 11 11 - A superset of `JSONDecoder` to support ATProto data. 12 12 - """ 13 13 - 14 14 - def __init__( 15 15 - self, 16 16 - *, 17 17 - object_hook: Callable[[dict[str, Any]], Any] | None = dataHook, 18 18 - parse_float: Callable[[str], Any] | None = None, 19 19 - parse_int: Callable[[str], Any] | None = None, 20 20 - parse_constant: Callable[[str], Any] | None = None, 21 21 - strict: bool = True, 22 22 - object_pairs_hook: Callable[[list[tuple[str, Any]]], Any] | None = None, 23 23 - ) -> None: 24 24 - super().__init__( 25 25 - object_hook=object_hook, 26 26 - parse_float=parse_float, 27 27 - parse_int=parse_int, 28 28 - parse_constant=parse_constant, 29 29 - strict=strict, 30 30 - object_pairs_hook=object_pairs_hook, 31 31 - ) 32 32 - 33 33 - 34 34 - # Screw it. I have to make 4 `json`-like functions. 35 35 - 36 36 - 37 37 - def _dataDecoratorForDump(func): 38 38 - @functools.wraps(func) 39 39 - def wrapper(obj, *args, **kwargs): 40 40 - kwargs.setdefault("cls", DataEncoder) 41 41 - return func(obj, *args, **kwargs) 42 42 - 43 43 - return wrapper 44 44 - 45 45 - 46 46 - def _dataDecoratorForLoad(func): 47 47 - @functools.wraps(func) 48 48 - def wrapper(obj, *args, **kwargs): 49 49 - kwargs.setdefault("cls", DataDecoder) 50 50 - return func(obj, *args, **kwargs) 51 51 - 52 52 - return wrapper 53 53 - 54 54 - 55 55 - dump = _dataDecoratorForDump(json.dump) 56 56 - dumps = _dataDecoratorForDump(json.dumps) 57 57 - load = _dataDecoratorForLoad(json.load) 58 58 - loads = _dataDecoratorForLoad(json.loads) 59 59 - """ 60 60 - Wrapper of the JSON functions to support ATProto data. 61 61 - """

-137

src/atpasser/data/cbor.py

··· 1 1 - from datetime import tzinfo 2 2 - import typing 3 3 - import cbor2 4 4 - import cid 5 5 - 6 6 - from .data import dataHook, Data 7 7 - 8 8 - 9 9 - def tagHook(decoder: cbor2.CBORDecoder, tag: cbor2.CBORTag, shareable_index=None): 10 10 - """ 11 11 - A simple tag hook for CID support. 12 12 - """ 13 13 - return cid.from_bytes(tag.value) if tag.tag == 42 else tag 14 14 - 15 15 - 16 16 - class CBOREncoder(cbor2.CBOREncoder): 17 17 - """ 18 18 - Wrapper of cbor2.CBOREncoder. 19 19 - """ 20 20 - 21 21 - def __init__( 22 22 - self, 23 23 - fp: typing.IO[bytes], 24 24 - datetime_as_timestamp: bool = False, 25 25 - timezone: tzinfo | None = None, 26 26 - value_sharing: bool = False, 27 27 - default: ( 28 28 - typing.Callable[[cbor2.CBOREncoder, typing.Any], typing.Any] | None 29 29 - ) = None, 30 30 - canonical: bool = False, 31 31 - date_as_datetime: bool = False, 32 32 - string_referencing: bool = False, 33 33 - indefinite_containers: bool = False, 34 34 - ): 35 35 - super().__init__( 36 36 - fp, 37 37 - datetime_as_timestamp, 38 38 - timezone, 39 39 - value_sharing, 40 40 - default, 41 41 - canonical, 42 42 - date_as_datetime, 43 43 - string_referencing, 44 44 - indefinite_containers, 45 45 - ) 46 46 - 47 47 - @cbor2.shareable_encoder 48 48 - def cidOrDataEncoder(self: cbor2.CBOREncoder, value: cid.CIDv0 | cid.CIDv1 | Data): 49 49 - """ 50 50 - Encode CID or Data to CBOR Tag. 51 51 - """ 52 52 - if isinstance(value, (cid.CIDv0, cid.CIDv1)): 53 53 - self.encode(cbor2.CBORTag(42, value.encode())) 54 54 - elif isinstance(value, Data): 55 55 - self.encode(value.data()) 56 56 - 57 57 - 58 58 - def _cborObjectHook(decoder: cbor2.CBORDecoder, value): 59 59 - return dataHook(value) 60 60 - 61 61 - 62 62 - class CBORDecoder(cbor2.CBORDecoder): 63 63 - """ 64 64 - Wrapper of cbor2.CBORDecoder. 65 65 - """ 66 66 - 67 67 - def __init__( 68 68 - self, 69 69 - fp: typing.IO[bytes], 70 70 - tag_hook: ( 71 71 - typing.Callable[[cbor2.CBORDecoder, cbor2.CBORTag], typing.Any] | None 72 72 - ) = tagHook, 73 73 - object_hook: ( 74 74 - typing.Callable[ 75 75 - [cbor2.CBORDecoder, dict[typing.Any, typing.Any]], typing.Any 76 76 - ] 77 77 - | None 78 78 - ) = _cborObjectHook, 79 79 - str_errors: typing.Literal["strict", "error", "replace"] = "strict", 80 80 - ): 81 81 - super().__init__(fp, tag_hook, object_hook, str_errors) 82 82 - 83 83 - 84 84 - # Make things for CBOR again. 85 85 - 86 86 - from io import BytesIO 87 87 - 88 88 - 89 89 - def dumps( 90 90 - obj: object, 91 91 - datetime_as_timestamp: bool = False, 92 92 - timezone: tzinfo | None = None, 93 93 - value_sharing: bool = False, 94 94 - default: typing.Callable[[cbor2.CBOREncoder, typing.Any], typing.Any] | None = None, 95 95 - canonical: bool = False, 96 96 - date_as_datetime: bool = False, 97 97 - string_referencing: bool = False, 98 98 - indefinite_containers: bool = False, 99 99 - ) -> bytes: 100 100 - with BytesIO() as fp: 101 101 - CBOREncoder( 102 102 - fp, 103 103 - datetime_as_timestamp=datetime_as_timestamp, 104 104 - timezone=timezone, 105 105 - value_sharing=value_sharing, 106 106 - default=default, 107 107 - canonical=canonical, 108 108 - date_as_datetime=date_as_datetime, 109 109 - string_referencing=string_referencing, 110 110 - indefinite_containers=indefinite_containers, 111 111 - ).encode(obj) 112 112 - return fp.getvalue() 113 113 - 114 114 - 115 115 - def dump( 116 116 - obj: object, 117 117 - fp: typing.IO[bytes], 118 118 - datetime_as_timestamp: bool = False, 119 119 - timezone: tzinfo | None = None, 120 120 - value_sharing: bool = False, 121 121 - default: typing.Callable[[cbor2.CBOREncoder, typing.Any], typing.Any] | None = None, 122 122 - canonical: bool = False, 123 123 - date_as_datetime: bool = False, 124 124 - string_referencing: bool = False, 125 125 - indefinite_containers: bool = False, 126 126 - ) -> None: 127 127 - CBOREncoder( 128 128 - fp, 129 129 - datetime_as_timestamp=datetime_as_timestamp, 130 130 - timezone=timezone, 131 131 - value_sharing=value_sharing, 132 132 - default=default, 133 133 - canonical=canonical, 134 134 - date_as_datetime=date_as_datetime, 135 135 - string_referencing=string_referencing, 136 136 - indefinite_containers=indefinite_containers, 137 137 - ).encode(obj)

-4

tests/__init__.py

··· 1 1 - if __name__ != "__main__": 2 2 - raise Exception("name != main") 3 3 - 4 4 - import _strings

-179

tests/_strings.py

··· 1 1 - from atpasser import did, handle, nsid, rKey, uri 2 2 - 3 3 - 4 4 - testStrings, testMethods = {}, {} 5 5 - 6 6 - 7 7 - testStrings[ 8 8 - "did" 9 9 - 10 10 - ] = """did:plc:z72i7hdynmk6r22z27h6tvur 11 11 - 12 12 - did:web:blueskyweb.xyz 13 13 - 14 14 - did:method:val:two 15 15 - 16 16 - did:m:v 17 17 - 18 18 - did:method::::val 19 19 - 20 20 - did:method:-:_:. 21 21 - 22 22 - did:key:zQ3shZc2QzApp2oymGvQbzP8eKheVshBHbU4ZYjeXqwSKEn6N 23 23 - 24 24 - did:METHOD:val 25 25 - 26 26 - did:m123:val 27 27 - 28 28 - DID:method:val 29 29 - did:method: 30 30 - 31 31 - did:method:val/two 32 32 - 33 33 - did:method:val?two 34 34 - 35 35 - did:method:val#two""" 36 36 - 37 37 - testMethods["did"] = did.DID 38 38 - 39 39 - 40 40 - testStrings[ 41 41 - "handle" 42 42 - 43 43 - ] = """jay.bsky.social 44 44 - 45 45 - 8.cn 46 46 - 47 47 - name.t--t 48 48 - 49 49 - XX.LCS.MIT.EDU 50 50 - a.co 51 51 - 52 52 - xn--notarealidn.com 53 53 - 54 54 - xn--fiqa61au8b7zsevnm8ak20mc4a87e.xn--fiqs8s 55 55 - 56 56 - xn--ls8h.test 57 57 - example.t 58 58 - 59 59 - jo@hn.test 60 60 - 61 61 - 💩.tes 62 62 - t 63 63 - john..test 64 64 - 65 65 - xn--bcher-.tld 66 66 - 67 67 - john.0 68 68 - 69 69 - cn.8 70 70 - 71 71 - www.masełkowski.pl.com 72 72 - 73 73 - org 74 74 - 75 75 - name.org. 76 76 - 77 77 - 2gzyxa5ihm7nsggfxnu52rck2vv4rvmdlkiu3zzui5du4xyclen53wid.onion 78 78 - laptop.local 79 79 - 80 80 - blah.arpa""" 81 81 - 82 82 - testMethods["handle"] = handle.Handle 83 83 - 84 84 - 85 85 - testStrings[ 86 86 - "nsid" 87 87 - 88 88 - ] = """com.example.fooBar 89 89 - 90 90 - net.users.bob.ping 91 91 - 92 92 - a-0.b-1.c 93 93 - 94 94 - a.b.c 95 95 - 96 96 - com.example.fooBarV2 97 97 - 98 98 - cn.8.lex.stuff 99 99 - 100 100 - com.exa💩ple.thin 101 101 - com.example 102 102 - 103 103 - com.example.3""" 104 104 - 105 105 - testMethods["nsid"] = nsid.NSID 106 106 - 107 107 - 108 108 - testStrings[ 109 109 - 110 110 - "rkey" 111 111 - 112 112 - ] = """3jui7kd54zh2y 113 113 - self 114 114 - example.com 115 115 - 116 116 - ~1.2-3_ 117 117 - 118 118 - dHJ1ZQ 119 119 - pre:fix 120 120 - 121 121 - _ 122 122 - 123 123 - alpha/beta 124 124 - . 125 125 - .. 126 126 - 127 127 - #extra 128 128 - 129 129 - @handle 130 130 - 131 131 - any space 132 132 - 133 133 - any+space 134 134 - 135 135 - number[3] 136 136 - 137 137 - number(3) 138 138 - 139 139 - "quote" 140 140 - 141 141 - dHJ1ZQ==""" 142 142 - 143 143 - testMethods["rkey"] = rKey.RKey 144 144 - 145 145 - 146 146 - testStrings[ 147 147 - "uri" 148 148 - 149 149 - ] = """at://foo.com/com.example.foo/123 150 150 - 151 151 - at://foo.com/example/123 152 152 - 153 153 - at://computer 154 154 - 155 155 - at://example.com:3000 156 156 - 157 157 - at://foo.com/ 158 158 - 159 159 - at://user:pass@foo.com""" 160 160 - 161 161 - testMethods["uri"] = uri.URI 162 162 - 163 163 - 164 164 - for item in testMethods: 165 165 - 166 166 - print(f"START TEST {item}") 167 167 - 168 168 - for value in testStrings[item].splitlines(): 169 169 - 170 170 - print(f"Value: {value}") 171 171 - 172 172 - try: 173 173 - 174 174 - print(f"str(): {str(testMethods[item](value))}") 175 175 - 176 176 - except Exception as e: 177 177 - 178 178 - print(f"× {e}") 179 179 -