format code

drivers/driver.py

@@ -3,12 +3,8 @@
 
 
 class Driver(ABC):
-
     @abstractmethod
-    def get_graph_data(self) -> tuple[
-        list[dict[str, Any]], 
-        list[dict[str, Any]]
-    ]:
+    def get_graph_data(self) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
         """
         Abstract method to get graph data.
         """
@@ -18,22 +14,21 @@ def get_graph_data(self) -> tuple[
     def get_graph_history(self, skip: int, per_page: int) -> dict[str, Any]:
         """
         Abstract method to get graph history.
-        
+
         :param skip: The number of items to skip.
         :param per_page: The number of items per page.
         :return: The graph history data.
         """
         pass
 
     @abstractmethod
-    def get_response_data(self, response_data: Any) -> tuple[
-        list[dict[str, Any]], 
-        list[dict[str, Any]]
-    ]:
+    def get_response_data(
+        self, response_data: Any
+    ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
         """
         Abstract method to process response data.
-        
+
         :param response_data: The response data to process.
         :return: The processed response data.
         """
-        pass
+        pass

drivers/falkordb.py

@@ -7,9 +7,7 @@
 
 
 class FalkorDB(Driver):
-
     def __init__(self):
-
         url = os.environ.get("FALKORDB_URL", "redis://localhost:6379")
 
         self.driver = FalkorDBDriver.from_url(url).select_graph("falkordb")
@@ -46,9 +44,7 @@ def get_graph_data(self) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
 
         return (nodes.result_set, edges.result_set)
 
-
     def get_graph_history(self, skip, per_page) -> dict[str, Any]:
-
         # Getting the total number of graphs
         result = self.driver.query(
             """
@@ -61,7 +57,7 @@ def get_graph_history(self, skip, per_page) -> dict[str, Any]:
 
         # if total_count == 0:
         #     return {"graph_history": [], "remaining": 0, "graph": False}
-        
+
         # Fetching 10 most recent graphs
         result = self.driver.query(
             """
@@ -76,15 +72,16 @@ def get_graph_history(self, skip, per_page) -> dict[str, Any]:
         )
 
         # Process the 'result' to format it as a list of graphs
-        graph_history = [FalkorDB._process_graph_data(record) for record in result.result_set]
+        graph_history = [
+            FalkorDB._process_graph_data(record) for record in result.result_set
+        ]
         remaining = max(0, total_count - skip - per_page)
 
         return {"graph_history": graph_history, "remaining": remaining, "graph": True}
-
-    def get_response_data(self, response_data)-> tuple[
-        list[dict[str, Any]], 
-        list[dict[str, Any]]
-    ]:
+
+    def get_response_data(
+        self, response_data
+    ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
         # Import nodes
         nodes = self.driver.query(
             """
@@ -109,7 +106,6 @@ def get_response_data(self, response_data)-> tuple[
         )
         return (nodes.result_set, relationships.result_set)
 
-
     @staticmethod
     def _process_graph_data(record) -> dict[str, Any]:
         """
@@ -132,4 +128,4 @@ def _process_graph_data(record) -> dict[str, Any]:
 
             return graph_data
         except Exception as e:
-            return {"error": str(e)}
+            return {"error": str(e)}

drivers/neo4j.py

@@ -7,7 +7,6 @@
 
 
 class Neo4j(Driver):
-
     def __init__(self):
         # If Neo4j credentials are set, then Neo4j is used to store information
         username = os.environ.get("NEO4J_USERNAME")
@@ -19,9 +18,7 @@ def __init__(self):
                 print("Obsolete: Please define NEO4J_URI instead")
 
         if username and password and url:
-            self.driver = GraphDatabase.driver(url, 
-                                          auth=(username, 
-                                                password))
+            self.driver = GraphDatabase.driver(url, auth=(username, password))
             # Check if connection is successful
             with self.driver.session() as session:
                 try:
@@ -57,7 +54,6 @@ def get_graph_data(self) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
 
         return (nodes, edges)
 
-
     def get_graph_history(self, skip, per_page) -> dict[str, Any]:
         # Getting the total number of graphs
         total_graphs, _, _ = self.driver.execute_query(
@@ -86,11 +82,10 @@ def get_graph_history(self, skip, per_page) -> dict[str, Any]:
         remaining = max(0, total_count - skip - per_page)
 
         return {"graph_history": graph_history, "remaining": remaining, "graph": True}
-
-    def get_response_data(self, response_data)-> tuple[
-        list[dict[str, Any]], 
-        list[dict[str, Any]]
-    ]:
+
+    def get_response_data(
+        self, response_data
+    ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
         # Import nodes
         nodes = self.driver.execute_query(
             """
@@ -115,7 +110,6 @@ def get_response_data(self, response_data)-> tuple[
         )
         return (nodes, relationships)
 
-
     @staticmethod
     def _process_graph_data(record):
         """
@@ -138,4 +132,4 @@ def _process_graph_data(record):
 
             return graph_data
         except Exception as e:
-            return {"error": str(e)}
+            return {"error": str(e)}

main.py

@@ -150,7 +150,6 @@ def _restore(e):
             results = driver.get_response_data(response_data)
             print("Results from Graph:", results)
 
-
     except Exception as e:
         print("An error occurred during the Graph operation:", e)
         return (
@@ -245,6 +244,7 @@ def get_graph_history():
     except Exception as e:
         return jsonify({"error": str(e), "graph": driver is not None}), 500
 
+
 @app.route("/")
 def index():
     return render_template("index.html")