return dbname, owner
-class MSDialect(default.DefaultDialect):
+class MSDialect(default._BackendsMultiReflection, default.DefaultDialect):
# will assume it's at least mssql2005
name = "mssql"
supports_statement_cache = True
lookup = {n.lower(): n for n in filter_names}
return lambda n: lookup.get(n.lower(), n)
- @staticmethod
- def _value_or_raise(data, table, schema):
- """Unwrap a single ``(schema, table)`` entry from a multi-method
- result, raising :exc:`.NoSuchTableError` when missing.
-
- Mirrors PostgreSQL's helper of the same name. Used by the
- single-table reflection wrappers that delegate to the multi
- implementation.
- """
- try:
- return dict(data)[(schema, table)]
- except KeyError:
- raise exc.NoSuchTableError(
- f"{schema}.{table}" if schema else table
- ) from None
-
@_db_plus_owner_multi
def get_multi_columns(
self,
exec_opts={"schema_translate_map": {"sys": "tempdb.sys"}},
)
- # --- Single-table reflection wrappers (delegate to multi) ---
+ # override mixin methods for which this dialect has no native
+ # get_multi_* implementation, preventing infinite recursion
+ # through DefaultDialect._default_multi_reflect
- @reflection.cache
- def get_columns(self, connection, table_name, schema=None, **kw):
- data = self.get_multi_columns(
- connection,
- schema=schema,
- filter_names=[table_name],
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
-
- @reflection.cache
- def get_pk_constraint(self, connection, table_name, schema=None, **kw):
- data = self.get_multi_pk_constraint(
- connection,
- schema=schema,
- filter_names=[table_name],
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
-
- @reflection.cache
- def get_foreign_keys(self, connection, table_name, schema=None, **kw):
- data = self.get_multi_foreign_keys(
- connection,
- schema=schema,
- filter_names=[table_name],
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
+ def get_unique_constraints(
+ self, connection, table_name, schema=None, **kw
+ ):
+ raise NotImplementedError()
- @reflection.cache
- def get_indexes(self, connection, table_name, schema=None, **kw):
- data = self.get_multi_indexes(
- connection,
- schema=schema,
- filter_names=[table_name],
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
+ def get_check_constraints(self, connection, table_name, schema=None, **kw):
+ raise NotImplementedError()
- @reflection.cache
- def get_table_comment(self, connection, table_name, schema=None, **kw):
- data = self.get_multi_table_comment(
- connection,
- schema=schema,
- filter_names=[table_name],
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
+ def get_table_options(self, connection, table_name, schema=None, **kw):
+ raise NotImplementedError()
)
-class OracleDialect(default.DefaultDialect):
+class OracleDialect(default._BackendsMultiReflection, default.DefaultDialect):
name = "oracle"
supports_statement_cache = True
supports_alter = True
else:
return False, {}
- @reflection.cache
- def get_table_options(self, connection, table_name, schema=None, **kw):
- """Supported kw arguments are: ``dblink`` to reflect via a db link;
- ``oracle_resolve_synonyms`` to resolve names to synonyms
- """
- data = self.get_multi_table_options(
- connection,
- schema=schema,
- filter_names=[table_name],
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
-
@lru_cache()
def _table_options_query(
self, owner, scope, kind, has_filter_names, has_mat_views
return options.items()
- @reflection.cache
- def get_columns(self, connection, table_name, schema=None, **kw):
- """Supported kw arguments are: ``dblink`` to reflect via a db link;
- ``oracle_resolve_synonyms`` to resolve names to synonyms
- """
-
- data = self.get_multi_columns(
- connection,
- schema=schema,
- filter_names=[table_name],
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
-
def _run_batches(
self, connection, query, dblink, returns_long, mappings, all_objects
):
identity["oracle_order"] = value == "Y"
return identity
- @reflection.cache
- def get_table_comment(self, connection, table_name, schema=None, **kw):
- """Supported kw arguments are: ``dblink`` to reflect via a db link;
- ``oracle_resolve_synonyms`` to resolve names to synonyms
- """
- data = self.get_multi_table_comment(
- connection,
- schema=schema,
- filter_names=[table_name],
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
-
@lru_cache()
def _comment_query(self, owner, scope, kind, has_filter_names):
# NOTE: all_tab_comments / all_mview_comments have a row for all
for table, comment in result
)
- @reflection.cache
- def get_indexes(self, connection, table_name, schema=None, **kw):
- """Supported kw arguments are: ``dblink`` to reflect via a db link;
- ``oracle_resolve_synonyms`` to resolve names to synonyms
- """
- data = self.get_multi_indexes(
- connection,
- schema=schema,
- filter_names=[table_name],
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
-
@lru_cache()
def _index_query(self, owner):
return (
)
)
- @reflection.cache
- def get_pk_constraint(self, connection, table_name, schema=None, **kw):
- """Supported kw arguments are: ``dblink`` to reflect via a db link;
- ``oracle_resolve_synonyms`` to resolve names to synonyms
- """
- data = self.get_multi_pk_constraint(
- connection,
- schema=schema,
- filter_names=[table_name],
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
-
@lru_cache()
def _constraint_query(self, owner):
local = dictionary.all_cons_columns.alias("local")
)
)
- @reflection.cache
- def get_foreign_keys(
- self,
- connection,
- table_name,
- schema=None,
- **kw,
- ):
- """Supported kw arguments are: ``dblink`` to reflect via a db link;
- ``oracle_resolve_synonyms`` to resolve names to synonyms
- """
- data = self.get_multi_foreign_keys(
- connection,
- schema=schema,
- filter_names=[table_name],
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
-
@_handle_synonyms_decorator
def get_multi_foreign_keys(
self,
)
)
- @reflection.cache
- def get_unique_constraints(
- self, connection, table_name, schema=None, **kw
- ):
- """Supported kw arguments are: ``dblink`` to reflect via a db link;
- ``oracle_resolve_synonyms`` to resolve names to synonyms
- """
- data = self.get_multi_unique_constraints(
- connection,
- schema=schema,
- filter_names=[table_name],
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
-
@_handle_synonyms_decorator
def get_multi_unique_constraints(
self,
else:
return rp
- @reflection.cache
- def get_check_constraints(
- self, connection, table_name, schema=None, include_all=False, **kw
- ):
- """Supported kw arguments are: ``dblink`` to reflect via a db link;
- ``oracle_resolve_synonyms`` to resolve names to synonyms
- """
- data = self.get_multi_check_constraints(
- connection,
- schema=schema,
- filter_names=[table_name],
- scope=ObjectScope.ANY,
- include_all=include_all,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
-
@_handle_synonyms_decorator
def get_multi_check_constraints(
self,
return dialect.get_deferrable(dbapi_conn)
-class PGDialect(default.DefaultDialect):
+class PGDialect(default._BackendsMultiReflection, default.DefaultDialect):
name = "postgresql"
supports_statement_cache = True
supports_alter = True
else:
return res
- def _value_or_raise(self, data, table, schema):
- try:
- return dict(data)[(schema, table)]
- except KeyError:
- raise exc.NoSuchTableError(
- f"{schema}.{table}" if schema else table
- ) from None
-
def _prepare_filter_names(self, filter_names):
if filter_names:
return True, {"filter_names": filter_names}
relkinds += pg_catalog.RELKINDS_MAT_VIEW
return relkinds
- @reflection.cache
- def get_table_options(self, connection, table_name, schema=None, **kw):
- data = self.get_multi_table_options(
- connection,
- schema=schema,
- filter_names=[table_name],
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
-
@lru_cache()
def _table_options_query(self, schema, has_filter_names, scope, kind):
inherits_sq = (
return table_options.items()
- @reflection.cache
- def get_columns(self, connection, table_name, schema=None, **kw):
- data = self.get_multi_columns(
- connection,
- schema=schema,
- filter_names=[table_name],
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
-
@lru_cache()
def _columns_query(self, schema, has_filter_names, scope, kind):
# NOTE: the query with the default and identity options scalar
else:
yield tablename, None, None, None, None
- @reflection.cache
- def get_pk_constraint(self, connection, table_name, schema=None, **kw):
- data = self.get_multi_pk_constraint(
- connection,
- schema=schema,
- filter_names=[table_name],
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
-
def get_multi_pk_constraint(
self, connection, schema, filter_names, scope, kind, **kw
):
for table_name, cols, pk_name, comment, opts in result
)
- @reflection.cache
- def get_foreign_keys(
- self,
- connection,
- table_name,
- schema=None,
- postgresql_ignore_search_path=False,
- **kw,
- ):
- data = self.get_multi_foreign_keys(
- connection,
- schema=schema,
- filter_names=[table_name],
- postgresql_ignore_search_path=postgresql_ignore_search_path,
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
-
@lru_cache()
def _foreing_key_query(self, schema, has_filter_names, scope, kind):
pg_class_ref = pg_catalog.pg_class.alias("cls_ref")
table_fks.append(fkey_d)
return fkeys.items()
- @reflection.cache
- def get_indexes(self, connection, table_name, schema=None, **kw):
- data = self.get_multi_indexes(
- connection,
- schema=schema,
- filter_names=[table_name],
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
-
@util.memoized_property
def _index_query(self):
# NOTE: pg_index is used as from two times to improve performance,
table_indexes.append(index)
return indexes.items()
- @reflection.cache
- def get_unique_constraints(
- self, connection, table_name, schema=None, **kw
- ):
- data = self.get_multi_unique_constraints(
- connection,
- schema=schema,
- filter_names=[table_name],
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
-
def get_multi_unique_constraints(
self,
connection,
uniques[(schema, table_name)].append(uc_dict)
return uniques.items()
- @reflection.cache
- def get_table_comment(self, connection, table_name, schema=None, **kw):
- data = self.get_multi_table_comment(
- connection,
- schema,
- [table_name],
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
-
@lru_cache()
def _comment_query(self, schema, has_filter_names, scope, kind):
relkinds = self._kind_to_relkinds(kind)
for table, comment in result
)
- @reflection.cache
- def get_check_constraints(self, connection, table_name, schema=None, **kw):
- data = self.get_multi_check_constraints(
- connection,
- schema,
- [table_name],
- scope=ObjectScope.ANY,
- kind=ObjectKind.ANY,
- **kw,
- )
- return self._value_or_raise(data, table_name, schema)
-
@lru_cache()
def _check_constraint_query(self, schema, has_filter_names, scope, kind):
relkinds = self._kind_to_relkinds(kind)
from . import characteristics
from . import cursor as _cursor
from . import interfaces
+from . import reflection
from .base import Connection
from .interfaces import CacheStats
from .interfaces import DBAPICursor
) = list(CacheStats)
+class _BackendsMultiReflection(Dialect):
+ """Mixin providing single-table reflection wrappers that delegate to
+ the corresponding ``get_multi_*`` methods.
+
+ Used by dialects that implement native multi-table reflection
+ (PostgreSQL, Oracle, MSSQL).
+ """
+
+ def _value_or_raise(self, data, table, schema):
+ try:
+ return dict(data)[(schema, table)]
+ except KeyError:
+ raise exc.NoSuchTableError(
+ f"{schema}.{table}" if schema else table
+ ) from None
+
+ @reflection.cache
+ def get_columns(self, connection, table_name, schema=None, **kw):
+ data = self.get_multi_columns(
+ connection,
+ schema=schema,
+ filter_names=[table_name],
+ scope=ObjectScope.ANY,
+ kind=ObjectKind.ANY,
+ **kw,
+ )
+ return self._value_or_raise(data, table_name, schema)
+
+ @reflection.cache
+ def get_table_options(self, connection, table_name, schema=None, **kw):
+ data = self.get_multi_table_options(
+ connection,
+ schema=schema,
+ filter_names=[table_name],
+ scope=ObjectScope.ANY,
+ kind=ObjectKind.ANY,
+ **kw,
+ )
+ return self._value_or_raise(data, table_name, schema)
+
+ @reflection.cache
+ def get_pk_constraint(self, connection, table_name, schema=None, **kw):
+ data = self.get_multi_pk_constraint(
+ connection,
+ schema=schema,
+ filter_names=[table_name],
+ scope=ObjectScope.ANY,
+ kind=ObjectKind.ANY,
+ **kw,
+ )
+ return self._value_or_raise(data, table_name, schema)
+
+ @reflection.cache
+ def get_foreign_keys(self, connection, table_name, schema=None, **kw):
+ data = self.get_multi_foreign_keys(
+ connection,
+ schema=schema,
+ filter_names=[table_name],
+ scope=ObjectScope.ANY,
+ kind=ObjectKind.ANY,
+ **kw,
+ )
+ return self._value_or_raise(data, table_name, schema)
+
+ @reflection.cache
+ def get_indexes(self, connection, table_name, schema=None, **kw):
+ data = self.get_multi_indexes(
+ connection,
+ schema=schema,
+ filter_names=[table_name],
+ scope=ObjectScope.ANY,
+ kind=ObjectKind.ANY,
+ **kw,
+ )
+ return self._value_or_raise(data, table_name, schema)
+
+ @reflection.cache
+ def get_unique_constraints(
+ self, connection, table_name, schema=None, **kw
+ ):
+ data = self.get_multi_unique_constraints(
+ connection,
+ schema=schema,
+ filter_names=[table_name],
+ scope=ObjectScope.ANY,
+ kind=ObjectKind.ANY,
+ **kw,
+ )
+ return self._value_or_raise(data, table_name, schema)
+
+ @reflection.cache
+ def get_check_constraints(self, connection, table_name, schema=None, **kw):
+ data = self.get_multi_check_constraints(
+ connection,
+ schema=schema,
+ filter_names=[table_name],
+ scope=ObjectScope.ANY,
+ kind=ObjectKind.ANY,
+ **kw,
+ )
+ return self._value_or_raise(data, table_name, schema)
+
+ @reflection.cache
+ def get_table_comment(self, connection, table_name, schema=None, **kw):
+ data = self.get_multi_table_comment(
+ connection,
+ schema=schema,
+ filter_names=[table_name],
+ scope=ObjectScope.ANY,
+ kind=ObjectKind.ANY,
+ **kw,
+ )
+ return self._value_or_raise(data, table_name, schema)
+
+
class DefaultDialect(Dialect):
"""Default implementation of Dialect"""
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
