added a trino connector

.env.example

@@ -29,6 +29,14 @@ TWILIO_ACCOUNT_SID=
 TWILIO_API_SID=
 TWILIO_API_SECRET=
 
+#########
+# MYSQL #
+#########
+MYSQL_ROOT_PASSWORD=
+MYSQL_DATABASE=
+MYSQL_USER=
+MYSQL_PASSWORD=
+
 ###########
 # URLSCAN #
 ###########
@@ -57,4 +65,12 @@ SPLUNK_VERIFY_SSL=false
 PSQL_DSN=DSN=PostgresLocal;
 PSQL_USER=testuser
 PSQL_PASS=testpass
-PSQL_DB=test_db
+PSQL_DB=test_db
+
+TRINO_HOST=localhost
+TRINO_USER=youruser
+TRINO_PORT=8080
+#optional
+TRINO_CATALOG=test
+TRINO_SCHEMA=test
+TRINO_PASS=yourpassword

.gitignore

@@ -6,6 +6,9 @@ __pycache__/
 # C extensions
 *.so
 
+# Trino
+dev_env/etc/catalog/*.properties
+
 # Distribution / packaging
 .Python
 build/

dev_env/etc/catalog/mysql.properties.example

@@ -0,0 +1,4 @@
+connector.name=mysql
+connection-url=jdbc:mysql://mysql-db:3306/target_db
+connection-user=
+connection-password=

dev_env/etc/catalog/pg.properties.example

@@ -0,0 +1,4 @@
+connector.name=postgresql
+connection-url=jdbc:postgresql://postgres-db:5432/target_db
+connection-user=
+connection-password=

dev_env/postgres/test_postgres.py

@@ -3,9 +3,8 @@
 from typing import Dict, Any
 
 env_config: Dict[str, Any] = combine_env_configs()
-
 logger = setup_logger("pg logger")
-with PostgresConnector(conn_str=env_config["PGSQL_DSN"], logger=logger) as conn:
+with PostgresConnector(conn_str=env_config["PSQL_DSN"], logger=logger) as conn:
 
     # Optional insert test
     logger.info("inserting one row")

dev_env/trino/test_trino.py

@@ -0,0 +1,20 @@
+from pyapiary.dbms_connectors.trino import TrinoConnector
+from pyapiary.helpers import combine_env_configs
+from typing import Dict, Any
+
+env_config: Dict[str, Any] = combine_env_configs()
+
+
+client = TrinoConnector(
+    host=env_config["TRINO_HOST"],
+    port=env_config["TRINO_PORT"],
+    user=env_config["TRINO_USER"]
+)
+
+# Find
+print(client.query("SELECT * FROM pg.public.employees"))
+
+
+mylist = [{'name':'Pat','department':'Facilities MGMT'}]
+
+print(client.bulk_insert('pg.public.employees',mylist))

docker-compose.yml

@@ -9,6 +9,8 @@ services:
     environment:
       MONGO_INITDB_ROOT_USERNAME: ${MONGO_USER}
       MONGO_INITDB_ROOT_PASSWORD: ${MONGO_PASS}
+    networks:
+      - dev_lab
 
   elasticsearch:
     image: docker.elastic.co/elasticsearch/elasticsearch:7.17.10
@@ -20,6 +22,8 @@ services:
     ports:
       - "9200:9200"
     mem_limit: 512m
+    networks:
+      - dev_lab
 
   es-init:
     image: python:3.11-slim
@@ -31,6 +35,8 @@ services:
       - ./.env
     entrypoint: >
       sh -c "pip install --quiet requests python-dotenv && python /init.py"
+    networks:
+      - dev_lab
 
   splunk:
     image: vivekrsplunk/splunk:9.2.1-78803f08aabb-ef8f72f-arm64-ubuntu-20
@@ -46,6 +52,8 @@ services:
     volumes:
       - ./dev_env/splunk/init_app.sh:/init_app.sh
     entrypoint: [ "/bin/bash", "-c", "/init_app.sh & /sbin/entrypoint.sh start-service" ]
+    networks:
+      - dev_lab
 
   # both PG and ODBC connector use this data source, this is why dev_env/postgres does not have an init.sql
   odbc_db:
@@ -59,8 +67,37 @@ services:
       - "5432:5432"
     volumes:
       - ./dev_env/odbc/init.sql:/docker-entrypoint-initdb.d/init.sql:ro
+    networks:
+      - dev_lab
+
+  mysql_db:
+    image: mysql:latest
+    environment:
+      MYSQL_ROOT_PASSWORD: ${MYSQL_ROOT_PASSWORD}
+      MYSQL_DATABASE: ${MYSQL_TARGET_DB}
+      MYSQL_USER: ${MYSQL_USER}
+      MYSQL_PASSWORD: ${MYSQL_PASSWORD}
+    ports:
+      - "3306:3306"
+    networks:
+      - dev_lab
 
   proxy:
     image: ubuntu/squid:latest
     ports:
       - "3128:3128"
+    networks:
+      - dev_lab
+
+  trino:
+    image: trinodb/trino:latest
+    ports: 
+      - "8080:8080"
+    volumes:
+      - ./etc/catalog:/etc/trino/catalog
+    networks:
+      - dev_lab
+
+networks:
+  dev_lab:
+    external: true

pyproject.toml

@@ -20,6 +20,7 @@ pyodbc = "^5.3.0"
 psycopg-pool = {extras = ["binary", "pool"], version = "^3.3.1"}
 psycopg = {extras = ["pool"], version = "^3.3.4"}
 psycopg-binary = "^3.3.4"
+trino = "^0.337.0"
 
 [tool.poetry.extras]
 odbc = ["pyodbc"]

src/pyapiary/dbms_connectors/trino.py

@@ -0,0 +1,32 @@
+from trino.dbapi import connect
+from typing import List, Dict, Any
+
+class TrinoConnector:
+    def __init__(self, host, port, user, catalog=None, schema=None):
+        self.conn = connect(
+            host=host,
+            port=port,
+            user=user,
+            catalog=catalog,
+            schema=schema
+        )
+
+    def query(self, query_str):
+        with self.conn.cursor() as cur:
+            cur.execute(query_str)
+            if cur.description:
+                return cur.fetchall()
+            else:
+                return None
+
+    def bulk_insert(self, table, data : List[Dict[str,Any]]):
+        if not len(data)>0:
+            print('Invalid List of Dict type passed. Must have more than one element.')
+            return None
+        columns = data[0].keys()
+        placeholders = ", ".join(["?"] * len(columns))
+        query = f"INSERT INTO {table} ({', '.join(columns)}) VALUES ({placeholders})"
+        values = [tuple(row[col] for col in columns) for row in data]
+        with self.conn.cursor() as cur:
+            cur.executemany(query, values)
+            return True