Merge pull request #404 from multinet-app/aql-table-creation

.env.default

@@ -21,3 +21,5 @@ ALLOWED_ORIGINS=http://localhost:8080
 ARANGO_HOST=localhost
 ARANGO_PORT=8529
 ARANGO_PROTOCOL=http
+
+ARANGO_READONLY_PASSWORD=letmein

.github/workflows/main.yml

@@ -50,3 +50,4 @@ jobs:
           ARANGO_HOST: ${{ secrets.CI_ARANGO_HOST }}
           ARANGO_PASSWORD: ${{ secrets.CI_ARANGO_PASSWORD }}
           ARANGO_PROTOCOL: ${{ secrets.CI_ARANGO_PROTOCOL }}
+          ARANGO_READONLY_PASSWORD: ${{ secrets.CI_ARANGO_READONLY_PASSWORD }}

devops/arangodb.yml

@@ -9,6 +9,9 @@
     - name: arangodb_root_password
       prompt: "Password for ArangoDB root account"
 
+    - name: arango_readonly_password
+      prompt: "Password for ArangoDB readonly user"
+
     - name: ssl_cert_path
       prompt: "Path to parent SSL cert directory"
       default: "/etc/letsencrypt/live/multinet.app/"
@@ -92,3 +95,19 @@
         name: arangodb3
         state: restarted
       become: true
+
+    - name: Install arangosh script for user provisioning to server
+      template:
+        src: create-readonly-user.js
+        dest: /tmp/create-readonly-user.js
+        owner: ubuntu
+        mode: 444
+
+    - name: Ensure there is a read-only user
+      command: arangosh --server.password {{ arangodb_root_password }} --server.endpoint ssl://127.0.0.1:8529 --javascript.execute /tmp/create-readonly-user.js
+      become: true
+
+    - name: Delete user provisioning script
+      file:
+        path: /tmp/create-readonly-user.js
+        state: absent

devops/create-readonly-user.js

@@ -0,0 +1,7 @@
+const users = require('@arangodb/users');
+
+if (!users.exists('readonly')) {
+    users.save('readonly', '{{ arango_readonly_password }}');
+}
+
+users.grantDatabase('readonly', '*', 'ro');

multinet/api.py

@@ -54,6 +54,17 @@ def get_workspace_tables(workspace: str, type: TableType = "all") -> Any:  # noq
     return util.stream(tables)
 
 
+@bp.route("/workspaces/<workspace>/tables", methods=["POST"])
+@use_kwargs({"table": fields.Str()})
+@swag_from("swagger/workspace_aql_tables.yaml")
+def create_aql_table(workspace: str, table: str) -> Any:
+    """Create a table from an AQL query."""
+    aql = request.data.decode()
+    table = db.create_aql_table(workspace, table, aql)
+
+    return table
+
+
 @bp.route("/workspaces/<workspace>/tables/<table>", methods=["GET"])
 @require_reader
 @use_kwargs({"offset": fields.Int(), "limit": fields.Int()})

multinet/db.py

@@ -6,15 +6,23 @@
 from arango.graph import Graph
 from arango.database import StandardDatabase
 from arango.collection import StandardCollection
-
-from arango.exceptions import DatabaseCreateError, EdgeDefinitionCreateError
+from arango.aql import AQL
+from arango.cursor import Cursor
+
+from arango.exceptions import (
+    DatabaseCreateError,
+    EdgeDefinitionCreateError,
+    AQLQueryValidateError,
+    AQLQueryExecuteError,
+)
 from requests.exceptions import ConnectionError
 
 from typing import Any, List, Dict, Set, Generator, Optional
 from typing_extensions import TypedDict
 from multinet.types import EdgeDirection, TableType, Workspace, WorkspaceDocument
 from multinet.auth.types import User
 from multinet.errors import InternalServerError
+from multinet.uploaders.csv import validate_csv
 from multinet import util
 
 from multinet.errors import (
@@ -25,6 +33,8 @@
     NodeNotFound,
     AlreadyExists,
     GraphCreationError,
+    AQLExecutionError,
+    AQLValidationError,
     DatabaseCorrupted,
 )
 
