Metabase and Snowflake Fixed (#1788)

* Metabase and Snowflake Fixed

* Removed DBT

ingestion/examples/airflow/dags/airflow_sample_dbt.py

@@ -1,86 +0,0 @@
-#  Copyright 2021 Collate
-#  Licensed under the Apache License, Version 2.0 (the "License");
-#  you may not use this file except in compliance with the License.
-#  You may obtain a copy of the License at
-#  http://www.apache.org/licenses/LICENSE-2.0
-#  Unless required by applicable law or agreed to in writing, software
-#  distributed under the License is distributed on an "AS IS" BASIS,
-#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-#  See the License for the specific language governing permissions and
-#  limitations under the License.
-
-import json
-import pathlib
-from datetime import timedelta
-
-from airflow import DAG
-
-try:
-    from airflow.operators.python import PythonOperator
-except ModuleNotFoundError:
-    from airflow.operators.python_operator import PythonOperator
-
-from airflow.utils.dates import days_ago
-
-from metadata.config.common import load_config_file
-from metadata.ingestion.api.workflow import Workflow
-
-default_args = {
-    "owner": "user_name",
-    "email": ["username@org.com"],
-    "email_on_failure": False,
-    "retries": 3,
-    "retry_delay": timedelta(seconds=10),
-    "execution_timeout": timedelta(minutes=60),
-}
-
-config = """
-{
-  "source": {
-    "type": "dbt",
-    "config": {
-      "service_name": "bigquery_dbt",
-      "service_type": "BigQuery",
-      "catalog_file": "./examples/sample_data/dbt/catalog.json",
-      "manifest_file": "./examples/sample_data/dbt/manifest.json",
-      "run_results_file": "./examples/sample_data/dbt/run_results.json",
-      "database": "shopify"
-    }
-  },
-  "sink": {
-    "type": "metadata-rest",
-    "config": {}
-  },
-  "metadata_server": {
-    "type": "metadata-server",
-    "config": {
-      "api_endpoint": "http://localhost:8585/api",
-      "auth_provider_type": "no-auth"
-    }
-  }
-}
-"""
-
-
-def metadata_ingestion_workflow():
-    workflow_config = json.loads(config)
-    workflow = Workflow.create(workflow_config)
-    workflow.execute()
-    workflow.raise_from_status()
-    workflow.print_status()
-    workflow.stop()
-
-
-with DAG(
-    "sample_dbt",
-    default_args=default_args,
-    description="An example DAG which runs a OpenMetadata ingestion workflow",
-    schedule_interval=timedelta(days=1),
-    start_date=days_ago(1),
-    is_paused_upon_creation=False,
-    catchup=False,
-) as dag:
-    ingest_task = PythonOperator(
-        task_id="ingest_using_recipe",
-        python_callable=metadata_ingestion_workflow,
-    )

ingestion/examples/workflows/dbt.json

@@ -1,24 +0,0 @@
-{
-  "source": {
-    "type": "dbt",
-    "config": {
-      "service_name": "bigquery",
-      "service_type": "BigQuery",
-      "catalog_file": "./examples/sample_data/dbt/catalog.json",
-      "manifest_file": "./examples/sample_data/dbt/manifest.json",
-      "run_results_file": "./examples/sample_data/dbt/run_results.json",
-      "database": "shopify"
-    }
-  },
-  "sink": {
-    "type": "metadata-rest",
-    "config": {}
-  },
-  "metadata_server": {
-    "type": "metadata-server",
-    "config": {
-      "api_endpoint": "http://localhost:8585/api",
-      "auth_provider_type": "no-auth"
-    }
-  }
-}

ingestion/src/metadata/ingestion/source/metabase.py

@@ -91,7 +91,7 @@ class MetabaseSource(Source[Entity]):
         self.metabase_session = {"X-Metabase-Session": session_id}
         self.dashboard_service = get_dashboard_service_or_create(
             config.service_name,
-            DashboardServiceType.Looker.name,
+            DashboardServiceType.Metabase.name,
             config.username,
             config.password.get_secret_value(),
             config.host_port,

ingestion/src/metadata/ingestion/source/snowflake.py

@@ -26,16 +26,14 @@ class SnowflakeConfig(SQLConnectionConfig):
     scheme = "snowflake"
     account: str
     database: Optional[str]
-    schema: Optional[str]
     warehouse: Optional[str]
     role: Optional[str]
     duration: Optional[int]
     service_type = "Snowflake"
+    connect_args: Optional[dict]
 
     def get_connection_url(self):
         connect_string = super().get_connection_url()
-        if self.schema and self.database:
-            connect_string += f"/{self.schema}"
         options = {
             "account": self.account,
             "warehouse": self.warehouse,
@@ -43,7 +41,7 @@ class SnowflakeConfig(SQLConnectionConfig):
         }
         params = "&".join(f"{key}={value}" for (key, value) in options.items() if value)
         if params:
-            connect_string = f"{connect_string}?{params}"
+            connect_string = f"{connect_string}{params}"
         return connect_string
 
 

ingestion/src/metadata/ingestion/source/sql_source.py

@@ -164,7 +164,11 @@ class SQLSource(Source[OMetaDatabaseAndTable]):
         self.status = SQLSourceStatus()
         self.sql_config = self.config
         self.connection_string = self.sql_config.get_connection_url()
-        self.engine = create_engine(self.connection_string, **self.sql_config.options)
+        self.engine = create_engine(
+            self.connection_string,
+            **self.sql_config.options,
+            connect_args=self.sql_config.connect_args,
+        )
         self.connection = self.engine.connect()
         self.data_profiler = None
         self.data_models = {}