feat: add support pinata.cloud and infura.io ipfs storage

utils/config.py

@@ -61,9 +61,9 @@ def get_deployer_account() -> Union[LocalAccount, Account]:
     return accounts.load(os.environ["DEPLOYER"]) if (is_live or "DEPLOYER" in os.environ) else accounts[4]
 
 
-def get_web3_storage_token() -> str:
+def get_web3_storage_token(silent = False) -> str:
     is_live = get_is_live()
-    if is_live and "WEB3_STORAGE_TOKEN" not in os.environ:
+    if is_live and not silent and "WEB3_STORAGE_TOKEN" not in os.environ:
         raise EnvironmentError(
             "Please set WEB3_STORAGE_TOKEN env variable to the web3.storage API token to be able to "
             "upload the vote description to IPFS by calling upload_vote_ipfs_description. Alternatively, "
@@ -72,6 +72,35 @@ def get_web3_storage_token() -> str:
 
     return os.environ["WEB3_STORAGE_TOKEN"] if (is_live or "WEB3_STORAGE_TOKEN" in os.environ) else ""
 
+def get_pinata_cloud_token(silent = False) -> str:
+    is_live = get_is_live()
+    if is_live and not silent and "PINATA_CLOUD_TOKEN" not in os.environ:
+        raise EnvironmentError(
+            "Please set PINATA_CLOUD_TOKEN env variable to the pinata.cloud API token to be able to "
+            "upload the vote description to IPFS by calling upload_vote_ipfs_description. Alternatively, "
+            "you can only calculate cid without uploading to IPFS by calling calculate_vote_ipfs_description"
+        )
+
+    return os.environ["PINATA_CLOUD_TOKEN"] if (is_live or "PINATA_CLOUD_TOKEN" in os.environ) else ""
+
+def get_infura_io_keys(silent = False) -> Tuple[str, str]:
+    is_live = get_is_live()
+    if is_live and not silent and (
+        "WEB3_INFURA_IPFS_PROJECT_ID" not in os.environ or "WEB3_INFURA_IPFS_PROJECT_SECRET" not in os.environ
+    ):
+        raise EnvironmentError(
+            "Please set WEB3_INFURA_IPFS_PROJECT_ID and WEB3_INFURA_IPFS_PROJECT_SECRET env variable "
+            "to the web3.storage api token"
+        )
+    project_id = (
+        os.environ["WEB3_INFURA_IPFS_PROJECT_ID"] if (is_live or "WEB3_INFURA_IPFS_PROJECT_ID" in os.environ) else ""
+    )
+    project_secret = (
+        os.environ["WEB3_INFURA_IPFS_PROJECT_SECRET"]
+        if (is_live or "WEB3_INFURA_IPFS_PROJECT_SECRET" in os.environ)
+        else ""
+    )
+    return project_id, project_secret
 
 def prompt_bool() -> Optional[bool]:
     choice = input().lower()
@@ -227,7 +256,7 @@ def dai_token(self) -> interface.ERC20:
     @property
     def usdt_token(self) -> interface.ERC20:
         return interface.ERC20(USDT_TOKEN)
-    
+
     @property
     def usdc_token(self) -> interface.ERC20:
         return interface.ERC20(USDC_TOKEN)

utils/ipfs.py

@@ -5,10 +5,11 @@
 import requests
 from typing import Tuple, TypedDict
 from os import linesep
+import json
 
 from ipfs_cid import cid_sha256_hash
 
-from utils.config import get_web3_storage_token
+from utils.config import get_pinata_cloud_token, get_infura_io_keys, get_web3_storage_token
 from utils.checksummed_address import checksum_verify
 
 #  https://github.com/multiformats/multibase/blob/master/multibase.csv
@@ -48,6 +49,41 @@ class IPFSUploadResult(TypedDict):
     messages: list[Tuple[str, str]]
 
 
+# alternative for upload_str_to_web3_storage
+def _upload_str_to_infura_io(text: str) -> str:
+    text_bytes = text.encode("utf-8")
+    text_file = io.BytesIO(text_bytes)
+    files = {"file": text_file}
+    (projectId, projectSecret) = get_infura_io_keys()
+
+    endpoint = "https://ipfs.infura.io:5001"
+
+    response = requests.post(endpoint + "/api/v0/add?cid-version=1", files=files, auth=(projectId, projectSecret))
+    response.raise_for_status()
+    response_json = response.json()
+
+    return response_json.get("Hash")
+
+
+# alternative for upload_str_to_web3_storage
+def _upload_str_to_pinata_cloud(text: str) -> str:
+    text_bytes = text.encode("utf-8")
+    text_file = io.BytesIO(text_bytes)
+    files = {"file": text_file}
+    pinata_cloud_token = get_pinata_cloud_keys()
+
+    endpoint = "https://api.pinata.cloud"
+
+    pinata_options = {"cidVersion": 1, "wrapWithDirectory": False}
+    payload = {"pinataOptions": json.dumps(pinata_options, separators=(",", ":"))}
+
+    response = requests.post(endpoint + "/pinning/pinFileToIPFS", data=payload, files=files, auth=pinata_cloud_token)
+    response.raise_for_status()
+    response_json = response.json()
+
+    return response_json.get("IpfsHash")
+
+
 # upload text to web3.storage ipfs
 def _upload_str_to_web3_storage(text: str) -> str:
     text_bytes = text.encode("utf-8")
@@ -65,6 +101,10 @@ def _upload_str_to_web3_storage(text: str) -> str:
 
 
 def _upload_str_to_ipfs(text: str) -> str:
+    if get_pinata_cloud_keys(silent = True):
+        return _upload_str_to_web3_storage(text)
+    if get_infura_io_keys(silent =True):
+        return _upload_str_to_infura_io(text)
     return _upload_str_to_web3_storage(text)
 
 
@@ -93,7 +133,6 @@ async def _fetch_cid_status_from_ipfs_async(cid: str) -> int:
 
     request_urls = [
         get_url_by_cid(cid),  # faster for uploaded files
-        f"https://api.web3.storage/status/{cid}",  # much faster for not uploaded files
     ]
 
     async with aiohttp.ClientSession() as session: