py_hamt

52def blake3_hashfn(input_bytes: bytes) -> bytes:
53    """
54    This is the default blake3 hash function used for the `HAMT`, with a 32 byte hash size.
55
56    """
57    # 32 bytes is the recommended byte size for blake3 and the default, but multihash forces us to explicitly specify
58    digest: bytes = b3.digest(input_bytes, size=32)
59    raw_bytes: bytes = b3.unwrap(digest)
60    return raw_bytes

288class HAMT:
289    """
290    An implementation of a Hash Array Mapped Trie for an arbitrary Content Addressed Storage (CAS) system, e.g. IPFS. This uses the IPLD data model.
291
292    Use this to store arbitrarily large key-value mappings in your CAS of choice.
293
294    For writing, this HAMT is async safe but NOT thread safe. Only write in an async event loop within the same thread.
295
296    When in read-only mode, the HAMT is both async and thread safe.
297
298    #### A note about memory management, read+write and read-only modes
299    The HAMT can be in either read+write mode or read-only mode. For either of these modes, the HAMT has some internal performance optimizations.
300
301    Note that in read+write, the real root node id IS NOT VALID. You should call `make_read_only()` to convert to read only mode and then read `root_node_id`.
302
303    These optimizations also trade off performance for memory use. Use `cache_size` to monitor the approximate memory usage. Be warned that for large key-value mapping sets this may take a bit to run. Use `cache_vacate` if you are over your memory limits.
304
305    #### IPFS HAMT Sample Code
306    ```python
307    kubo_cas = KuboCAS() # connects to a local kubo node with the default endpoints
308    hamt = await HAMT.build(cas=kubo_cas)
309    await hamt.set("foo", "bar")
310    assert (await hamt.get("foo")) == "bar"
311    await hamt.make_read_only()
312    cid = hamt.root_node_id # our root node CID
313    print(cid)
314    ```
315    """
316
317    def __init__(
318        self,
319        cas: ContentAddressedStore,
320        hash_fn: Callable[[bytes], bytes] = blake3_hashfn,
321        root_node_id: IPLDKind | None = None,
322        read_only: bool = False,
323        max_bucket_size: int = 4,
324        values_are_bytes: bool = False,
325    ):
326        """
327        Use `build` if you need to create a completely empty HAMT, as this requires some async operations with the CAS. For what each of the constructor input variables refer to, check the documentation with the matching names below.
328        """
329
330        self.cas: ContentAddressedStore = cas
331        """The backing storage system. py-hamt provides an implementation `KuboCAS` for IPFS."""
332
333        self.hash_fn: Callable[[bytes], bytes] = hash_fn
334        """
335        This is the hash function used to place a key-value within the HAMT.
336
337        To provide your own hash function, create a function that takes in arbitrarily long bytes and returns the hash bytes.
338
339        It's important to note that the resulting hash must must always be a multiple of 8 bits since python bytes object can only represent in segments of bytes, and thus 8 bits.
340
341        Theoretically your hash size must only be a minimum of 1 byte, and there can be less than or the same number of hash collisions as the bucket size. Any more and the HAMT will most likely throw errors.
342        """
343
344        self.lock: asyncio.Lock = asyncio.Lock()
345        """@private"""
346
347        self.values_are_bytes: bool = values_are_bytes
348        """Set this to true if you are only going to be storing python bytes objects into the hamt. This will improve performance by skipping a serialization step from IPLDKind.
349
350        This is theoretically safe to change in between operations, but this has not been verified in testing, so only do this at your own risk.
351        """
352
353        if max_bucket_size < 1:
354            raise ValueError("Bucket size maximum must be a positive integer")
355        self.max_bucket_size: int = max_bucket_size
356        """
357        This is only important for tuning performance when writing! For reading a HAMT that was written with a different max bucket size, this does not need to match and can be left unprovided.
358
359        This is an internal detail that has been exposed for performance tuning. The HAMT handles large key-value mapping sets even on a content addressed system by essentially sharding all the mappings across many smaller Nodes. The memory footprint of each of these Nodes footprint is a linear function of the maximum bucket size. Larger bucket sizes will result in larger Nodes, but more time taken to retrieve and decode these nodes from your backing CAS.
360
361        This must be a positive integer with a minimum of 1.
362        """
363
364        self.root_node_id: IPLDKind = root_node_id
365        """
366        This is type IPLDKind but the documentation generator pdoc mangles it a bit.
367
368        Read from this only when in read mode to get something valid!
369        """
370
371        self.read_only: bool = read_only
372        """Clients should NOT modify this.
373
374        This is here for checking whether the HAMT is in read only or read/write mode.
375
376        The distinction is made for performance and correctness reasons. In read only mode, the HAMT has an internal read cache that can speed up operations. In read/write mode, for reads the HAMT maintains strong consistency for reads by using async locks, and for writes the HAMT writes to an in memory buffer rather than performing (possibly) network calls to the underlying CAS.
377        """
378        self.node_store: NodeStore
379        """@private"""
380        if read_only:
381            self.node_store = ReadCacheStore(self)
382        else:
383            self.node_store = InMemoryTreeStore(self)
384
385    @classmethod
386    async def build(cls, *args: Any, **kwargs: Any) -> "HAMT":
387        """
388        Use this if you are initializing a completely empty HAMT! That means passing in None for the root_node_id. Method arguments are the exact same as `__init__`. If the root_node_id is not None, this will have no difference than creating a HAMT instance with __init__.
389
390        This separate async method is required since initializing an empty HAMT means sending some internal objects to the underlying CAS, which requires async operations. python does not allow for an async __init__, so this method is separately provided.
391        """
392        hamt = cls(*args, **kwargs)
393        if hamt.root_node_id is None:
394            hamt.root_node_id = await hamt.node_store.save(None, Node())
395        return hamt
396
397    # This is typically a massive blocking operation, you dont want to be running this concurrently with a bunch of other operations, so it's ok to have it not be async
398    async def make_read_only(self) -> None:
399        """
400        Makes the HAMT read only, which allows for more parallel read operations. The HAMT also needs to be in read only mode to get the real root node ID.
401
402        In read+write mode, the HAMT normally has to block separate get calls to enable strong consistency in case a set/delete operation falls in between.
403        """
404        async with self.lock:
405            inmemory_tree: InMemoryTreeStore = cast(InMemoryTreeStore, self.node_store)
406            await inmemory_tree.vacate()
407
408            self.read_only = True
409            self.node_store = ReadCacheStore(self)
410
411    async def enable_write(self) -> None:
412        """
413        Enable both reads and writes. This creates an internal structure for performance optimizations which will result in the root node ID no longer being valid, in order to read that at the end of your operations you must first use `make_read_only`.
414        """
415        async with self.lock:
416            # The read cache has no writes that need to be sent upstream so we can remove it without vacating
417            self.read_only = False
418            self.node_store = InMemoryTreeStore(self)
419
420    async def cache_size(self) -> int:
421        """
422        Returns the memory used by some internal performance optimization tools in bytes.
423
424        This is async concurrency safe, so call it whenever. This does mean it will block and wait for other writes to finish however.
425
426        Be warned that this may take a while to run for large HAMTs.
427
428        For more on memory management, see the `HAMT` class documentation.
429        """
430        if self.read_only:
431            return self.node_store.size()
432        async with self.lock:
433            return self.node_store.size()
434
435    async def cache_vacate(self) -> None:
436        """
437        Vacate and completely empty out the internal read/write cache.
438
439        Be warned that this may take a while if there have been a lot of write operations.
440
441        For more on memory management, see the `HAMT` class documentation.
442        """
443        if self.read_only:
444            await self.node_store.vacate()
445        else:
446            async with self.lock:
447                await self.node_store.vacate()
448
449    async def _reserialize_and_link(
450        self, node_stack: list[tuple[IPLDKind, Node]]
451    ) -> None:
452        """
453        This function starts from the node at the end of the list and reserializes so that each node holds valid new IDs after insertion into the store
454        Takes a stack of nodes, we represent a stack with a list where the first element is the root element and the last element is the top of the stack
455        Each element in the list is a tuple where the first element is the ID from the store and the second element is the Node in python
456        If a node ends up being empty, then it is deleted entirely, unless it is the root node
457        Modifies in place
458        """
459        # iterate in the reverse direction, this range goes from n-1 to 0, from the bottommost tree node to the root
460        for stack_index in range(len(node_stack) - 1, -1, -1):
461            old_id, node = node_stack[stack_index]
462
463            # If this node is empty, and it's not the root node, then we can delete it entirely from the list
464            is_root: bool = stack_index == 0
465            if node.is_empty() and not is_root:
466                # Unlink from the rest of the tree
467                _, prev_node = node_stack[stack_index - 1]
468                # When removing links, don't worry about two nodes having the same link since all nodes are guaranteed to be different by the removal of empty nodes after every single operation
469                for link_index in prev_node.iter_link_indices():
470                    link = prev_node.get_link(link_index)
471                    if link == old_id:
472                        # Delete the link by making it an empty bucket
473                        prev_node.data[link_index] = {}
474                        break
475
476                # Remove from our stack, continue reserializing up the tree
477                node_stack.pop(stack_index)
478                continue
479
480            # If not an empty node, just reserialize like normal and replace this one
481            new_store_id: IPLDKind = await self.node_store.save(old_id, node)
482            node_stack[stack_index] = (new_store_id, node)
483
484            # If this is not the last i.e. root node, we need to change the linking of the node prior in the list since we just reserialized
485            if not is_root:
486                _, prev_node = node_stack[stack_index - 1]
487                prev_node.replace_link(old_id, new_store_id)
488
489    # automatically skip encoding if the value provided is of the bytes variety
490    async def set(self, key: str, val: IPLDKind) -> None:
491        """Write a key-value mapping."""
492        if self.read_only:
493            raise Exception("Cannot call set on a read only HAMT")
494
495        data: bytes
496        if self.values_are_bytes:
497            data = cast(
498                bytes, val
499            )  # let users get an exception if they pass in a non bytes when they want to skip encoding
500        else:
501            data = dag_cbor.encode(val)
502
503        pointer: IPLDKind = await self.cas.save(data, codec="raw")
504        await self._set_pointer(key, pointer)
505
506    async def _set_pointer(self, key: str, val_ptr: IPLDKind) -> None:
507        async with self.lock:
508            node_stack: list[tuple[IPLDKind, Node]] = []
509            root_node: Node = await self.node_store.load(self.root_node_id)
510            node_stack.append((self.root_node_id, root_node))
511
512            # FIFO queue to keep track of all the KVs we need to insert
513            # This is needed if any buckets overflow and so we need to reinsert all those KVs
514            kvs_queue: list[tuple[str, IPLDKind]] = []
515            kvs_queue.append((key, val_ptr))
516
517            while len(kvs_queue) > 0:
518                _, top_node = node_stack[-1]
519                curr_key, curr_val_ptr = kvs_queue[0]
520
521                raw_hash: bytes = self.hash_fn(curr_key.encode())
522                map_key: int = extract_bits(raw_hash, len(node_stack) - 1, 8)
523
524                item = top_node.data[map_key]
525                if isinstance(item, list):
526                    next_node_id: IPLDKind = item[0]
527                    next_node: Node = await self.node_store.load(next_node_id)
528                    node_stack.append((next_node_id, next_node))
529                elif isinstance(item, dict):
530                    bucket: dict[str, IPLDKind] = item
531
532                    # If this bucket already has this same key, or has space, just rewrite the value and then go work on the others in the queue
533                    if curr_key in bucket or len(bucket) < self.max_bucket_size:
534                        bucket[curr_key] = curr_val_ptr
535                        kvs_queue.pop(0)
536                        continue
537
538                    # The current key is not in the bucket and the bucket is too full, so empty KVs from the bucket and restart insertion
539                    for k in bucket:
540                        v_ptr = bucket[k]
541                        kvs_queue.append((k, v_ptr))
542
543                    # Create a new link to a new node so that we can reflow these KVs into a new subtree
544                    new_node = Node()
545                    new_node_id: IPLDKind = await self.node_store.save(None, new_node)
546                    link: list[IPLDKind] = [new_node_id]
547                    top_node.data[map_key] = link
548
549            # Finally, reserialize and fix all links, deleting empty nodes as needed
550            await self._reserialize_and_link(node_stack)
551            self.root_node_id = node_stack[0][0]
552
553    async def delete(self, key: str) -> None:
554        """Delete a key-value mapping."""
555
556        # Also deletes the pointer at the same time so this doesn't have a _delete_pointer duo
557        if self.read_only:
558            raise Exception("Cannot call delete on a read only HAMT")
559
560        async with self.lock:
561            raw_hash: bytes = self.hash_fn(key.encode())
562
563            node_stack: list[tuple[IPLDKind, Node]] = []
564            root_node: Node = await self.node_store.load(self.root_node_id)
565            node_stack.append((self.root_node_id, root_node))
566
567            created_change: bool = False
568            while True:
569                _, top_node = node_stack[-1]
570                map_key: int = extract_bits(raw_hash, len(node_stack) - 1, 8)
571
572                item = top_node.data[map_key]
573                if isinstance(item, dict):
574                    bucket = item
575                    if key in bucket:
576                        del bucket[key]
577                        created_change = True
578                    # Break out since whether or not the key is in the bucket, it should have been here so either now reserialize or raise a KeyError
579                    break
580                elif isinstance(item, list):
581                    link: IPLDKind = item[0]
582                    next_node: Node = await self.node_store.load(link)
583                    node_stack.append((link, next_node))
584
585            # Finally, reserialize and fix all links, deleting empty nodes as needed
586            if created_change:
587                await self._reserialize_and_link(node_stack)
588                self.root_node_id = node_stack[0][0]
589            else:
590                # If we didn't make a change, then this key must not exist within the HAMT
591                raise KeyError
592
593    async def get(self, key: str) -> IPLDKind:
594        """Get a value."""
595        pointer: IPLDKind = await self.get_pointer(key)
596        data: bytes = await self.cas.load(pointer)
597        if self.values_are_bytes:
598            return data
599        else:
600            return dag_cbor.decode(data)
601
602    async def get_pointer(self, key: str) -> IPLDKind:
603        """
604        Get a store ID that points to the value for this key.
605
606        This is useful for some applications that want to implement a read cache. Due to the restrictions of `ContentAddressedStore` on IDs, pointers are regarded as immutable by python so they can be easily used as IDs for read caches. This is utilized in `ZarrHAMTStore` for example.
607        """
608        # If read only, no need to acquire a lock
609        pointer: IPLDKind
610        if self.read_only:
611            pointer = await self._get_pointer(key)
612        else:
613            async with self.lock:
614                pointer = await self._get_pointer(key)
615
616        return pointer
617
618    # Callers MUST handle acquiring a lock
619    async def _get_pointer(self, key: str) -> IPLDKind:
620        raw_hash: bytes = self.hash_fn(key.encode())
621
622        current_id: IPLDKind = self.root_node_id
623        current_depth: int = 0
624
625        # Don't check if result is none but use a boolean to indicate finding something, this is because None is a possible value of IPLDKind
626        result_ptr: IPLDKind = None
627        found_a_result: bool = False
628        while True:
629            top_id: IPLDKind = current_id
630            top_node: Node = await self.node_store.load(top_id)
631            map_key: int = extract_bits(raw_hash, current_depth, 8)
632
633            # Check if this key is in one of the buckets
634            item = top_node.data[map_key]
635            if isinstance(item, dict):
636                bucket = item
637                if key in bucket:
638                    result_ptr = bucket[key]
639                    found_a_result = True
640                    break
641
642            if isinstance(item, list):
643                link: IPLDKind = item[0]
644                current_id = link
645                current_depth += 1
646                continue
647
648            # Nowhere left to go, stop walking down the tree
649            break
650
651        if not found_a_result:
652            raise KeyError
653
654        return result_ptr
655
656    # Callers MUST handle locking or not on their own
657    async def _iter_nodes(self) -> AsyncIterator[tuple[IPLDKind, Node]]:
658        node_id_stack: list[IPLDKind] = [self.root_node_id]
659        while len(node_id_stack) > 0:
660            top_id: IPLDKind = node_id_stack.pop()
661            node: Node = await self.node_store.load(top_id)
662            yield (top_id, node)
663            node_id_stack.extend(list(node.iter_links()))
664
665    async def keys(self) -> AsyncIterator[str]:
666        """
667        AsyncIterator returning all keys in the HAMT.
668
669        If the HAMT is write enabled, to maintain strong consistency this will obtain an async lock and not allow any other operations to proceed.
670
671        When the HAMT is in read only mode however, this can be run concurrently with get operations.
672        """
673        if self.read_only:
674            async for k in self._keys_no_locking():
675                yield k
676        else:
677            async with self.lock:
678                async for k in self._keys_no_locking():
679                    yield k
680
681    async def _keys_no_locking(self) -> AsyncIterator[str]:
682        async for _, node in self._iter_nodes():
683            for bucket in node.iter_buckets():
684                for key in bucket:
685                    yield key
686
687    async def len(self) -> int:
688        """
689        Return the number of key value mappings in this HAMT.
690
691        When the HAMT is write enabled, to maintain strong consistency it will acquire a lock and thus not allow any other operations to proceed until the length is fully done being calculated. If read only, then this can be run concurrently with other operations.
692        """
693        count: int = 0
694        async for _ in self.keys():
695            count += 1
696
697        return count

