support for readable Objects

Author: xlassix

src/lighthouseweb3/functions/upload.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+
+from io import BufferedReader
+from typing import Dict, List, Tuple
+from .axios import Axios
+from .utils import is_dir, walk_dir_tree, extract_file_name, NamedBufferedReader
+from .config import Config
+from . import types as t
+
+
+def upload(source: str | BufferedReader | NamedBufferedReader, token: str) -> t.Upload:
+    """
+    Deploy a file or directory to the lighthouse network
+    @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
+
+        if (isinstance(source, str)):
+            file_dict: t.FileDict = {}
+
+            # check if source is a directory
+            if is_dir(source):
+                # walk directory tree and add files to list
+                file_dict["files"], root = walk_dir_tree(source)
+                file_dict["is_dir"] = True
+                file_dict["path"] = root
+            else:
+                # add file to list
+                file_dict["files"] = [source]
+                file_dict["is_dir"] = False
+                file_dict["path"] = source
+            return {"data": axios.post_files(file_dict, headers)}
+        else:
+            return {"data": axios.post_blob(source, headers)}
+    except Exception as e:
+        print(e)
+        raise e

tests/test_upload.py

@@ -0,0 +1,57 @@
+#!/usr/bin/env python3
+import os
+import io
+import unittest
+from src.lighthouseweb3 import Lighthouse
+from src.lighthouseweb3.functions.utils import NamedBufferedReader
+from .setup import parse_env
+
+
+class TestUpload(unittest.TestCase):
+    def setUp(self) -> None:
+        """setup test environment"""
+        parse_env()
+
+    def test_env(self):
+        """test env var"""
+        self.assertNotEqual(
+            os.environ.get("LIGHTHOUSE_TOKEN"), None, "token is not None"
+        )
+
+    def test_Upload_file(self):
+        """test Upload function"""
+        l = Lighthouse()  # will use env var
+        res = l.upload("tests/testdir/testfile.txt")
+        self.assertNotEqual(res.get("data"), None, "data is None")
+        self.assertNotEqual(res.get("data").get("Hash"), None, "data is None")
+
+    def test_Upload_dir(self):
+        """test Upload function"""
+        l = Lighthouse(os.environ.get("LIGHTHOUSE_TOKEN"))
+        res = l.upload("tests/testdir/")
+        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")
+
+    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"))
+        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")
+
+    def test_Upload_File(self):
+        """test Upload function"""
+        l = Lighthouse(os.environ.get("LIGHTHOUSE_TOKEN"))
+        with open("./.gitignore", "rb") as file:
+            res = l.upload(file)
+            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")
+
+
+if __name__ == "__main__":
+    unittest.main()