@@ -45,7 +55,20 @@
 def db(name: str) -> StandardDatabase:
     """Return a handle for Arango database `name`."""
     return arango.db(
-        name, username="root", password=os.environ.get("ARANGO_PASSWORD", "letmein")
+        name,
+        username="root",
+        password=os.environ.get("ARANGO_PASSWORD", "letmein"),
+        verify=True,
+    )
+
+
+def read_only_db(name: str) -> StandardDatabase:
+    """Return a read-only handle for the Arango database `name`."""
+    return arango.db(
+        name,
+        username="readonly",
+        password=os.environ.get("ARANGO_READONLY_PASSWORD", "letmein"),
+        verify=True,
     )
 
 
@@ -208,13 +231,14 @@ def get_workspace_metadata(name: str) -> Workspace:
 
 # Caches the reference to the StandardDatabase instance for each workspace
 @lru_cache()
-def get_workspace_db(name: str) -> StandardDatabase:
+def get_workspace_db(name: str, readonly: bool = False) -> StandardDatabase:
     """Return the Arango database associated with a workspace, if it exists."""
     doc = workspace_mapping(name)
     if not doc:
         raise WorkspaceNotFound(name)
 
-    return db(doc["internal"])
+    name = doc["internal"]
+    return read_only_db(name) if readonly else db(name)
 
 
 def get_graph_collection(workspace: str, graph: str) -> Graph:
