Method get_table_names

lib/sqlalchemy/dialects/oracle/base.py:2578–2650

Supported kw arguments are: ``dblink`` to reflect via a db link.

(self, connection, schema=None, dblink=None, **kw)

    @reflection.cache
    def get_table_names(self, connection, schema=None, dblink=None, **kw):
        """Supported kw arguments are: ``dblink`` to reflect via a db link."""
        # note that table_names() isn't loading DBLINKed or synonym'ed tables
        if schema is None:
            schema = self.default_schema_name

        den_schema = self.denormalize_schema_name(schema)
        if kw.get("oracle_resolve_synonyms", False):
            tables = (
                select(
                    dictionary.all_tables.c.table_name,
                    dictionary.all_tables.c.owner,
                    dictionary.all_tables.c.iot_name,
                    dictionary.all_tables.c.duration,
                    dictionary.all_tables.c.tablespace_name,
                )
                .union_all(
                    select(
                        dictionary.all_synonyms.c.synonym_name.label(
                            "table_name"
                        ),
                        dictionary.all_synonyms.c.owner,
                        dictionary.all_tables.c.iot_name,
                        dictionary.all_tables.c.duration,
                        dictionary.all_tables.c.tablespace_name,
                    )
                    .select_from(dictionary.all_tables)
                    .join(
                        dictionary.all_synonyms,
                        and_(
                            dictionary.all_tables.c.table_name
                            == dictionary.all_synonyms.c.table_name,
                            dictionary.all_tables.c.owner
                            == func.coalesce(
                                dictionary.all_synonyms.c.table_owner,
                                dictionary.all_synonyms.c.owner,
                            ),
                        ),
                    )
                )
                .subquery("available_tables")
            )
        else:
            tables = dictionary.all_tables

        query = select(tables.c.table_name)
        if self.exclude_tablespaces:
            query = query.where(
                func.coalesce(
                    tables.c.tablespace_name, "no tablespace"
                ).not_in(self.exclude_tablespaces)
            )
        query = query.where(
            tables.c.owner == den_schema,
            tables.c.iot_name.is_(null()),
            tables.c.duration.is_(null()),
        )