From 56a5c49943e0ef177095c4819ade5422146bf8fa Mon Sep 17 00:00:00 2001 From: Guy Korland Date: Wed, 20 Dec 2023 12:32:58 +0200 Subject: [PATCH] format code --- drivers/driver.py | 19 +++++++------------ drivers/falkordb.py | 22 +++++++++------------- drivers/neo4j.py | 18 ++++++------------ main.py | 2 +- 4 files changed, 23 insertions(+), 38 deletions(-) diff --git a/drivers/driver.py b/drivers/driver.py index 53f334d..3d36aa5 100644 --- a/drivers/driver.py +++ b/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,7 +14,7 @@ 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. @@ -26,14 +22,13 @@ def get_graph_history(self, skip: int, per_page: int) -> dict[str, Any]: 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 \ No newline at end of file + pass diff --git a/drivers/falkordb.py b/drivers/falkordb.py index d4a4f66..04e5ed3 100644 --- a/drivers/falkordb.py +++ b/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)} \ No newline at end of file + return {"error": str(e)} diff --git a/drivers/neo4j.py b/drivers/neo4j.py index 84a2437..154a120 100644 --- a/drivers/neo4j.py +++ b/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)} \ No newline at end of file + return {"error": str(e)} diff --git a/main.py b/main.py index 1098cb9..b190eb6 100644 --- a/main.py +++ b/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")