feat: support sqlalchemy engine kwargs (#517)

sqllineage/core/metadata/sqlalchemy.py

@@ -1,7 +1,7 @@
 import logging
-from typing import List
+from typing import Any, Dict, List, Optional
 
-from sqlalchemy import MetaData, Table, create_engine, inspect, make_url
+from sqlalchemy import MetaData, Table, create_engine, make_url
 from sqlalchemy.exc import NoSuchModuleError, NoSuchTableError, OperationalError
 
 from sqllineage.core.metadata_provider import MetaDataProvider
@@ -16,11 +16,17 @@ class SQLAlchemyMetaDataProvider(MetaDataProvider):
     SQLAlchemyMetaDataProvider queries metadata from database using SQLAlchemy
     """
 
-    def __init__(self, url: str):
+    def __init__(self, url: str, engine_kwargs: Optional[Dict[str, Any]] = None):
+        """
+        :param url: sqlalchemy url
+        :param engine_kwargs: a dictionary of keyword arguments that will be passed to sqlalchemy create_engine
+        """
         super().__init__()
         self.metadata_obj = MetaData()
         try:
-            self.engine = create_engine(url)
+            if engine_kwargs is None:
+                engine_kwargs = {}
+            self.engine = create_engine(url, **engine_kwargs)
         except NoSuchModuleError as e:
             u = make_url(url)
             raise MetaDataProviderException(
@@ -33,14 +39,16 @@ def __init__(self, url: str):
 
     def _get_table_columns(self, schema: str, table: str, **kwargs) -> List[str]:
         columns = []
-        if inspect(self.engine).has_schema(schema):
-            try:
-                sqlalchemy_table = Table(
-                    table, self.metadata_obj, schema=schema, autoload_with=self.engine
-                )
-                columns = [c.name for c in sqlalchemy_table.columns]
-            except NoSuchTableError:
-                logger.warning("%s does not exist in database %s", table, schema)
-        else:
-            logger.warning("Schema %s does not exist", schema)
+        try:
+            sqlalchemy_table = Table(
+                table, self.metadata_obj, schema=schema, autoload_with=self.engine
+            )
+            columns = [c.name for c in sqlalchemy_table.columns]
+        except (NoSuchTableError, OperationalError):
+            logger.warning(
+                "error listing columns for table %s.%s in %s, return empty list instead",
+                schema,
+                table,
+                self.engine.url,
+            )
         return columns