minor fixes.

Author: faridseifi

src/python_activator/api.py

@@ -11,23 +11,17 @@
 import typer
 from python_activator.manifest import *
 
+
 class knowledge_object:
-    def __init__(self, name, status, function ,id):
-        self.id= id
+    def __init__(self, name, status, function, id):
+        self.id = id
         self.name = name
         self.status = status
         self.function = function
 
-class CustomEncoder(json.JSONEncoder):
-    def default(self, obj):
-        if isinstance(obj, knowledge_object):
-            return {'name': obj.name, 'status': obj.status, 'id': obj.id}
-        return super().default(obj)
 
 app = FastAPI()
-Knowledge_Objects = (
-    {}
-)
+Knowledge_Objects = {}
 
 # this dictionary used to keep loaded objects (key:is and value:function)
 object_directory = ""  # used for location of knowledge objects
@@ -39,9 +33,8 @@ def hello():
 
 
 @app.get("/endpoints")
-def hello():
-    return json.loads(json.dumps(Knowledge_Objects,cls=CustomEncoder, indent=4))
-
+def endpoints():
+    return Knowledge_Objects  # json.loads(json.dumps(Knowledge_Objects,cls=CustomEncoder, indent=4))
 
 
 # end point to expose all packages
@@ -52,7 +45,7 @@ async def execute_endpoint(
     # get the method
     if endpoint_key in Knowledge_Objects:
         function = Knowledge_Objects[endpoint_key].function
-        if function and Knowledge_Objects[endpoint_key].status=="Activated":
+        if function and Knowledge_Objects[endpoint_key].status == "Activated":
             # run the imported method and pass the request json
             # if content_type == 'application/json':
             data = await request.json()
@@ -62,7 +55,7 @@ async def execute_endpoint(
             result = function(data)
 
             return {"result": result}
-        elif Knowledge_Objects[endpoint_key].status!="Activated":
+        elif Knowledge_Objects[endpoint_key].status != "Activated":
             return {"result": Knowledge_Objects[endpoint_key].status}
         else:
             return {"result": "Knowledge object not found!"}
@@ -86,41 +79,43 @@ def install_requirements(modulepath):
 
 # look into the main directory that has all the packages and have the python ones installed
 def install_packages_from_directory(directory, manifest: dict):
-    for ko in manifest:        
+    for ko in manifest:
         install_module(directory, manifest[ko])
     list_installed_packages()
 
 
-def install_module(
-    directory, ko
-):  # TO DO: test how it works for windows installation
-    Knowledge_Objects[ko.name] = knowledge_object(ko.name,ko.status,None,"")
+def install_module(directory, ko):  # TO DO: test how it works for windows installation
+    Knowledge_Objects[ko.name] = knowledge_object(ko.name, ko.status, None, "")
 
     try:
         modulepath = directory + ko.name + "/"
 
         #########delete me: temporarily ignoring execute package
         if ko.name == "python-executive-v1.0":
             return
-        
-        if ko.status!="Ready for install":
-            return
 
+        if ko.status != "Ready for install":
+            return
 
         Knowledge_Objects[ko.name].status = "Activating"
-        
+
         # get metadata and deployment files
         with open(modulepath + "deployment.yaml", "r") as file:
             deployment_data = yaml.safe_load(file)
         with open(modulepath + "metadata.json", "r") as file:
             metadata = json.load(file)
         first_key = next(iter(deployment_data))
         second_key = next(iter(deployment_data[first_key]))
-        
+
         # do not install non python packages
         if deployment_data[first_key][second_key]["engine"] != "python":
             del Knowledge_Objects[ko.name]
-            Knowledge_Objects[metadata["@id"]] = knowledge_object(ko.name,"Knowledge object is not activated. It is not a python object.",None,metadata["@id"])
+            Knowledge_Objects[metadata["@id"]] = knowledge_object(
+                ko.name,
+                "Knowledge object is not activated. It is not a python object.",
+                None,
+                metadata["@id"],
+            )
             return
 
         # install requirements
@@ -140,17 +135,28 @@ def install_module(
         )
         mymethod = getattr(module, deployment_data[first_key][second_key]["function"])
         del Knowledge_Objects[ko.name]
-        Knowledge_Objects[metadata["@id"]] = knowledge_object(ko.name,"Activated",mymethod,metadata["@id"])
+        Knowledge_Objects[metadata["@id"]] = knowledge_object(
+            ko.name, "Activated", mymethod, metadata["@id"]
+        )
     except Exception as e:
-        Knowledge_Objects[ko.name].status="Faield activating with error: "+ repr(e)
+        Knowledge_Objects[ko.name].status = "Faield activating with error: " + repr(e)
+
 
 def list_installed_packages():
     print("-------------------\nPackages installed:")
-    #keys_list = list(KnowledgeObjects.keys())
-    print("{:<4}. {:<30} {:<30} {:<30}".format("","ID","NAME","STATUS"))
-    for i,item in  enumerate(Knowledge_Objects):
-        print("{:<4}. {:<30} {:<30} {:<30}".format(str(i),Knowledge_Objects[item].id,Knowledge_Objects[item].name[:30], Knowledge_Objects[item].status[:50]))
-    
+    # keys_list = list(KnowledgeObjects.keys())
+    #print(Knowledge_Objects)
+    print("{:<4}. {:<30} {:<30} {:<30}".format("", "ID", "NAME", "STATUS"))
+    for i, item in enumerate(Knowledge_Objects):
+        print(
+            "{:<4}. {:<30} {:<30} {:<30}".format(
+                str(i),
+                Knowledge_Objects[item].id,
+                Knowledge_Objects[item].name[:30],
+                Knowledge_Objects[item].status[:50],
+            )
+        )
+
     print("-------------------")
 
 
@@ -161,17 +167,20 @@ async def startup_event():
     object_directory = os.environ["COLLECTION_PATH"]
     manifest = process_manifest(object_directory)
     install_packages_from_directory(object_directory, manifest)
-    del os.environ["COLLECTION_PATH"]
-    del os.environ["MANIFEST_PATH"]
+    try:
+        del os.environ["COLLECTION_PATH"]
+        del os.environ["MANIFEST_PATH"]
+    except Exception as e:
+        print("error deleting env variables")
 
 
 # run virtual server when running this .py file directly for debugging. It will look for objects at {code folder}/pyshelf
 if __name__ == "__main__":
     print(">>>>>running with debug<<<<<")
-    os.environ["MANIFEST_PATH"]="/home/faridsei/dev/test/package/manifest.json"
-    #os.environ[
+    # os.environ["MANIFEST_PATH"]="/home/faridsei/dev/test/package/manifest.json"
+    # os.environ[
     #    "MANIFEST_PATH"
-    #] = "https://github.com/kgrid-objects/example-collection/releases/download/4.2.1/manifest.json"
+    # ] = "https://github.com/kgrid-objects/example-collection/releases/download/4.2.1/manifest.json"
     os.environ["COLLECTION_PATH"] = "/home/faridsei/dev/test/pyshelf/"
 
     uvicorn.run(app, host="127.0.0.1", port=8000)

src/python_activator/cli.py

@@ -8,15 +8,11 @@
 @cli.command()
 def run(collection_path: str = "", manifest_path: str = ""):
     object_directory = set_object_directory(collection_path)
-    #manifest_path=set_manifest_path(manifest_path)
+   
     if manifest_path:
         os.environ["MANIFEST_PATH"]=manifest_path
     os.environ["COLLECTION_PATH"]=object_directory
 
-    #manifest = process_manifest(object_directory)
-
-    #install_packages_from_directory(object_directory, manifest)
-
     uvicorn.run(app, host="127.0.0.1", port=8000)
 
 
@@ -42,14 +38,7 @@ def set_object_directory(collection_path: str) -> str:
         object_directory += "/"
     return object_directory
 
-def set_manifest_path(manifest_path: str) -> str:
-    if manifest_path:
-        return manifest_path
-    elif os.environ.get("MANIFEST_PATH"):
-        return os.environ.get("MANIFEST_PATH")
-    else:
-        return manifest_path
-    
+   
 
 @cli.command()
 def version():

src/python_activator/manifest.py

@@ -20,7 +20,7 @@ def process_manifest(object_directory: str)->dict:
     # 0. if no manifest provided consider list of existing knowledge object folders as manifest
     if not manifest_path:
         for item in scanned_directories:
-            output_manifest[ko_name]=str.replace(item, ".zip", "")
+            output_manifest[str.replace(item, ".zip", "")]=ko_object(str.replace(item, ".zip", ""),"Ready for install")
         return output_manifest
 
     input_manifest, scanned_files = init_process(manifest_path, object_directory)
@@ -41,6 +41,7 @@ def process_manifest(object_directory: str)->dict:
             continue
 
         # 2. if zip file does not exists in the folder download it from local path or http (uses uri)
+        
         if not ko_filename in scanned_files:
             try:
                 urllib.request.urlretrieve(

src/test/test_fastapi_exceptions.py

@@ -0,0 +1,35 @@
+from fastapi import FastAPI, HTTPException,Request
+from fastapi.responses import JSONResponse
+import uvicorn
+
+class UnicornException(Exception):
+    def __init__(self, name: str):
+        self.name = name
+
+app = FastAPI()
+
+items = {"foo": "The Foo Wrestlers"}
+
+@app.exception_handler(UnicornException)
+async def unicorn_exception_handler(request: Request, exc: UnicornException):
+    return JSONResponse(
+        status_code=418,
+        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
+    )
+
+@app.get("/unicorns/{name}")
+async def read_unicorn(name: str):
+    if name == "yolo":
+        raise UnicornException(name=name)
+    return {"unicorn_name": name}
+
+@app.get("/items/{item_id}")
+async def read_item(item_id: str):
+    if item_id not in items:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return {"item": items[item_id]}
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="127.0.0.1", port=8000)
+    
+    

tests/test_api.py

@@ -1,9 +1,10 @@
 import pytest
 from src.python_activator.api import hello,execute_endpoint
-
+#from fastapi.testclient import TestClient
+from src.test.test_fastapi_exceptions import app  # Assuming your FastAPI app instance is named `app`
 def test_root():
 
-	expected = {"Hello": "World"}
+	expected = {"Hello": "World9"}
 
 	# Act
 	output = hello()
@@ -21,6 +22,11 @@ def test_hard_coded_root_nessage_doesnt_know_Farid():
 	# Assert
 	assert output != expected
 
- 
+def test_how_exceptions_are_handled_in_apis():
+    #client = TestClient(app)
+    #response = client.get("/item/foo")
+    #assert response.status_code == 200
+    #assert response.json() == {"item": "The Foo Wrestlers"}
+    assert 1!=1
 
 

tests/test_cli.py

@@ -1,4 +1,5 @@
 from src.python_activator.api import install_module
+from src.python_activator.manifest import ko_object
 import os
 import sys
 import importlib
@@ -9,13 +10,13 @@ def test_basics():
     assert os.path.abspath("etc/pyshelf").endswith("etc/pyshelf")
 
     directory="/home/faridsei/dev/code/python-activator/etc/pyshelf/"
-    sys.path.append(        directory    )
+    #sys.path.append(        directory    )
 
-    install_module(directory, "python-multiartifact-v1-0")
-    del sys.modules["ko_folder"]
-    sys.path.remove(
-        directory
-    )
+    install_module(directory, ko_object("python-multiartifact-v1-0","Ready to install"))
+    #del sys.modules["ko_folder"]
+    #sys.path.remove(
+    #    directory
+    #)
 
 def test_direct_import_of_module():
 

tests/test_dot_dirs.py

@@ -0,0 +1,33 @@
+import pytest
+from src.python_activator.api import *
+
+def test_process_local_manifest():
+	os.environ["MANIFEST_PATH"]="/home/faridsei/dev/test/package/manifest.json"
+	process_manifest("/home/faridsei/dev/test/pyshelf/")
+	del os.environ["MANIFEST_PATH"]
+
+
+	# Assert
+	assert 1==1
+	
+def test_process_internet_manifest():
+	os.environ["MANIFEST_PATH"]="https://github.com/kgrid-objects/example-collection/releases/download/4.2.1/manifest.json"
+	process_manifest("/home/faridsei/dev/test/pyshelf/")
+	del os.environ["MANIFEST_PATH"]
+
+	# Assert
+	assert 1==1
+ 
+def test_process_no_manifest():
+	
+	process_manifest("/home/faridsei/dev/test/pyshelf/")
+
+	# Assert
+	assert 1==1
+
+ 
+def test_using_uri_to_download_zip():
+    assert 1!=1
+    
+def test_formatting_dictionary_of_object_to_json():
+    assert 1!=1