feat: refactored into python modules

Author: opensaucerer

.gitignore

@@ -50,6 +50,7 @@ coverage.xml
 .hypothesis/
 .pytest_cache/
 cover/
+*_test.txt
 
 # Translations
 *.mo

axios.py lighthouse/axios.pyaxios.py renamed to lighthouse/axios.py

@@ -13,8 +13,7 @@ def parse_url_params(self, params: Dict[str, str]):
                 for key, value in params.items():
                     self.url += f"&{key}={value}"
         except Exception as e:
-            print(e)
-            return e
+            raise e
 
     def get(self, headers: Dict[str, str] = None, **kwargs) -> dict | Exception:
         try:
@@ -23,8 +22,7 @@ def get(self, headers: Dict[str, str] = None, **kwargs) -> dict | Exception:
             r.raise_for_status()
             return r.json()
         except Exception as e:
-            print(e)
-            return e
+            raise e
 
     def post(
         self, body=None, headers: Dict[str, str] = None, **kwargs
@@ -35,17 +33,15 @@ def post(
             r.raise_for_status()
             return r.json()
         except Exception as e:
-            print(e)
-            return e
+            raise e
 
     def read_files(self, files: List[str]) -> List[BufferedReader]:
         file_list = []
         for file in files:
-            with open(file, "rb") as f:
-                file_list.append(f)
+            file_list.append(open(file, "rb"))
         return file_list
 
-    def post_file(
+    def post_files(
         self, file: List[str], headers: Dict[str, str] = None, **kwargs
     ) -> dict | Exception:
         try:
@@ -55,5 +51,4 @@ def post_file(
             r.raise_for_status()
             return r.json()
         except Exception as e:
-            print(e)
-            return e
+            raise e

config.py lighthouse/config.pyconfig.py renamed to lighthouse/config.py

@@ -1,19 +1,21 @@
-from axios import Axios
-from utils import is_dir, walk_dir_tree
-from config import Config
+from typing import List
+from .axios import Axios
+from .utils import is_dir, walk_dir_tree
+from .config import Config
+from . import types as t
 
 
-def deploy(source: str, token: str):
+def deploy(source: str, token: str) -> t.Deploy:
     """
     Deploy a file or directory to the lighthouse network
     @params {source}: str, path to file or directory
     @params {token}: str, lighthouse api token
     """
     try:
         # create http object
-        axios = Axios(Config.lighthouse_node)
+        axios = Axios(Config.lighthouse_node + "/api/v0/add")
         # create list of files to upload
-        file_list = []
+        file_list: List[str] = []
         # check if source is a directory
         if is_dir(source):
             # walk directory tree and add files to list
@@ -23,18 +25,13 @@ def deploy(source: str, token: str):
             file_list.append(source)
 
         # create headers
-        """
-        "Content-type": `multipart/form-data; boundary= ${data._boundary}`,
-        "Encryption": false,
-        "Mime-Type": mimeType,
-        """
         headers = {
             "Authorization": f"Bearer {token}",
             "Content-Type": "multipart/form-data",
             "Encryption": "false",
             "Mime-Type": "application/octet-stream",
         }
         # upload files
-        axios.post_file(file_list, headers)
+        return axios.post_files(file_list, headers)
     except Exception as e:
-        print(e)
+        raise e

deploy.py lighthouse/deploy.pydeploy.py renamed to lighthouse/deploy.py

@@ -0,0 +1,9 @@
+from dataclasses import dataclass
+from typing import Dict, NewType, List, Tuple, TypedDict
+
+
+@dataclass
+class Deploy(TypedDict):
+    """typings for deploy function
+    """
+    data: dict

lighthouse/types.py

@@ -0,0 +1,19 @@
+#!/usr/bin/env python3
+
+import lighthouse.deploy as d
+from lighthouse import types as t
+
+
+class Lighthouse:
+    def __init__(self, token: str):
+        self.token = token
+
+    def deploy(self, source: str) -> t.Deploy:
+        """
+        Deploy a file or directory to the lighthouse network
+        @params {source}: str, path to file or directory
+        """
+        try:
+            return d.deploy(source, self.token)
+        except Exception as e:
+            raise e

utils.py lighthouse/utils.pyutils.py renamed to lighthouse/utils.py

@@ -0,0 +1,30 @@
+#!/usr/bin/env python3
+
+import os
+from main import Lighthouse
+
+
+def run_test():
+    """setup test environment and run tests"""
+    parse_env()
+    test_deploy()
+
+
+def test_deploy():
+    """test deploy function"""
+    l = Lighthouse(os.environ["LH_TOKEN"])
+    assert l.deploy("tests/upload_test.txt") == {"data": {}}
+
+
+def parse_env():
+    """parse .env file"""
+    with open(".env", "r") as f:
+        for line in f.readlines():
+            if line.startswith("#"):
+                continue
+            key, value = line.split("=")
+            os.environ[key] = value.strip()
+
+
+if __name__ == "__main__":
+    run_test()