Dosql (#32)

connectors/connector-postgresql/README.md

@@ -10,9 +10,10 @@ database_host: str
 database_port: int
 database_user: str
 database_password: str
-table_name: str
 ```
 
+All commands except for `DoSQL` require `table_name: str` as well.
+
 ### CreateTable
 
 Additional parameters:
@@ -219,3 +220,22 @@ For example:
   ]
 }
 ```
+
+### DoSQL
+
+Additional parameters:
+
+```
+schema: Dict[str, Any]
+```
+
+Performs a SQL statement of your chosing.
+
+The schema parameter expects:
+
+| Key | Value |
+|-----|-------|
+| `sql` | The SQL to `do`. Uses `%s` for variable bindings. |
+| `values` | (optional) A list of values to bind. |
+| `fetch_results` | (options) Bool to indicate if a list of results should be returned. |
+

connectors/connector-postgresql/connector_postgresql/baseCommand.py

@@ -25,8 +25,7 @@ class BaseCommand:
             status = 200
         except Exception as e:
             status = 500
-            # TODO: better error message, e has no reason and str repr contains quotes
-            response = '{"error": "Error executing sql statement"}'
+            response = f'{"error": "Error executing sql statement: {e}"}'
         finally:
             conn.close()
 

connectors/connector-postgresql/connector_postgresql/commands/doSQL.py

@@ -0,0 +1,42 @@
+import json
+
+from typing import Any
+from typing import Dict
+
+from psycopg2.extras import Json
+from psycopg2.extensions import register_adapter
+
+register_adapter(dict, Json)
+
+from connector_postgresql.baseCommand import BaseCommand, ConnectionConfig
+
+class DoSQL(BaseCommand):
+    """DoSQL."""
+
+    def __init__(self,
+        database_name: str,
+        database_host: str,
+        database_port: int,
+        database_user: str,
+        database_password: str,
+        schema: Dict[str, Any]
+    ):
+        """__init__."""
+        self.connection_config = ConnectionConfig(
+            database_name, 
+            database_host, 
+            database_port, 
+            database_user, 
+            database_password)
+        self.schema = schema
+
+    def execute(self, config, task_data):
+
+        sql = self.schema["sql"]
+        values = self.schema.get("values", [])
+        fetch_results = self.schema.get("fetch_results", False)
+
+        if fetch_results:
+            return self.fetchall(sql, self.connection_config, values)
+
+        return self.execute_query(sql, self.connection_config, values)