feat: add Pinata IPFS provider

src/main.py

@@ -12,7 +12,7 @@
 from src.modules.ejector.ejector import Ejector
 from src.modules.checks.checks_module import ChecksModule
 from src.modules.csm.csm import CSOracle
-from src.providers.ipfs import DummyIPFSProvider, GW3, IPFSProvider, MultiIPFSProvider
+from src.providers.ipfs import DummyIPFSProvider, GW3, IPFSProvider, MultiIPFSProvider, Pinata
 from src.typings import OracleModule
 from src.utils.build import get_build_info
 from src.web3py.extensions import (
@@ -147,7 +147,13 @@ def ipfs_providers() -> Iterable[IPFSProvider]:
         yield GW3(
             variables.GW3_ACCESS_KEY,
             variables.GW3_SECRET_KEY,
-            variables.HTTP_REQUEST_TIMEOUT_IPFS
+            timeout=variables.HTTP_REQUEST_TIMEOUT_IPFS,
+        )
+
+    if variables.PINATA_JWT:
+        yield Pinata(
+            variables.PINATA_JWT,
+            timeout=variables.HTTP_REQUEST_TIMEOUT_IPFS,
         )
 
     yield DummyIPFSProvider()  # FIXME: Remove after migration.

src/providers/ipfs/__init__.py

@@ -2,4 +2,5 @@
 from .dummy import *
 from .gw3 import *
 from .multi import *
+from .pinata import *
 from .types import *

src/providers/ipfs/pinata.py

@@ -0,0 +1,46 @@
+import requests
+
+from .cid import CIDv0, CIDv1, is_cid_v0
+from .types import FetchError, IPFSProvider, PinError, UploadError
+
+
+class Pinata(IPFSProvider):
+    """pinata.cloud IPFS provider"""
+
+    API_ENDPOINT = "https://api.pinata.cloud"
+    GATEWAY = "https://gateway.pinata.cloud"
+
+    def __init__(self, jwt_token: str, *, timeout: int) -> None:
+        super().__init__()
+        self.timeout = timeout
+        self.session = requests.Session()
+        self.session.headers["Authorization"] = f"Bearer {jwt_token}"
+
+    def fetch(self, cid: CIDv0 | CIDv1) -> bytes:
+        url = f"{self.GATEWAY}/ipfs/{cid}"
+        try:
+            resp = requests.get(url, timeout=self.timeout)
+            resp.raise_for_status()
+        except requests.RequestException as ex:
+            raise FetchError(cid) from ex
+        return resp.content
+
+    def upload(self, content: bytes, name: str | None = None) -> CIDv0 | CIDv1:
+        url = f"{self.API_ENDPOINT}/pinning/pinFileToIPFS"
+        try:
+            with self.session as s:
+                resp = s.post(url, files={"file": content})
+                resp.raise_for_status()
+        except requests.RequestException as ex:
+            raise UploadError from ex
+        cid = resp.json()["IpfsHash"]
+        return CIDv0(cid) if is_cid_v0(cid) else CIDv1(cid)
+
+    def pin(self, cid: CIDv0 | CIDv1) -> None:
+        url = f"{self.API_ENDPOINT}/pinning/pinByHash"
+        try:
+            with self.session as s:
+                resp = s.post(url, json={"hashToPin": str(cid)})
+                resp.raise_for_status()
+        except requests.RequestException as ex:
+            raise PinError(cid) from ex

src/variables.py

@@ -8,6 +8,7 @@
 KEYS_API_URI = os.getenv('KEYS_API_URI', '').split(',')
 GW3_ACCESS_KEY = os.getenv('GW3_ACCESS_KEY')
 GW3_SECRET_KEY = os.getenv('GW3_SECRET_KEY')
+PINATA_JWT = os.getenv('PINATA_JWT')
 
 # - Account -
 ACCOUNT = None