Start integrating with py-ipfs-car-decoder

* Add a coroutine in DagAPI to unpack a UnixFS CAR export to a directory
* DagAPI.export() now accepts an optional Path to write the CAR to
* Use asyncio_mode=auto for pytest
* Turn "iclient" into an async fixture that closes the session on exit
* GH workflow: test all python versions up to 3.11, test kubo 0.24.0

revbump to 0.6.5

.github/workflows/python-package.yml

@@ -2,15 +2,15 @@ name: aioipfs-build
 
 on:
   push:
-    branches: [ master, devel, ci, kubo ]
+    branches: [ master, devel, ci, kubo, car ]
   pull_request:
     branches: [ master ]
 
 jobs:
   build:
     strategy:
       matrix:
-        python-version: [3.7, 3.8, 3.9]
+        python-version: ['3.7', '3.8', '3.9', '3.10', '3.11']
         os: [ubuntu-latest, macos-latest, windows-latest]
 
     runs-on: ${{ matrix.os }}
@@ -25,35 +25,30 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install -r requirements-dev.txt
-        pip install -r requirements.txt
-
-    - name: Lint with flake8
-      run: |
-        flake8 aioipfs tests --count --select=E9,F63,F7,F82 --show-source --statistics
+        pip install '.[car,test]'
 
     - name: Fetch kubo (win)
       uses: engineerd/configurator@v0.0.8
       if: startsWith(matrix.os, 'windows')
       with:
         name: ipfs.exe
-        url: "https://dist.ipfs.tech/kubo/v0.17.0/kubo_v0.17.0_windows-amd64.zip"
+        url: "https://dist.ipfs.tech/kubo/v0.24.0/kubo_v0.24.0_windows-amd64.zip"
         pathInArchive: "kubo/ipfs.exe"
 
     - name: Fetch kubo (linux)
       uses: engineerd/configurator@v0.0.8
       if: startsWith(matrix.os, 'ubuntu')
       with:
         name: ipfs
-        url: "https://dist.ipfs.tech/kubo/v0.17.0/kubo_v0.17.0_linux-amd64.tar.gz"
+        url: "https://dist.ipfs.tech/kubo/v0.24.0/kubo_v0.24.0_linux-amd64.tar.gz"
         pathInArchive: "kubo/ipfs"
 
     - name: Fetch kubo (macos)
       uses: engineerd/configurator@v0.0.8
       if: startsWith(matrix.os, 'macos')
       with:
         name: ipfs
-        url: "https://dist.ipfs.tech/kubo/v0.17.0/kubo_v0.17.0_darwin-amd64.tar.gz"
+        url: "https://dist.ipfs.tech/kubo/v0.24.0/kubo_v0.24.0_darwin-amd64.tar.gz"
         pathInArchive: "kubo/ipfs"
 
     - name: Test with pytest
@@ -77,8 +72,7 @@ jobs:
     - name: Build wheel
       run: |
         python -m pip install --upgrade pip
-        pip install -r requirements-dev.txt
-        pip install -r requirements.txt
+        pip install '.[car,test]'
         python setup.py sdist bdist_wheel
 
     - name: Upload to PyPI
@@ -88,4 +82,4 @@ jobs:
         TWINE_PASSWORD: ${{ secrets.PYPI_PASSWORD }}
       run: |
         pip install twine
-        twine upload dist/*
+        twine check dist/*.whl

README.rst

@@ -5,7 +5,7 @@ aioipfs
 :info: Asynchronous IPFS_ client library
 
 **aioipfs** is a python3 library providing an asynchronous API for IPFS_.
-Supported python versions: *3.6*, *3.7*, *3.8*, *3.9*, *3.10*, *3.11*.
+Supported python versions: *3.6*, *3.7*, *3.8*, *3.9*, *3.10*, *3.11*, *3.12*.
 
 This library supports the
 `RPC API specifications <https://docs.ipfs.tech/reference/kubo/rpc>`_
@@ -27,6 +27,14 @@ Installation
 
     pip install aioipfs
 
+Support for CAR (Content-addressed Archives) decoding (with the
+`ipfs-car-decoder package <https://github.com/kralverde/py-ipfs-car-decoder/>`_)
+can be enabled with the *car* extra:
+
+.. code-block:: shell
+
+    pip install 'aioipfs[car]'
+
 By default the *json* module from the standard Python library is used
 to decode JSON messages, but orjson_ will be used if it is installed:
 

aioipfs/__init__.py

@@ -1,4 +1,4 @@
-__version__ = '0.6.4'
+__version__ = '0.6.5'
 
 from yarl import URL
 from distutils.version import StrictVersion

aioipfs/apis/__init__.py

@@ -11,6 +11,7 @@
 
 from aioipfs.helpers import *  # noqa
 from aioipfs.exceptions import *  # noqa
+from aioipfs.util import car_decoder, have_car_decoder
 
 
 DEFAULT_TIMEOUT = 60 * 60
@@ -102,6 +103,20 @@ async def fetch_raw(self, url, params={}, timeout=DEFAULT_TIMEOUT):
 
             return data
 
+    async def car_stream(self, url, params={}):
+        if not have_car_decoder:
+            raise Exception('The CAR decoding library is not available')
+
+        stream = car_decoder.ChunkedMemoryByteStream()
+
+        async with self.driver.session.post(url, params=params) as resp:
+            async for chunk, _ in resp.content.iter_chunks():
+                await stream.append_bytes(chunk)
+
+        await stream.mark_complete()
+
+        return stream
+
     async def fetch_json(self, url, params={}, timeout=DEFAULT_TIMEOUT):
         return await self.post(url, params=params, outformat='json')
 

aioipfs/apis/dag.py

@@ -1,11 +1,13 @@
 import os.path
+from pathlib import Path
 from aiohttp import payload
 
 from aioipfs.api import SubAPI
 from aioipfs.api import boolarg
 from aioipfs.api import ARG_PARAM
 from aioipfs import multi
 from aioipfs import UnknownAPIError
+from aioipfs import util
 
 
 class DagAPI(SubAPI):
@@ -44,18 +46,63 @@ async def put(self, filename,
             return await self.post(self.url('dag/put'), mpwriter,
                                    params=params, outformat='json')
 
-    async def car_export(self, cid, progress=False):
+    async def car_export(self, cid: str, progress: bool = False,
+                         output_path: Path = None):
         """
-        Streams the selected DAG as a .car stream on stdout.
+        Streams the selected DAG as a .car stream and return it
+        as a raw buffer or write it to a file if output_path is
+        passed.
 
-        :param str cid: CID of a root to recursively export
+        :param str cid: Root CID of a DAG to recursively export
+        :param bool progress: Stream progress
+        :param Path output_path: Write the CAR data to this file (optional)
         """
 
-        return await self.fetch_raw(
+        car_data = await self.fetch_raw(
             self.url('dag/export'),
             params={ARG_PARAM: cid, 'progress': boolarg(progress)}
         )
 
+        if output_path is None:
+            return car_data
+        else:
+            with open(output_path, 'wb') as car:
+                car.write(car_data)
+
+    export = car_export
+
+    async def export_to_directory(self,
+                                  cid: str,
+                                  dst_dir: Path) -> bool:
+        """
+        Export a UnixFS DAG to a CAR and unpack it to a directory
+
+        :param str cid: CID of a UnixFS DAG to recursively export
+        :param Path dst_dir: Filesystem destination path
+        :rtype: bool
+        """
+
+        if not util.have_car_decoder:
+            raise util.CARDecoderMissing(
+                'The CAR decoding library is not available')
+
+        if not dst_dir.exists():
+            dst_dir.mkdir(parents=True, exist_ok=True)
+
+        stream = await self.car_stream(
+            self.url('dag/export'),
+            params={ARG_PARAM: cid, 'progress': boolarg(False)}
+        )
+
+        if not stream:
+            raise ValueError(f'Failed to get car stream for {cid}')
+
+        await util.car_decoder.write_car_filesystem_to_path(
+            cid, stream, str(dst_dir)
+        )
+
+        return True
+
     async def get(self, objpath, output_codec=None):
         """
         Get a DAG node from IPFS

aioipfs/util.py

@@ -1,5 +1,47 @@
 import json
 from functools import reduce
+from pathlib import Path
+
+try:
+    import ipfs_car_decoder as car_decoder
+    have_car_decoder = True
+except Exception:
+    have_car_decoder = False
+    car_decoder = None
+
+
+class CARDecoderMissing(Exception):
+    pass
+
+
+def car_open(car_path: Path):
+    """
+    Open a Content-Adressed aRchive file and return the CAR stream.
+
+    :param Path car_path: CAR file path
+    """
+
+    if not have_car_decoder:
+        raise CARDecoderMissing()
+
+    return car_decoder.FileByteStream(car_path)
+
+
+async def car_bytes(stream, cid: str) -> bytes:
+    """
+    CAR stream to bytes
+
+    :param stream: CAR stream
+    :param str cid: CID of the UnixFS directory to export
+    :rtype: bytes
+    """
+
+    buff = b''
+
+    async for chunk in car_decoder.stream_bytes(cid, stream):
+        buff += chunk
+
+    return buff
 
 
 class DotJSON(dict):

requirements-dev.txt

@@ -1,4 +1,3 @@
-importlib_metadata
 pytest
 pytest-asyncio
 tox

requirements.txt

@@ -1,7 +1,8 @@
 aiohttp>=3.7.4
 aiofiles>=0.7.0
 base58>=1.0.2
-gitignore-parser>=0.1.9
+gitignore-parser==0.1.9
 multiaddr>=0.0.9
 py-multibase>=1.0.3
 py-multiformats-cid>=0.4.3
+setuptools>=67.7.0

setup.py

@@ -7,11 +7,7 @@
 from setuptools import setup
 from setuptools import find_packages
 
-PY_VER = sys.version_info
-
-if PY_VER >= (3, 6):
-    pass
-else:
+if sys.version_info < (3, 6):
     raise RuntimeError("You need python3.6 or newer")
 
 with codecs.open(os.path.join(os.path.abspath(os.path.dirname(
@@ -28,6 +24,9 @@
 with open('requirements.txt') as f:
     install_reqs = f.read().splitlines()
 
+with open('requirements-dev.txt') as f:
+    install_test_reqs = f.read().splitlines()
+
 setup(
     name='aioipfs',
     version=version,
@@ -42,7 +41,9 @@
     include_package_data=False,
     install_requires=install_reqs,
     extras_require={
-        'orjson': ['orjson>=3.0']
+        'orjson': ['orjson>=3.0'],
+        'car': ['ipfs-car-decoder==0.1.1'],
+        'test': install_test_reqs
     },
     classifiers=[
         'Programming Language :: Python',
@@ -52,6 +53,7 @@
         'Programming Language :: Python :: 3.9',
         'Programming Language :: Python :: 3.10',
         'Programming Language :: Python :: 3.11',
+        'Programming Language :: Python :: 3.12'
         'Intended Audience :: Developers',
         'Development Status :: 5 - Production/Stable',
         'Natural Language :: English',