def get_table_names(self, connection, schema=None, **kw):
schema = self.denormalize_name(schema or self.default_schema_name)
# note that table_names() isn't loading DBLINKed or synonym'ed tables
if schema is None:
schema = self.default_schema_name
sql_str = "SELECT table_name FROM all_tables WHERE "
if self.exclude_tablespaces:
sql_str += (
"nvl(tablespace_name, 'no tablespace') "
"NOT IN (%s) AND " % (
', '.join(["'%s'" % ts for ts in self.exclude_tablespaces])
)
)
sql_str += (
"OWNER = :owner "
"AND IOT_NAME IS NULL "
"AND DURATION IS NULL")
cursor = connection.execute(sql.text(sql_str), owner=schema)
return [self.normalize_name(row[0]) for row in cursor]
评论列表
文章目录