Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@classmethod
def factory(cls, ns):
logging.debug("Relational Db Factory entered")
explorer = None
if ns.connection_type == "mysql":
explorer = MySQLExplorer(ns)
elif ns.connection_type == "postgres" or ns.connection_type == "redshift":
explorer = PostgreSQLExplorer(ns)
elif ns.connection_type == "oracle":
explorer = OracleExplorer(ns)
assert explorer is not None
return explorer
class MySQLExplorer(RelDbExplorer):
_catalog_query = """
SELECT
TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, DATA_TYPE
FROM
INFORMATION_SCHEMA.COLUMNS
WHERE
TABLE_SCHEMA NOT IN ('information_schema', 'performance_schema', 'sys', 'mysql')
AND DATA_TYPE RLIKE 'char.*|varchar.*|text'
ORDER BY table_schema, table_name, column_name
"""
_sample_query_template = (
"select {column_list} from {schema_name}.{table_name} limit 10"
)
def __init__(self, ns):
def _get_catalog_query(self):
return self._catalog_query
@classmethod
def _get_sample_query(cls, schema_name, table_name, column_list):
return cls._sample_query_template.format(
column_list='"{0}"'.format(
'","'.join(col.get_name() for col in column_list)
),
schema_name=schema_name.get_name(),
table_name=table_name.get_name(),
)
class PostgreSQLExplorer(RelDbExplorer):
_catalog_query = """
SELECT
TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, DATA_TYPE
FROM
INFORMATION_SCHEMA.COLUMNS
WHERE
TABLE_SCHEMA NOT IN ('information_schema', 'pg_catalog')
AND DATA_TYPE SIMILAR TO '%char%|%text%'
ORDER BY table_schema, table_name, column_name
"""
_sample_query_template = "select {column_list} from {schema_name}.{table_name} TABLESAMPLE BERNOULLI (10)"
def __init__(self, ns):
super(PostgreSQLExplorer, self).__init__(ns)
self._pg_database = (
def __init__(self, ns):
super(RelDbExplorer, self).__init__(ns)
self.host = ns.host
self.user = ns.user
self.password = ns.password
self.port = (
int(ns.port)
if "port" in vars(ns) and ns.port is not None
else self.default_port
)
def _get_catalog_query(self):
return self._catalog_query
@classmethod
def _get_sample_query(cls, schema_name, table_name, column_list):
return cls._sample_query_template.format(
column_list='"{0}"'.format(
'","'.join(col.get_name() for col in column_list)
),
schema_name=schema_name.get_name(),
table_name=table_name.get_name(),
)
class OracleExplorer(RelDbExplorer):
_catalog_query = """
SELECT
'{db}', TABLE_NAME, COLUMN_NAME
FROM
USER_TAB_COLUMNS
WHERE UPPER(DATA_TYPE) LIKE '%CHAR%'
ORDER BY TABLE_NAME, COLUMN_ID
"""
_sample_query_template = "select {column_list} from {table_name} sample(5)"
_select_query_template = "select {column_list} from {table_name}"
_count_query = "select count(*) from {table_name}"
def __init__(self, ns):
super(OracleExplorer, self).__init__(ns)
self._oracle_database = ns.database