Repository URL to install this package:
|
Version:
3.0.0.dev3 ▾
|
from typing import Any
from snsql.sql.reader.base import NameCompare
from snsql.sql.reader.engine import Engine
from snsql.sql.reader.mysql import MySqlNameCompare
from snsql.sql.reader.postgres import PostgresNameCompare
from snsql.sql.reader.sql_server import SqlServerNameCompare
from .big_query import BigQueryNameCompare # type: ignore
class SQLEngine(Engine): # pylint: disable=too-few-public-methods
"""Supported engines"""
BIG_QUERY = "BigQuery"
known_engines = {
Engine.POSTGRES,
Engine.SQL_SERVER,
BIG_QUERY,
Engine.MYSQL,
}
class_map = {
Engine.POSTGRES: "postgres",
Engine.SQL_SERVER: "sql_server",
BIG_QUERY: "big_query",
Engine.MYSQL: "my_sql",
}
@staticmethod
def name_compare(engine: str) -> NameCompare:
if engine == "Postgres":
return PostgresNameCompare()
if engine == "SqlServer":
return SqlServerNameCompare()
if engine == "BigQuery":
return BigQueryNameCompare()
if engine == "MySql":
return MySqlNameCompare()
return NameCompare()
@staticmethod
def from_uri(uri: str) -> Any:
if uri.startswith("mssql"):
return SQLEngine.SQL_SERVER
if uri.startswith("bigquery"):
return SQLEngine.BIG_QUERY
if uri.startswith("mysql"):
return SQLEngine.MYSQL
return SQLEngine.POSTGRES