feat: public 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, Pinata
+from src.providers.ipfs import DummyIPFSProvider, GW3, IPFSProvider, MultiIPFSProvider, Pinata, PublicIPFS
 from src.typings import OracleModule
 from src.utils.build import get_build_info
 from src.web3py.extensions import (
@@ -156,6 +156,8 @@ def ipfs_providers() -> Iterable[IPFSProvider]:
             timeout=variables.HTTP_REQUEST_TIMEOUT_IPFS,
         )
 
+    yield PublicIPFS(timeout=variables.HTTP_REQUEST_TIMEOUT_IPFS)
+
     yield DummyIPFSProvider()  # FIXME: Remove after migration.
 
 

src/providers/ipfs/__init__.py

@@ -3,4 +3,5 @@
 from .gw3 import *
 from .multi import *
 from .pinata import *
+from .public import *
 from .types import *

src/providers/ipfs/public.py

@@ -0,0 +1,34 @@
+import logging
+import requests
+
+from .types import FetchError, IPFSProvider, PinError, UploadError
+from .cid import CIDv0, CIDv1
+
+
+logger = logging.getLogger(__name__)
+
+
+class PublicIPFS(IPFSProvider):
+    """Public IPFS gateway (fetch-only provider)"""
+
+    GATEWAY = "https://ipfs.io"
+
+    def __init__(self, *, timeout: int) -> None:
+        super().__init__()
+        self.timeout = timeout
+
+    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:
+            logger.error({"msg": "Request has been failed", "error": str(ex)})
+            raise FetchError(cid) from ex
+        return resp.content
+
+    def upload(self, content: bytes, name: str | None = None) -> CIDv0 | CIDv1:
+        raise UploadError
+
+    def pin(self, cid: CIDv0 | CIDv1) -> None:
+        raise PinError(cid)