--- /dev/null
+.. change::
+ :tags: feature, reflection, performance
+ :tickets: 13311
+
+ Created new reflection method :meth:`_engine.Inspector.has_multi_table`
+ to check the existence of multiple tables at once, allowing for
+ better performance when checking many tables. Like the other "multi"
+ reflection methods, the default dialect offers a default implementation
+ that just call the single method in a loop. Backends that wish to take
+ advantage of this new method can implement it in their dialects.
+ The PostgreSQL, Oracle and SQL Server dialects have been updated to use
+ this new method.
+ The implementation of :meth:`_schema.MetaData.create_all` has been updated
+ to make use of this new method to check the existence of the tables,
+ reducing the number of round trips to the database when creating many tables.
else:
return self.schema_name
+ @reflection.cache
@_db_plus_owner
def has_table(self, connection, tablename, dbname, owner, schema, **kw):
self._ensure_has_table_connection(connection)
- return self._internal_has_table(connection, tablename, owner, **kw)
+ return self._internal_has_multi_table(
+ connection, [tablename], owner, schema, **kw
+ )[schema, tablename]
+
+ def has_multi_table(self, connection, table_names, schema=None, **kw):
+ # this does not really matche the signature of _db_plus_owner_multi
+ # so manually do the unpacking here
+ dbname, owner = _owner_plus_db(self, schema)
+ return _switch_db(
+ dbname,
+ connection,
+ self._internal_has_multi_table,
+ connection,
+ table_names,
+ owner,
+ schema,
+ **kw,
+ ).items()
@reflection.cache
@_db_plus_owner
view_names = [r[0] for r in connection.execute(s)]
return view_names
- @reflection.cache
- def _internal_has_table(self, connection, tablename, owner, **kw):
- if tablename.startswith("#"): # temporary table
- # mssql does not support temporary views
- # SQL Error [4103] [S0001]: "#v": Temporary views are not allowed
- return bool(
- connection.scalar(
- # U filters on user tables only.
- text("SELECT object_id(:table_name, 'U')"),
- {"table_name": f"tempdb.dbo.[{tablename}]"},
- )
- )
- else:
- tables = ischema.tables
-
- s = sql.select(tables.c.table_name).where(
- sql.and_(
- sql.or_(
- tables.c.table_type == "BASE TABLE",
- tables.c.table_type == "VIEW",
- ),
- tables.c.table_name == tablename,
- )
+ def _internal_has_multi_table(
+ self, connection, table_names, owner, schema, **kw
+ ):
+ regular_names, multi_object_names, temp_names = (
+ self._partition_filter_names(
+ connection, owner, table_names, ObjectScope.ANY, ObjectKind.ANY
)
+ )
- if owner:
- s = s.where(tables.c.table_schema == owner)
+ result = {}
+ if multi_object_names:
+ # in multi_object_names we already have the names that exist in
+ # the database, no need to do another query here
+ name_map = self._multi_name_map(regular_names)
+ for server_name in multi_object_names:
+ result[(schema, name_map(server_name))] = True
- c = connection.execute(s)
+ if temp_names:
+ # mssql does not support temporary views
+ # U filters on user tables only.
+ query = text("SELECT object_id(:table_name, 'U')")
+ for temp_name in temp_names:
+ result[(schema, temp_name)] = bool(
+ connection.scalar(
+ query,
+ {"table_name": f"tempdb..[{temp_name}]"},
+ )
+ )
- return c.first() is not None
+ for name in table_names:
+ result.setdefault((schema, name), False)
+ return result
@reflection.cache
@_db_plus_owner
# --- multi-reflection API ---
+ @lru_cache
+ def _partition_filter_names_query(self, add_filter_names: bool):
+ sys_objects = ischema.sys_objects
+ sys_schemas = ischema.sys_schemas
+
+ s = (
+ sql.select(sys_objects.c.name)
+ .select_from(sys_objects)
+ .join(
+ sys_schemas,
+ onclause=sys_objects.c.schema_id == sys_schemas.c.schema_id,
+ )
+ .where(
+ sys_schemas.c.name == sql.bindparam("schema"),
+ sys_objects.c.type.in_(
+ sql.bindparam("type_filter", expanding=True)
+ ),
+ )
+ )
+ if add_filter_names:
+ s = s.where(
+ sys_objects.c.name.in_(
+ sql.bindparam("filter_names", expanding=True)
+ )
+ )
+
+ return s
+
def _partition_filter_names(
self, connection, owner, filter_names, scope, kind
):
type_filter = []
if ObjectKind.TABLE in kind:
- type_filter.append("'U'")
+ type_filter.append("U")
if ObjectKind.VIEW in kind:
- type_filter.append("'V'")
+ type_filter.append("V")
# SQL Server does not support materialized views, so ignore them
if not type_filter:
return (regular_names, [], temp_names)
- query = (
- "SELECT o.name "
- "FROM sys.objects o "
- "JOIN sys.schemas s ON o.schema_id = s.schema_id "
- "WHERE s.name = :owner "
- "AND o.type IN (%s)" % ", ".join(type_filter)
- )
- params = [sql.bindparam("owner", owner, ischema.CoerceUnicode())]
- if regular_names:
- query += " AND o.name IN :filter_names"
- params.append(
- sql.bindparam("filter_names", regular_names, expanding=True)
- )
+ has_regular_names = bool(regular_names)
+ query = self._partition_filter_names_query(has_regular_names)
+ params = {"schema": owner, "type_filter": type_filter}
+ if has_regular_names:
+ params["filter_names"] = regular_names
- rp = connection.execute(sql.text(query).bindparams(*params))
- multi_object_names = [row[0] for row in rp]
+ multi_object_names = connection.scalars(query, params).all()
return (regular_names, multi_object_names, temp_names)
# return empty foreign_keys for temp tables that exist. We
# preserve that here by checking existence and returning the
# default (empty) reflection for each temp that is reachable.
+ has_temp_tables = self._internal_has_multi_table(
+ connection, temp_names, owner="dbo", schema=schema, **kw
+ )
for name in temp_names:
- if self._internal_has_table(connection, name, owner="dbo"):
- final[(schema, name)] = ReflectionDefaults.foreign_keys()
+ key = (schema, name)
+ if has_temp_tables.get(key):
+ final[key] = ReflectionDefaults.foreign_keys()
return final.items()
)
@util.memoized_property
- def _has_table_query(self):
+ def _has_multi_table_query(self):
# materialized views are returned by all_tables
tables = (
select(
)
query = select(tables.c.table_name).where(
- tables.c.table_name == bindparam("table_name"),
+ tables.c.table_name.in_(bindparam("table_names")),
tables.c.owner == bindparam("owner"),
)
return query
- @reflection.cache
- def has_table(
- self, connection, table_name, schema=None, dblink=None, **kw
+ def has_multi_table(
+ self, connection, table_names, schema=None, dblink=None, **kw
):
"""Supported kw arguments are: ``dblink`` to reflect via a db link."""
- self._ensure_has_table_connection(connection)
-
- if not schema:
- schema = self.default_schema_name
+ owner = schema or self.default_schema_name
+ db_tns = [self.denormalize_name(tn) for tn in table_names]
params = {
- "table_name": self.denormalize_name(table_name),
- "owner": self.denormalize_schema_name(schema),
+ "table_names": db_tns,
+ "owner": self.denormalize_schema_name(owner),
}
cursor = self._execute_reflection(
connection,
- self._has_table_query,
+ self._has_multi_table_query,
dblink,
returns_long=False,
params=params,
)
- return bool(cursor.scalar())
+ existing = set(cursor.scalars().all())
+ retval = {
+ (schema, table): db_tn in existing
+ for table, db_tn in zip(table_names, db_tns)
+ }
+ return retval.items()
@reflection.cache
def has_sequence(
return pg_class_table.c.relkind == sql.any_(_array.array(relkinds))
@lru_cache()
- def _has_table_query(self, schema):
+ def _has_multi_table_query(self, schema):
query = select(pg_catalog.pg_class.c.relname).where(
- pg_catalog.pg_class.c.relname == bindparam("table_name"),
+ pg_catalog.pg_class.c.relname.in_(
+ bindparam("table_names", expanding=True)
+ ),
self._pg_class_relkind_condition(
pg_catalog.RELKINDS_ALL_TABLE_LIKE
),
@reflection.cache
def has_table(self, connection, table_name, schema=None, **kw):
+ # NOTE: it's not worth calling into the multi table since the query
+ # is compatible also with this single case.
self._ensure_has_table_connection(connection)
- query = self._has_table_query(schema)
- return bool(connection.scalar(query, {"table_name": table_name}))
+ query = self._has_multi_table_query(schema)
+ return bool(connection.scalar(query, {"table_names": [table_name]}))
+
+ def has_multi_table(self, connection, table_names, schema=None, **kw):
+ query = self._has_multi_table_query(schema)
+ params = {"table_names": table_names}
+ existing = set(connection.scalars(query, params).all())
+ retval = {(schema, table): table in existing for table in table_names}
+ return retval.items()
@reflection.cache
def has_sequence(self, connection, sequence_name, schema=None, **kw):
from typing import cast
from typing import Dict
from typing import Final
+from typing import Iterable
from typing import List
from typing import Literal
from typing import Mapping
from .interfaces import DBAPIModule
from .interfaces import DBAPIType
from .interfaces import IsolationLevel
+ from .interfaces import TableKey
from .row import Row
from .url import URL
from ..event import _ListenerFnType
(PostgreSQL, Oracle, MSSQL).
"""
+ @reflection.cache
+ def has_table(
+ self,
+ connection: Connection,
+ table_name: str,
+ schema: Optional[str] = None,
+ **kw: Any,
+ ) -> bool:
+ # NOTE: assume it's a subclass of DefaultDialect
+ self._ensure_has_table_connection(connection) # type: ignore
+ multi_res = self.has_multi_table(
+ connection,
+ table_names=[table_name],
+ schema=schema,
+ **kw,
+ )
+ # has_multi_table returns all the input table names so it's not
+ # possible for the key to be missing
+ return dict(multi_res)[(schema, table_name)]
+
def _value_or_raise(self, data, table, schema):
try:
return dict(data)[(schema, table)]
except exc.NoSuchTableError:
pass
+ def has_multi_table(
+ self,
+ connection: Connection,
+ table_names: Sequence[str],
+ schema: Optional[str] = None,
+ **kw: Any,
+ ) -> Iterable[tuple[TableKey, bool]]:
+ for table_name in table_names:
+ exist = self.has_table(connection, table_name, schema=schema, **kw)
+ yield (schema, table_name), exist
+
def get_multi_table_options(self, connection, **kw):
return self._default_multi_reflect(
self.get_table_options, connection, **kw
raise NotImplementedError()
+ def has_multi_table(
+ self,
+ connection: Connection,
+ table_names: Sequence[str],
+ schema: Optional[str] = None,
+ **kw: Any,
+ ) -> Iterable[Tuple[TableKey, bool]]:
+ """For internal dialect use, check the existence of a particular list
+ of tables or views in the database.
+
+ This is an internal dialect method. Applications should use
+ :meth:`.Inspector.has_multi_table`.
+
+ .. note:: The :class:`_engine.DefaultDialect` provides a default
+ implementation that will call the single table method for
+ each table name provided. Dialects that want to support a faster
+ implementation should implement this method.
+
+ .. versionadded:: 2.1
+
+ """
+
+ raise NotImplementedError()
+
def has_index(
self,
connection: Connection,
consistent interface as well as caching support for previously
fetched metadata.
+ The caching behavior is dialect-specific: as a general rule single table
+ reflection methods should cache their result, while multi-table reflection
+ methods should not, but behavior may vary.
+ Following DDL operations on the database, it is recommended to either
+ create a new :class:`_reflection.Inspector` instance or to clear the cache
+ of existing instances using the :meth:`.Inspector.clear_cache` method.
+
A :class:`_reflection.Inspector` object is usually created via the
:func:`_sa.inspect` function, which may be passed an
:class:`_engine.Engine`
self.info_cache = {}
def clear_cache(self) -> None:
- """reset the cache for this :class:`.Inspector`.
+ """Reset the cache for this :class:`.Inspector`.
Inspection methods that have data cached will emit SQL queries
when next called to get new data.
conn, table_name, schema, info_cache=self.info_cache, **kw
)
+ def has_multi_table(
+ self,
+ table_names: Sequence[str],
+ schema: Optional[str] = None,
+ **kw: Any,
+ ) -> Dict[TableKey, bool]:
+ r"""Return a dict indicating for each table name whether the backend
+ has a table, view, or temporary table with the given name.
+
+ :param table_names: sequence of table names to check.
+ :param schema: schema name to query, if not the default schema.
+ :param \**kw: Additional keyword argument to pass to the dialect
+ specific implementation. See the documentation of the dialect
+ in use for more information.
+
+ .. versionadded:: 2.1
+
+ .. seealso:: :meth:`Inspector.has_table`
+
+ """
+ with self._operation_context() as conn:
+ return dict(
+ self.dialect.has_multi_table(
+ conn, table_names, schema, info_cache=self.info_cache, **kw
+ )
+ )
+
def has_sequence(
self, sequence_name: str, schema: Optional[str] = None, **kw: Any
) -> bool:
return super()._missing_(value)
-class SchemaGenerator(InvokeCreateDDLBase):
+class _SchemaTableReflector:
+ _effective_tables: typing_Sequence[Table] | None = None
+ _has_tables: dict[tuple[Optional[str], str], bool]
+
+ def __init__(
+ self,
+ dialect: Dialect,
+ connection: Connection,
+ tables: typing_Sequence[Table] | None,
+ ):
+ self.dialect = dialect
+ self.connection = connection
+ self.tables = self._effective_tables = tables
+ self._has_tables = {}
+
+ def _update_effective_tables(self, metadata):
+ if self.tables is not None:
+ self._effective_tables = self.tables
+ else:
+ self._effective_tables = list(metadata.tables.values())
+ return self._effective_tables
+
+ def _has_table(self, table: Table, schema: str | None) -> bool:
+ res = self._has_tables.get((schema, table.name))
+ if res is not None:
+ return res
+ if self._effective_tables is None:
+ to_check = [table.name]
+ else:
+ is_view = table.is_view
+ to_check = [
+ t.name
+ for t in self._effective_tables
+ if t.is_view == is_view
+ and self.connection.schema_for_object(t) == schema
+ ]
+ assert table.name in to_check
+ self._has_tables.update(
+ self.dialect.has_multi_table(
+ self.connection, to_check, schema=schema
+ )
+ )
+ return self._has_tables[(schema, table.name)]
+
+
+class SchemaGenerator(InvokeCreateDDLBase, _SchemaTableReflector):
def __init__(
self,
dialect,
tables=None,
**kwargs,
):
- super().__init__(connection, **kwargs)
+ InvokeCreateDDLBase.__init__(self, connection, **kwargs)
+ _SchemaTableReflector.__init__(self, dialect, connection, tables)
self.checkfirst = CheckFirst(checkfirst)
- self.tables = tables
self.preparer = dialect.identifier_preparer
- self.dialect = dialect
self.memo = {}
def _can_create_table(self, table):
bool_to_check = (
CheckFirst.TABLES if not table.is_view else CheckFirst.VIEWS
)
- return (
- not self.checkfirst & bool_to_check
- or not self.dialect.has_table(
- self.connection, table.name, schema=effective_schema
- )
+ return not self.checkfirst & bool_to_check or not self._has_table(
+ table, effective_schema
)
def _can_create_index(self, index):
)
def visit_metadata(self, metadata):
- if self.tables is not None:
- tables = self.tables
- else:
- tables = list(metadata.tables.values())
+ tables = self._update_effective_tables(metadata)
collection = sort_tables_and_constraints(
[t for t in tables if self._can_create_table(t)]
CreateIndex(index)._invoke_with(self.connection)
-class SchemaDropper(InvokeDropDDLBase):
+class SchemaDropper(InvokeDropDDLBase, _SchemaTableReflector):
def __init__(
self,
dialect,
tables=None,
**kwargs,
):
- super().__init__(connection, **kwargs)
+ InvokeDropDDLBase.__init__(self, connection, **kwargs)
+ _SchemaTableReflector.__init__(self, dialect, connection, tables)
self.checkfirst = CheckFirst(checkfirst)
- self.tables = tables
self.preparer = dialect.identifier_preparer
- self.dialect = dialect
self.memo = {}
def visit_metadata(self, metadata):
- if self.tables is not None:
- tables = self.tables
- else:
- tables = list(metadata.tables.values())
+ tables = self._update_effective_tables(metadata)
try:
unsorted_tables = [t for t in tables if self._can_drop_table(t)]
CheckFirst.TABLES if not table.is_view else CheckFirst.VIEWS
)
- return not self.checkfirst & bool_to_check or self.dialect.has_table(
- self.connection, table.name, schema=effective_schema
+ return not self.checkfirst & bool_to_check or self._has_table(
+ table, effective_schema
)
def _can_drop_index(self, index):
is_false(config.db.dialect.has_table(conn, "test_table_s"))
is_false(config.db.dialect.has_table(conn, "nonexistent_table"))
+ def test_has_multi_table(self):
+ with config.db.begin() as conn:
+ res = dict(
+ config.db.dialect.has_multi_table(
+ conn, ["test_table", "test_table_s", "nonexistent_table"]
+ )
+ )
+ exp = {
+ (None, "test_table"): True,
+ (None, "test_table_s"): False,
+ (None, "nonexistent_table"): False,
+ }
+ eq_(res, exp)
+
def test_has_table_cache(self, metadata):
insp = inspect(config.db)
is_true(insp.has_table("test_table"))
)
)
+ @testing.requires.schemas
+ def test_has_multi_table_schema(self):
+
+ with config.db.begin() as conn:
+ res = dict(
+ config.db.dialect.has_multi_table(
+ conn,
+ ["test_table", "test_table_s", "nonexistent_table"],
+ schema=config.test_schema,
+ )
+ )
+ exp = {
+ (config.test_schema, "test_table"): False,
+ (config.test_schema, "test_table_s"): True,
+ (config.test_schema, "nonexistent_table"): False,
+ }
+ eq_(res, exp)
+
@testing.requires.schemas
def test_has_table_nonexistent_schema(self):
with config.db.begin() as conn:
conn, "test_table", schema="nonexistent_schema"
)
)
+ eq_(
+ dict(
+ config.db.dialect.has_multi_table(
+ conn, ["test_table"], schema="nonexistent_schema"
+ )
+ ),
+ {("nonexistent_schema", "test_table"): False},
+ )
@testing.requires.views
def test_has_table_view(self, connection):
table_names,
timing,
mode,
+ multi_needs_table_names=False,
):
single = None
if "single" in mode:
insp = sa.inspect(engine)
multi_fn = getattr(Inspector, multi_fn_name)
with timing(multi_fn.__name__):
- multi = multi_fn(insp, schema=schema_name)
+ if multi_needs_table_names:
+ multi = multi_fn(insp, table_names, schema=schema_name)
+ else:
+ multi = multi_fn(insp, schema=schema_name)
return (multi, single)
verify_dict(multi, single)
+@define_test
+def has_tables(engine, schema_name, table_names, timing, mode, ignore_diff):
+ multi, single = _single_test(
+ "has_table",
+ "has_multi_table",
+ engine,
+ schema_name,
+ table_names,
+ timing,
+ mode,
+ multi_needs_table_names=True,
+ )
+ if not ignore_diff:
+ verify_dict(multi, single, str_compare=True)
+
+
def _apply_events(engine):
queries = defaultdict(list)
class EmitDDLTest(fixtures.TestBase):
def _mock_connection(self, item_exists):
+ def has_many_item(connection, names, schema):
+ return [((schema, name), item_exists(name)) for name in names]
+
def has_item(connection, name, schema):
return item_exists(name)
return Mock(
dialect=Mock(
supports_sequences=True,
- has_table=Mock(side_effect=has_item),
+ has_multi_table=Mock(side_effect=has_many_item),
has_sequence=Mock(side_effect=has_item),
has_index=Mock(side_effect=has_index),
supports_comments=True,
inline_comments=False,
),
+ schema_for_object=lambda t: t.schema,
_schema_translate_map=None,
)
Table("t%d" % i, m, Column("x", Integer)) for i in range(1, 6)
)
- def _table_and_view_fixture(self):
+ def _table_and_view_fixture(self, multi_schema=False):
m = MetaData()
tables = [
- Table("t%d" % i, m, Column("x", Integer)) for i in range(1, 4)
+ Table(
+ "t%d" % i,
+ m,
+ Column("x", Integer),
+ schema="s1" if multi_schema and i > 1 else None,
+ )
+ for i in range(1, 4)
]
t1, t2, t3 = tables
views = [
CreateView(select(t1), "v1", metadata=m).table,
- CreateView(select(t3), "v2", metadata=m).table,
+ CreateView(
+ select(t3),
+ "v2",
+ metadata=m,
+ schema="s1" if multi_schema else None,
+ ).table,
]
return (m,) + tuple(tables) + tuple(views)
self._assert_drop_tables([t1, t2, t3, v1, v2], generator, m, False)
+ def test_create_metadata_wviews_many_schema(self):
+ m, t1, t2, t3, v1, v2 = self._table_and_view_fixture(multi_schema=True)
+ generator = self._mock_create_fixture(
+ True, None, item_exists=lambda t: t not in ("t2", "v2")
+ )
+
+ self._assert_create_tables([t2, v2], generator, m, True)
+
+ def test_drop_metadata_wviews_many_schema(self):
+ m, t1, t2, t3, v1, v2 = self._table_and_view_fixture(multi_schema=True)
+ generator = self._mock_drop_fixture(
+ True, None, item_exists=lambda t: t in ("t2", "v2")
+ )
+
+ self._assert_drop_tables([t2, v2], generator, m, True)
+
def test_create_metadata_auto_alter_fk(self):
m, t1, t2 = self._use_alter_fixture_one()
generator = self._mock_create_fixture(False, [t1, t2])
m,
)
+ def _check_mock(self, generator, tables, views):
+ if tables or views:
+ tbs = {}
+ for t in tables:
+ tbs.setdefault(t.schema, []).append(t.name)
+ vbs = {}
+ for t in views:
+ vbs.setdefault(t.schema, []).append(t.name)
+
+ calls = [
+ mock.call(mock.ANY, names, schema=schema)
+ for data in (tbs, vbs)
+ for schema, names in data.items()
+ ]
+ eq_(generator.dialect.has_multi_table.mock_calls, calls)
+ else:
+ eq_(
+ generator.dialect.has_multi_table.mock_calls,
+ [],
+ )
+
def _assert_create_tables(self, elements, generator, argument, checkfirst):
self._assert_ddl(
(schema.CreateTable, schema.CreateView),
argument,
)
- tables = []
+ tables, views = [], []
if CheckFirst(checkfirst) & CheckFirst.TABLES:
if generator.tables is not None:
tables.extend([t for t in generator.tables if not t.is_view])
if CheckFirst(checkfirst) & CheckFirst.VIEWS:
if generator.tables is not None:
- tables.extend([t for t in generator.tables if t.is_view])
+ views.extend([t for t in generator.tables if t.is_view])
elif isinstance(argument, MetaData):
- tables.extend(
+ views.extend(
[t for t in argument.tables.values() if t.is_view]
)
else:
assert False, "don't know what views we are checking"
- if tables:
- eq_(
- generator.dialect.has_table.mock_calls,
- [
- mock.call(mock.ANY, tablename, schema=mock.ANY)
- for tablename in [t.name for t in tables]
- ],
- )
- else:
- eq_(
- generator.dialect.has_index.mock_calls,
- [],
- )
+ self._check_mock(generator, tables, views)
def _assert_drop_tables(self, elements, generator, argument, checkfirst):
self._assert_ddl(
(schema.DropTable, schema.DropView), elements, generator, argument
)
- tables = []
+ tables, views = [], []
if CheckFirst(checkfirst) & CheckFirst.TABLES:
if generator.tables is not None:
tables.extend([t for t in generator.tables if not t.is_view])
if CheckFirst(checkfirst) & CheckFirst.VIEWS:
if generator.tables is not None:
- tables.extend([t for t in generator.tables if t.is_view])
+ views.extend([t for t in generator.tables if t.is_view])
elif isinstance(argument, MetaData):
- tables.extend(
+ views.extend(
[t for t in argument.tables.values() if t.is_view]
)
else:
assert False, "don't know what views we are checking"
-
- if tables:
- eq_(
- generator.dialect.has_table.mock_calls,
- [
- mock.call(mock.ANY, tablename, schema=mock.ANY)
- for tablename in [t.name for t in tables]
- ],
- )
- else:
- eq_(
- generator.dialect.has_index.mock_calls,
- [],
- )
+ self._check_mock(generator, tables, views)
def _assert_create(self, elements, generator, argument):
self._assert_ddl(
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