@@ -328,6 +352,25 @@ def workspace_table_keys(
     return keys
 
 
+def create_aql_table(workspace: str, name: str, aql: str) -> str:
+    """Create a new table from an AQL query."""
+    db = get_workspace_db(workspace, readonly=True)
+
+    if db.has_collection(name):
+        raise AlreadyExists("table", name)
+
+    # In the future, the result of this validation can be
+    # used to determine dependencies in virtual tables
+    rows = list(_run_aql_query(db.aql, aql))
+    validate_csv(rows, "_key", False)
+
+    db = get_workspace_db(workspace, readonly=False)
+    coll = db.create_collection(name, sync=True)
+    coll.insert_many(rows)
+
+    return name
+
+
 def graph_node(workspace: str, graph: str, table: str, node: str) -> dict:
     """Return the data associated with a particular node in a graph."""
     space = get_workspace_db(workspace)
@@ -410,14 +453,24 @@ def delete_table(workspace: str, table: str) -> str:
     return table
 
 
-def aql_query(workspace: str, query: str) -> Generator[Any, None, None]:
-    """Perform an AQL query in the given workspace."""
-    aql = get_workspace_db(workspace).aql
+def _run_aql_query(aql: AQL, query: str) -> Cursor:
+    try:
+        aql.validate(query)
+        cursor = aql.execute(query)
+    except AQLQueryValidateError as e:
+        raise AQLValidationError(str(e))
+    except AQLQueryExecuteError as e:
+        raise AQLExecutionError(str(e))
 
-    cursor = aql.execute(query)
     return cursor
 
 
+def aql_query(workspace: str, query: str) -> Cursor:
+    """Perform an AQL query in the given workspace."""
+    aql = get_workspace_db(workspace, readonly=True).aql
+    return _run_aql_query(aql, query)
+
+
 def create_graph(
     workspace: str,
     graph: str,

multinet/errors.py

@@ -28,9 +28,13 @@ def flask_response(self) -> FlaskTuple:
 class InternalServerError(ServerError):
     """General exception for internal server errors."""
 
+    def __init__(self, message: str = ""):
+        """Initialize the exception."""
+        self.message = message
+
     def flask_response(self) -> FlaskTuple:
         """Generate a 500 level error."""
-        return ("", "500 Internal Server Error")
+        return (self.message, "500 Internal Server Error")
 
 
 class DatabaseCorrupted(ServerError):
@@ -214,3 +218,27 @@ def __init__(self, message: str):
     def flask_response(self) -> FlaskTuple:
         """Generate a 500 error."""
         return (self.message, "500 Graph Creation Failed")
+
+
+class AQLValidationError(ServerError):
+    """Exception for errors when validating an aql query in Arango."""
+
+    def __init__(self, message: str = ""):
+        """Initialize error message."""
+        self.message = message
+
+    def flask_response(self) -> FlaskTuple:
+        """Generate a 400 error."""
+        return (self.message, "400 AQL Validation Failed")
+
+
+class AQLExecutionError(ServerError):
+    """Exception for errors when executing an aql query in Arango."""
+
+    def __init__(self, message: str = ""):
+        """Initialize error message."""
+        self.message = message
+
+    def flask_response(self) -> FlaskTuple:
+        """Generate a 400 error."""
+        return (self.message, "400 Error during AQL Execution")

multinet/swagger/template.yaml

@@ -135,6 +135,15 @@ parameters:
       type: string
       example: workspace3
 
+  aql:
+    name: aql
+    in: body
+    description: An AQL query
+    required: true
+    schema:
+      type: string
+      example: FOR member IN members RETURN member
+
   table:
     name: table
     in: path

multinet/swagger/workspace_aql_tables.yaml

@@ -0,0 +1,34 @@
+Create a table from an AQL query
+---
+parameters:
+  - $ref: "#/parameters/workspace"
+  - name: table
+    in: query
+    description: Name of table to create
+    required: true
+    schema:
+      type: string
+      example: table4
+
+  - $ref: "#/parameters/aql"
+
+responses:
+  200:
+    description: The name of the table created
+    schema:
+      type: string
+
+  400:
+    description: Malformed AQL
+
+  401:
+    description: Insufficient permissions to perform the desired AQL query
+
+  404:
+    description: Specified workspace could not be found
+    schema:
+      type: string
+      example: workspace_that_doesnt_exist
+
+tags:
+  - table

mypy_stubs/arango/__init__.pyi

@@ -2,4 +2,10 @@ from typing import Any
 
 class ArangoClient:
     def __init__(self, host: str, port: int, protocol: str): ...
-    def db(self, database: str, username: str, password: str) -> Any: ...
+    def db(
+        self,
+        name: str = "_system",
+        username: str = "root",
+        password: str = "",
+        verify: bool = False,
+    ) -> Any: ...

mypy_stubs/arango/aql.pyi

@@ -0,0 +1,40 @@
+from typing import Dict, Optional, Union
+from arango.connection import Connection  # type: ignore
+from arango.executor import Executor  # type: ignore
+from arango.cursor import Cursor
+
+class AQL:
+    """AQL (ArangoDB Query Language) API wrapper.
+
+    :param connection: HTTP connection.
+    :type connection: arango.connection.Connection
+    :param executor: API executor.
+    :type executor: arango.executor.Executor
+    """
+
+    def __init__(self, connection: Connection, executor: Executor): ...
+    def validate(self, query: str) -> Dict: ...
+    def execute(
+        self,
+        query: str,
+        count: bool = False,
+        batch_size: Optional[int] = None,
+        ttl: Optional[int] = None,
+        bind_vars: Optional[Dict] = None,
+        full_count: Optional[bool] = None,
+        max_plans: Optional[int] = None,
+        optimizer_rules: Optional[str] = None,
+        cache: Optional[bool] = None,
+        memory_limit: int = 0,
+        fail_on_warning: Optional[bool] = None,
+        profile: Optional[bool] = None,
+        max_transaction_size: Optional[int] = None,
+        max_warning_count: Optional[int] = None,
+        intermediate_commit_count: Optional[int] = None,
+        intermediate_commit_size: Optional[int] = None,
+        satellite_sync_wait: Optional[Union[int, float]] = None,
+        read_collections: Optional[str] = None,
+        write_collections: Optional[str] = None,
+        stream: Optional[bool] = None,
+        skip_inaccessible_cols: Optional[bool] = None,
+    ) -> Cursor: ...

mypy_stubs/arango/cursor.pyi

@@ -0,0 +1,3 @@
+from typing import Any, Generator
+
+Cursor = Generator[Any, None, None]

mypy_stubs/arango/exceptions.pyi

@@ -1,2 +1,4 @@
 class DatabaseCreateError(Exception): ...
 class EdgeDefinitionCreateError(Exception): ...
+class AQLQueryValidateError(Exception): ...
+class AQLQueryExecuteError(Exception): ...