14class ContentAddressedStore(ABC):
15    """
16    Abstract class that represents a content addressed storage that the `HAMT` can use for keeping data.
17
18    Note that the return type of save and input to load is really type `IPLDKind`, but the documentation generator pdoc mangles it unfortunately.
19
20    #### A note on the IPLDKind return types
21    Save and load return the type IPLDKind and not just a CID. As long as python regards the underlying type as immutable it can be used, allowing for more flexibility. There are two exceptions:
22    1. No lists or dicts, since python does not classify these as immutable.
23    2. No `None` values since this is used in HAMT's `__init__` to indicate that an empty HAMT needs to be initialized.
24    """
25
26    CodecInput = Literal["raw", "dag-cbor"]
27
28    @abstractmethod
29    async def save(self, data: bytes, codec: CodecInput) -> IPLDKind:
30        """Save data to a storage mechanism, and return an ID for the data in the IPLDKind type.
31
32        `codec` will be set to "dag-cbor" if this data should be marked as special linked data a la IPLD data model.
33        """
34
35    @abstractmethod
36    async def load(self, id: IPLDKind) -> bytes:
37        """Retrieve data."""

40class InMemoryCAS(ContentAddressedStore):
41    """Used mostly for faster testing, this is why this is not exported. It hashes all inputs and uses that as a key to an in-memory python dict, mimicking a content addressed storage system. The hash bytes are the ID that `save` returns and `load` takes in."""
42
43    store: dict[bytes, bytes]
44    hash_alg: Multihash
45
46    def __init__(self):
47        self.store = dict()
48        self.hash_alg = multihash.get("blake3")
49
50    async def save(self, data: bytes, codec: ContentAddressedStore.CodecInput) -> bytes:
51        hash: bytes = self.hash_alg.digest(data, size=32)
52        self.store[hash] = data
53        return hash
54
55    async def load(self, id: IPLDKind) -> bytes:
56        """
57        `ContentAddressedStore` allows any IPLD scalar key.  For the in-memory
58        backend we *require* a `bytes` hash; anything else is rejected at run
59        time. In OO type-checking, a subclass may widen (make more general) argument types,
60        but it must never narrow them; otherwise callers that expect the base-class contract can break.
61        Mypy enforces this contra-variance rule and emits the "violates Liskov substitution principle" error.
62        This is why we use `cast` here, to tell mypy that we know what we are doing.
63        h/t https://stackoverflow.com/questions/75209249/overriding-a-method-mypy-throws-an-incompatible-with-super-type-error-when-ch
64        """
65        key = cast(bytes, id)
66        if not isinstance(key, (bytes, bytearray)):  # defensive guard
67            raise TypeError(
68                f"InMemoryCAS only supports byte‐hash keys; got {type(id).__name__}"
69            )
70
71        try:
72            return self.store[key]
73        except KeyError as exc:
74            raise KeyError("Object not found in in-memory store") from exc

 77class KuboCAS(ContentAddressedStore):
 78    """
 79    Connects to an **IPFS Kubo** daemon.
 80
 81    The IDs in save and load are IPLD CIDs.
 82
 83    * **save()**  → RPC  (`/api/v0/add`)
 84    * **load()**  → HTTP gateway  (`/ipfs/{cid}`)
 85
 86    `save` uses the RPC API and `load` uses the HTTP Gateway. This means that read-only HAMTs will only access the HTTP Gateway, so no RPC endpoint is required for use.
 87
 88    ### Authentication / custom headers
 89    You have two options:
 90
 91    1. **Bring your own `httpx.AsyncClient`**
 92       Pass it via `client=...` — any default headers or auth
 93       configured on that client are reused for **every** request.
 94    2. **Let `KuboCAS` build the client** but pass
 95       `headers=` *and*/or `auth=` kwargs; they are forwarded to the
 96       internally–created `AsyncClient`.
 97
 98    ```python
 99    import httpx
100    from py_hamt import KuboCAS
101
102    # Option 1: user-supplied client
103    client = httpx.AsyncClient(
104        headers={"Authorization": "Bearer <token>"},
105        auth=("user", "pass"),
106    )
107    cas = KuboCAS(client=client)
108
109    # Option 2: let KuboCAS create the client
110    cas = KuboCAS(
111        headers={"X-My-Header": "yes"},
112        auth=("user", "pass"),
113    )
114    ```
115
116    ### Parameters
117    - **hasher** (str): multihash name (defaults to *blake3*).
118    - **client** (`httpx.AsyncClient | None`): reuse an existing
119      client; if *None* KuboCAS will create one lazily.
120    - **headers** (dict[str, str] | None): default headers for the
121      internally-created client.
122    - **auth** (`tuple[str, str] | None`): authentication tuple (username, password)
123      for the internally-created client.
124    - **rpc_base_url / gateway_base_url** (str | None): override daemon
125      endpoints (defaults match the local daemon ports).
126    - **chunker** (str): chunking algorithm specification for Kubo's `add`
127      RPC. Accepted formats are `"size-<positive int>"`, `"rabin"`, or
128      `"rabin-<min>-<avg>-<max>"`.
129
130    ...
131    """
132
133    KUBO_DEFAULT_LOCAL_GATEWAY_BASE_URL: str = "http://127.0.0.1:8080"
134    KUBO_DEFAULT_LOCAL_RPC_BASE_URL: str = "http://127.0.0.1:5001"
135
136    DAG_PB_MARKER: int = 0x70
137    """@private"""
138
139    # Take in a httpx client that can be reused across POSTs and GETs to a specific IPFS daemon
140    def __init__(
141        self,
142        hasher: str = "blake3",
143        client: httpx.AsyncClient | None = None,
144        rpc_base_url: str | None = None,
145        gateway_base_url: str | None = None,
146        concurrency: int = 32,
147        *,
148        headers: dict[str, str] | None = None,
149        auth: Tuple[str, str] | None = None,
150        chunker: str = "size-1048576",
151        max_retries: int = 3,
152        initial_delay: float = 1.0,
153        backoff_factor: float = 2.0,
154    ):
155        """
156        If None is passed into the rpc or gateway base url, then the default for kubo local daemons will be used. The default local values will also be used if nothing is passed in at all.
157
158        ### `httpx.AsyncClient` Management
159        If `client` is not provided, it will be automatically initialized. It is the responsibility of the user to close this at an appropriate time, using `await cas.aclose()`
160        as a class instance cannot know when it will no longer be in use, unless explicitly told to do so.
161
162        If you are using the `KuboCAS` instance in an `async with` block, it will automatically close the client when the block is exited which is what we suggest below:
163        ```python
164        async with httpx.AsyncClient() as client, KuboCAS(
165            rpc_base_url=rpc_base_url,
166            gateway_base_url=gateway_base_url,
167            client=client,
168        ) as kubo_cas:
169            hamt = await HAMT.build(cas=kubo_cas, values_are_bytes=True)
170            zhs = ZarrHAMTStore(hamt)
171            # Use the KuboCAS instance as needed
172            # ...
173        ```
174        As mentioned, if you do not use the `async with` syntax, you should call `await cas.aclose()` when you are done using the instance to ensure that all resources are cleaned up.
175        ``` python
176        cas = KuboCAS(rpc_base_url=rpc_base_url, gateway_base_url=gateway_base_url)
177        # Use the KuboCAS instance as needed
178        # ...
179        await cas.aclose()  # Ensure resources are cleaned up
180        ```
181
182        ### Authenticated RPC/Gateway Access
183        Users can set whatever headers and auth credentials they need if they are connecting to an authenticated kubo instance by setting them in their own `httpx.AsyncClient` and then passing that in.
184        Alternatively, they can pass in `headers` and `auth` parameters to the constructor, which will be used to create a new `httpx.AsyncClient` if one is not provided.
185        If you do not need authentication, you can leave these parameters as `None`.
186
187        ### RPC and HTTP Gateway Base URLs
188        These are the first part of the url, defaults that refer to the default that kubo launches with on a local machine are provided.
189        """
190
191        chunker_pattern = r"(?:size-[1-9]\d*|rabin(?:-[1-9]\d*-[1-9]\d*-[1-9]\d*)?)"
192        if re.fullmatch(chunker_pattern, chunker) is None:
193            raise ValueError("Invalid chunker specification")
194        self.chunker: str = chunker
195
196        self.hasher: str = hasher
197        """The hash function to send to IPFS when storing bytes. Cannot be changed after initialization. The default blake3 follows the default hashing algorithm used by HAMT."""
198
199        if rpc_base_url is None:
200            rpc_base_url = KuboCAS.KUBO_DEFAULT_LOCAL_RPC_BASE_URL  # pragma
201        if gateway_base_url is None:
202            gateway_base_url = KuboCAS.KUBO_DEFAULT_LOCAL_GATEWAY_BASE_URL
203
204        if "/ipfs/" in gateway_base_url:
205            gateway_base_url = gateway_base_url.split("/ipfs/")[0]
206
207        # Standard gateway URL construction with proper path handling
208        if gateway_base_url.endswith("/"):
209            gateway_base_url = f"{gateway_base_url}ipfs/"
210        else:
211            gateway_base_url = f"{gateway_base_url}/ipfs/"
212
213        self.rpc_url: str = f"{rpc_base_url}/api/v0/add?hash={self.hasher}&chunker={self.chunker}&pin=false"
214        """@private"""
215        self.gateway_base_url: str = gateway_base_url
216        """@private"""
217
218        if client is not None:
219            # A client was supplied by the user. We don't own it.
220            self._owns_client = False
221            self._client_per_loop = {asyncio.get_running_loop(): client}
222        else:
223            # No client supplied. We will own any clients we create.
224            self._owns_client = True
225            self._client_per_loop = {}
226
227        # store for later use by _loop_client()
228        self._default_headers = headers
229        self._default_auth = auth
230
231        self._sem: asyncio.Semaphore = asyncio.Semaphore(concurrency)
232        self._closed: bool = False
233
234        # Validate retry parameters
235        if max_retries < 0:
236            raise ValueError("max_retries must be non-negative")
237        if initial_delay <= 0:
238            raise ValueError("initial_delay must be positive")
239        if backoff_factor < 1.0:
240            raise ValueError("backoff_factor must be >= 1.0 for exponential backoff")
241
242        self.max_retries = max_retries
243        self.initial_delay = initial_delay
244        self.backoff_factor = backoff_factor
245
246    # --------------------------------------------------------------------- #
247    # helper: get or create the client bound to the current running loop    #
248    # --------------------------------------------------------------------- #
249    def _loop_client(self) -> httpx.AsyncClient:
250        """Get or create a client for the current event loop.
251
252        If the instance was previously closed but owns its clients, a fresh
253        client mapping is lazily created on demand.  Users that supplied their
254        own ``httpx.AsyncClient`` still receive an error when the instance has
255        been closed, as we cannot safely recreate their client.
256        """
257        if self._closed:
258            if not self._owns_client:
259                raise RuntimeError("KuboCAS is closed; create a new instance")
260            # We previously closed all internally-owned clients. Reset the
261            # state so that new clients can be created lazily.
262            self._closed = False
263            self._client_per_loop = {}
264
265        loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
266        try:
267            return self._client_per_loop[loop]
268        except KeyError:
269            # Create a new client
270            client = httpx.AsyncClient(
271                timeout=60.0,
272                headers=self._default_headers,
273                auth=self._default_auth,
274                limits=httpx.Limits(max_connections=64, max_keepalive_connections=32),
275                # Uncomment when they finally support Robust HTTP/2 GOAWAY responses
276                # http2=True,
277            )
278            self._client_per_loop[loop] = client
279            return client
280
281    # --------------------------------------------------------------------- #
282    # graceful shutdown: close **all** clients we own                       #
283    # --------------------------------------------------------------------- #
284    async def aclose(self) -> None:
285        """
286        Closes all internally-created clients. Must be called from an async context.
287        """
288        if self._owns_client is False:  # external client → caller closes
289            return
290
291        # This method is async, so we can reliably await the async close method.
292        # The complex sync/async logic is handled by __del__.
293        for client in list(self._client_per_loop.values()):
294            if not client.is_closed:
295                try:
296                    await client.aclose()
297                except Exception:
298                    pass  # best-effort cleanup
299
300        self._client_per_loop.clear()
301        self._closed = True
302
303    # At this point, _client_per_loop should be empty or only contain
304    # clients from loops we haven't seen (which shouldn't happen in practice)
305    async def __aenter__(self) -> "KuboCAS":
306        return self
307
308    async def __aexit__(self, *exc: Any) -> None:
309        await self.aclose()
310
311    def __del__(self) -> None:
312        """Best-effort close for internally-created clients."""
313        if not hasattr(self, "_owns_client") or not hasattr(self, "_closed"):
314            return
315
316        if not self._owns_client or self._closed:
317            return
318
319        # Attempt proper cleanup if possible
320        try:
321            loop = asyncio.get_running_loop()
322        except RuntimeError:
323            # No running loop - can't do async cleanup
324            # Just clear the client references synchronously
325            if hasattr(self, "_client_per_loop"):
326                # We can't await client.aclose() without a loop,
327                # so just clear the references
328                self._client_per_loop.clear()
329                self._closed = True
330            return
331
332        # If we get here, we have a running loop
333        try:
334            if loop.is_running():
335                # Schedule cleanup in the existing loop
336                loop.create_task(self.aclose())
337            else:
338                # Loop exists but not running - try asyncio.run
339                coro = self.aclose()  # Create the coroutine
340                try:
341                    asyncio.run(coro)
342                except Exception:
343                    # If asyncio.run fails, we need to close the coroutine properly
344                    coro.close()  # This prevents the RuntimeWarning
345                    raise  # Re-raise to hit the outer except block
346        except Exception:
347            # If all else fails, just clear references
348            if hasattr(self, "_client_per_loop"):
349                self._client_per_loop.clear()
350                self._closed = True
351
352    # --------------------------------------------------------------------- #
353    # save() – now uses the per-loop client                                 #
354    # --------------------------------------------------------------------- #
355    async def save(self, data: bytes, codec: ContentAddressedStore.CodecInput) -> CID:
356        async with self._sem:
357            files = {"file": data}
358            client = self._loop_client()
359            retry_count = 0
360
361            while retry_count <= self.max_retries:
362                try:
363                    response = await client.post(
364                        self.rpc_url, files=files, timeout=60.0
365                    )
366                    response.raise_for_status()
367                    cid_str: str = response.json()["Hash"]
368                    cid: CID = CID.decode(cid_str)
369                    if cid.codec.code != self.DAG_PB_MARKER:
370                        cid = cid.set(codec=codec)
371                    return cid
372
373                except (httpx.TimeoutException, httpx.RequestError) as e:
374                    retry_count += 1
375                    if retry_count > self.max_retries:
376                        raise httpx.TimeoutException(
377                            f"Failed to save data after {self.max_retries} retries: {str(e)}",
378                            request=e.request
379                            if isinstance(e, httpx.RequestError)
380                            else None,
381                        )
382
383                    # Calculate backoff delay
384                    delay = self.initial_delay * (
385                        self.backoff_factor ** (retry_count - 1)
386                    )
387                    # Add some jitter to prevent thundering herd
388                    jitter = delay * 0.1 * (random.random() - 0.5)
389                    await asyncio.sleep(delay + jitter)
390
391                except httpx.HTTPStatusError:
392                    # Re-raise non-timeout HTTP errors immediately
393                    raise
394        raise RuntimeError("Exited the retry loop unexpectedly.")  # pragma: no cover
395
396    async def load(self, id: IPLDKind) -> bytes:
397        cid = cast(CID, id)
398        url: str = f"{self.gateway_base_url + str(cid)}"
399        async with self._sem:
400            client = self._loop_client()
401            retry_count = 0
402
403            while retry_count <= self.max_retries:
404                try:
405                    response = await client.get(url, timeout=60.0)
406                    response.raise_for_status()
407                    return response.content
408
409                except (httpx.TimeoutException, httpx.RequestError) as e:
410                    retry_count += 1
411                    if retry_count > self.max_retries:
412                        raise httpx.TimeoutException(
413                            f"Failed to load data after {self.max_retries} retries: {str(e)}",
414                            request=e.request
415                            if isinstance(e, httpx.RequestError)
416                            else None,
417                        )
418
419                    # Calculate backoff delay
420                    delay = self.initial_delay * (
421                        self.backoff_factor ** (retry_count - 1)
422                    )
423                    # Add some jitter to prevent thundering herd
424                    jitter = delay * 0.1 * (random.random() - 0.5)
425                    await asyncio.sleep(delay + jitter)
426
427                except httpx.HTTPStatusError:
428                    # Re-raise non-timeout HTTP errors immediately
429                    raise
430        raise RuntimeError("Exited the retry loop unexpectedly.")  # pragma: no cover