feat: UploadBlob

Author: xlassix

src/lighthouseweb3/__init__.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 
 import os
+import io
 from typing import List
 from .functions import upload as d, types as t, deal_status, get_uploads as getUploads
 
@@ -25,6 +26,18 @@ def upload(self, source: str) -> t.Upload:
         except Exception as e:
             raise e
 
+    def uploadBlob(self, source: io.BufferedReader, filename: str) -> t.Upload:
+        """
+        Upload Blob a file or directory to the Lighthouse.
+
+        :param source: str, path to file or directory
+        :return: t.Upload, the upload result
+        """
+        try:
+            return d.uploadBlob(source, filename, self.token)
+        except Exception as e:
+            raise e
+
     @staticmethod
     def getDealStatus(cid: str) -> List[t.DealData]:
         """

src/lighthouseweb3/functions/axios.py

@@ -61,14 +61,14 @@ def post_files(
             raise e
 
     def post_blob(
-        self, file: BufferedReader, headers: Dict[str, str] = None, **kwargs
+        self, file: BufferedReader, filename: str, headers: Dict[str, str] = None, **kwargs
     ) -> dict | Exception:
         try:
             self.parse_url_query(kwargs.get("query", None))
             files = [(
                 "file",
                 (
-                    utils.extract_file_name(file.name),
+                    utils.extract_file_name(filename),
                     file.read(),
                     "application/octet-stream",
                 ),

src/lighthouseweb3/functions/upload.py

@@ -42,7 +42,30 @@ def upload(source: str | BufferedReader | NamedBufferedReader, token: str) -> t.
                 file_dict["path"] = source
             return {"data": axios.post_files(file_dict, headers)}
         else:
-            return {"data": axios.post_blob(source, headers)}
+            return {"data": axios.post_blob(source, source.name, headers)}
+    except Exception as e:
+        print(e)
+        raise e
+
+
+def uploadBlob(source:  BufferedReader, filename: str, token: str) -> t.Upload:
+    """
+    Upload a Buffer or readable Object
+    @params {source}: str, path to file or directory
+    @params {token}: str, lighthouse api token
+    """
+    # create headers
+    headers = {
+        "Authorization": f"Bearer {token}",
+        # "Content-Type": "multipart/form-data",
+        "Encryption": "false",
+        "Mime-Type": "application/octet-stream",
+    }
+    try:
+        # create http object
+        axios = Axios(Config.lighthouse_node + "/api/v0/add")
+        # create list of files to upload
+        return {"data": axios.post_blob(source, filename, headers)}
     except Exception as e:
         print(e)
         raise e

tests/test_upload.py

@@ -5,6 +5,13 @@
 from src.lighthouseweb3 import Lighthouse
 from src.lighthouseweb3.functions.utils import NamedBufferedReader
 from .setup import parse_env
+import string
+import secrets
+
+
+def generate_random_string(length: int) -> str:
+    characters = string.ascii_letters + string.digits
+    return ''.join(secrets.choice(characters) for _ in range(length))
 
 
 class TestUpload(unittest.TestCase):
@@ -36,8 +43,8 @@ def test_Upload_dir(self):
     def test_Upload_Blob(self):
         """test Upload function"""
         l = Lighthouse(os.environ.get("LIGHTHOUSE_TOKEN"))
-        res = l.upload(NamedBufferedReader(
-            io.BytesIO(b"tests/testdir/"), "hwh.txt"))
+        res = l.uploadBlob(
+            io.BytesIO(b"tests/testdir/"), f"{generate_random_string(16)}.txt")
         self.assertNotEqual(res.get("data"), None, "data is None")
         self.assertIsInstance(res.get("data"), dict, "data is a dict")
         self.assertNotEqual(res.get("data").get("Hash"), None, "data is None")