Add filebase pinning client

sw_utils/ipfs.py

@@ -1,6 +1,7 @@
 import asyncio
 import logging
 from abc import ABC, abstractmethod
+from pprint import pprint
 from typing import TYPE_CHECKING, Any
 from urllib.parse import urljoin
 
@@ -203,6 +204,38 @@ async def _upload(self, form_data: aiohttp.FormData) -> str:
         return _strip_ipfs_prefix(ipfs_id)
 
 
+class FilebasePinClient:
+    """
+    https://docs.filebase.com/api-documentation/ipfs-pinning-service-api
+    """
+
+    base_url = 'https://api.filebase.io/v1/ipfs/'
+
+    def __init__(self, bucket: str, api_token: str, timeout: int = IPFS_DEFAULT_TIMEOUT):
+        self.bucket = bucket
+        self.api_token = api_token
+        self.timeout = timeout
+
+    async def pin(self, ipfs_hash: str) -> str:
+        data = {
+            'cid': ipfs_hash,
+        }
+        response = await self._call('pins', data=data)
+        pprint(response)
+        return response['pin']['cid']
+
+    async def _call(self, endpoint: str, data: dict) -> dict:
+        url = urljoin(self.base_url, endpoint)
+
+        # User and bucket are determined by token
+        headers = {'Authorization': f'Bearer {self.api_token}'}
+
+        async with aiohttp.ClientSession() as session:
+            async with session.post(url, json=data, headers=headers) as response:
+                response.raise_for_status()
+                return await response.json()
+
+
 class IpfsMultiUploadClient(BaseUploadClient):
     def __init__(self, clients: list[BaseUploadClient], retry_timeout: int = 120):
         if len(clients) == 0: