Move utility functions into own module

Author: waxlamp

multinet/api.py

@@ -1,46 +1,19 @@
 """Flask blueprint for Multinet REST API."""
-import json
-
-from flask import Blueprint, request, Response
+from flask import Blueprint, request
 from webargs import fields
 from webargs.flaskparser import use_kwargs
 
-from . import db
+from . import db, util
 from .errors import ValidationFailed
 
 bp = Blueprint("multinet", __name__)
-
-
-def generate(iterator):
-    """Return a generator that yields an iterator's contents into a JSON list."""
-    yield "["
-
-    comma = ""
-    for row in iterator:
-        yield f"{comma}{json.dumps(row)}"
-        comma = ","
-
-    yield "]"
-
-
-def stream(iterator):
-    """Convert an iterator to a Flask response."""
-    return Response(generate(iterator), mimetype="application/json")
-
-
-def require_db():
-    """Check if the db is live."""
-    if not db.check_db():
-        return ("", "500 Database Not Live")
-
-
-bp.before_request(require_db)
+bp.before_request(util.require_db)
 
 
 @bp.route("/workspaces", methods=["GET"])
 def get_workspaces():
     """Retrieve list of workspaces."""
-    return stream(db.get_workspaces())
+    return util.stream(db.get_workspaces())
 
 
 @bp.route("/workspaces/<workspace>", methods=["GET"])
@@ -54,22 +27,22 @@ def get_workspace(workspace):
 def get_workspace_tables(workspace, type="all"):
     """Retrieve the tables of a single workspace."""
     tables = db.workspace_tables(workspace, type)
-    return stream(tables)
+    return util.stream(tables)
 
 
 @bp.route("/workspaces/<workspace>/tables/<table>", methods=["GET"])
 @use_kwargs({"offset": fields.Int(), "limit": fields.Int()})
 def get_table_rows(workspace, table, offset=0, limit=30):
     """Retrieve the rows and headers of a table."""
     rows = db.workspace_table(workspace, table, offset, limit)
-    return stream(rows)
+    return util.stream(rows)
 
 
 @bp.route("/workspaces/<workspace>/graphs", methods=["GET"])
 def get_workspace_graphs(workspace):
     """Retrieve the graphs of a single workspace."""
     graphs = db.workspace_graphs(workspace)
-    return stream(graphs)
+    return util.stream(graphs)
 
 
 @bp.route("/workspaces/<workspace>/graphs/<graph>", methods=["GET"])
@@ -121,7 +94,7 @@ def aql(workspace):
         return (query, "400 Malformed Request Body")
 
     result = db.aql_query(workspace, query)
-    return stream(result)
+    return util.stream(result)
 
 
 @bp.route("/workspaces/<workspace>", methods=["DELETE"])

multinet/errors.py

@@ -119,5 +119,13 @@ def __init__(self, errors):
         self.errors = errors
 
     def flask_response(self):
-        """Generate a 400 error for the bad argument."""
+        """Generate a 400 error."""
         return ({"errors": self.errors}, "400 Validation Failed")
+
+
+class DatabaseNotLive(ServerError):
+    """Exception for when Arango database is not live."""
+
+    def flask_response(self):
+        """Generate a 500 error."""
+        return ("", "500 Database Not Live")

multinet/uploaders/csv.py

@@ -3,13 +3,13 @@
 from io import StringIO
 import re
 
-from .. import db, api
+from .. import db, util
 
 from flask import Blueprint, request
 from flask import current_app as app
 
 bp = Blueprint("csv", __name__)
-bp.before_request(api.require_db)
+bp.before_request(util.require_db)
 
 
 def validate_csv(rows):

multinet/uploaders/nested_json.py

@@ -2,12 +2,12 @@
 import itertools
 import json
 
-from .. import db, api
+from .. import db, util
 
 from flask import Blueprint, request
 
 bp = Blueprint("nested_json", __name__)
-bp.before_request(api.require_db)
+bp.before_request(util.require_db)
 
 
 def analyze_nested_json(data, int_table_name, leaf_table_name):

multinet/uploaders/newick.py

@@ -2,13 +2,13 @@
 import uuid
 import newick
 
-from .. import db, api
+from .. import db, util
 
 from flask import Blueprint, request
 from flask import current_app as app
 
 bp = Blueprint("newick", __name__)
-bp.before_request(api.require_db)
+bp.before_request(util.require_db)
 
 
 @bp.route("/<workspace>/<table>", methods=["POST"])

multinet/util.py

@@ -0,0 +1,30 @@
+"""Utility functions."""
+import json
+
+from flask import Response
+
+from . import db
+from .errors import DatabaseNotLive
+
+
+def generate(iterator):
+    """Return a generator that yields an iterator's contents into a JSON list."""
+    yield "["
+
+    comma = ""
+    for row in iterator:
+        yield f"{comma}{json.dumps(row)}"
+        comma = ","
+
+    yield "]"
+
+
+def stream(iterator):
+    """Convert an iterator to a Flask response."""
+    return Response(generate(iterator), mimetype="application/json")
+
+
+def require_db():
+    """Check if the db is live."""
+    if not db.check_db():
+        raise DatabaseNotLive()