[WIP] Add trino Connection Support

great_expectations_provider/operators/great_expectations.py

@@ -304,7 +304,19 @@ def make_connection_configuration(self) -> Dict[str, str]:
             else:
                 uri_string = f"awsathena+rest://@athena.{region}.amazonaws.com/?s3_staging_dir={s3_path}"
             # TODO: Add other AWS sources here as needed
-        # TODO: Add and Trino support (if possible)
+        elif conn_type == "trino":
+            catalog = self.conn.extra_dejson.get("catalog", "hive")
+            if self.conn.schema:
+                catalog += f"/{self.conn.schema}"
+            if self.conn.password:
+                uri_string = f"trino://{self.conn.login}:{self.conn.password}@{self.conn.host}:{self.conn.port}/{catalog}"  # noqa
+            elif self.conn.extra_dejson.get("auth") == "jwt":
+                uri_string = f"trino://{self.conn.login}@{self.conn.host}:{self.conn.port}/{catalog}?access_token={self.conn.extra_dejson['jwt__token']}"  # noqa
+            # TODO: Add other auth methods as needed
+            elif self.conn.extra_dejson["auth"] in ("kerberos", "certs"):
+                raise NotImplementedError("kerberos and cert auth methods are not implemented.")  # noqa
+            else:
+                raise ValueError("Unrecognized Trino Auth method.")
         else:
             raise ValueError(f"Conn type: {conn_type} is not supported.")
         return {"connection_string": uri_string}

tests/operators/test_great_expectations.py

@@ -972,6 +972,48 @@ def test_great_expectations_operator__make_connection_string_athena_without_db()
     assert operator.make_connection_configuration() == test_conn_conf
 
 
+def test_great_expectations_operator__make_connection_string_trino_basic_auth():
+    test_conn_conf = {"connection_string": "trino://user:pass@host:8080/hive"}
+    operator = GreatExpectationsOperator(
+        task_id="task_id",
+        data_context_config=in_memory_data_context_config,
+        data_asset_name="table_name",
+        conn_id="trino_default",
+        expectation_suite_name="suite",
+    )
+    operator.conn = Connection(
+        conn_id="trino_default",
+        conn_type="trino",
+        host="host",
+        port=8080,
+        login="user",
+        password="pass",
+    )
+    operator.conn_type = operator.conn.conn_type
+    assert operator.make_connection_configuration() == test_conn_conf
+
+
+def test_great_expectations_operator__make_connection_string_trino_jwt_auth():
+    test_conn_conf = {"connection_string": "trino://user@host:8080/tpch?access_token=TEST_JWT_TOKEN"}
+    operator = GreatExpectationsOperator(
+        task_id="task_id",
+        data_context_config=in_memory_data_context_config,
+        data_asset_name="table_name",
+        conn_id="trino_default",
+        expectation_suite_name="suite",
+    )
+    operator.conn = Connection(
+        conn_id="trino_default",
+        conn_type="trino",
+        host="host",
+        port=8080,
+        login="user",
+        extra={"catalog": "tpch", "auth": "jwt", "jwt__token": "TEST_JWT_TOKEN"},
+    )
+    operator.conn_type = operator.conn.conn_type
+    assert operator.make_connection_configuration() == test_conn_conf
+
+
 def test_great_expectations_operator__make_connection_string_schema_parameter(mocker):
     test_conn_conf = {
         "url": URL.create(