summaryrefslogtreecommitdiff
path: root/venv/lib/python3.11/site-packages/sqlalchemy/engine/reflection.py
diff options
context:
space:
mode:
authorcyfraeviolae <cyfraeviolae>2024-04-03 03:17:55 -0400
committercyfraeviolae <cyfraeviolae>2024-04-03 03:17:55 -0400
commit12cf076118570eebbff08c6b3090e0d4798447a1 (patch)
tree3ba25e17e3c3a5e82316558ba3864b955919ff72 /venv/lib/python3.11/site-packages/sqlalchemy/engine/reflection.py
parentc45662ff3923b34614ddcc8feb9195541166dcc5 (diff)
no venv
Diffstat (limited to 'venv/lib/python3.11/site-packages/sqlalchemy/engine/reflection.py')
-rw-r--r--venv/lib/python3.11/site-packages/sqlalchemy/engine/reflection.py2089
1 files changed, 0 insertions, 2089 deletions
diff --git a/venv/lib/python3.11/site-packages/sqlalchemy/engine/reflection.py b/venv/lib/python3.11/site-packages/sqlalchemy/engine/reflection.py
deleted file mode 100644
index ef1e566..0000000
--- a/venv/lib/python3.11/site-packages/sqlalchemy/engine/reflection.py
+++ /dev/null
@@ -1,2089 +0,0 @@
-# engine/reflection.py
-# Copyright (C) 2005-2024 the SQLAlchemy authors and contributors
-# <see AUTHORS file>
-#
-# This module is part of SQLAlchemy and is released under
-# the MIT License: https://www.opensource.org/licenses/mit-license.php
-
-"""Provides an abstraction for obtaining database schema information.
-
-Usage Notes:
-
-Here are some general conventions when accessing the low level inspector
-methods such as get_table_names, get_columns, etc.
-
-1. Inspector methods return lists of dicts in most cases for the following
- reasons:
-
- * They're both standard types that can be serialized.
- * Using a dict instead of a tuple allows easy expansion of attributes.
- * Using a list for the outer structure maintains order and is easy to work
- with (e.g. list comprehension [d['name'] for d in cols]).
-
-2. Records that contain a name, such as the column name in a column record
- use the key 'name'. So for most return values, each record will have a
- 'name' attribute..
-"""
-from __future__ import annotations
-
-import contextlib
-from dataclasses import dataclass
-from enum import auto
-from enum import Flag
-from enum import unique
-from typing import Any
-from typing import Callable
-from typing import Collection
-from typing import Dict
-from typing import Generator
-from typing import Iterable
-from typing import List
-from typing import Optional
-from typing import Sequence
-from typing import Set
-from typing import Tuple
-from typing import TYPE_CHECKING
-from typing import TypeVar
-from typing import Union
-
-from .base import Connection
-from .base import Engine
-from .. import exc
-from .. import inspection
-from .. import sql
-from .. import util
-from ..sql import operators
-from ..sql import schema as sa_schema
-from ..sql.cache_key import _ad_hoc_cache_key_from_args
-from ..sql.elements import TextClause
-from ..sql.type_api import TypeEngine
-from ..sql.visitors import InternalTraversal
-from ..util import topological
-from ..util.typing import final
-
-if TYPE_CHECKING:
- from .interfaces import Dialect
- from .interfaces import ReflectedCheckConstraint
- from .interfaces import ReflectedColumn
- from .interfaces import ReflectedForeignKeyConstraint
- from .interfaces import ReflectedIndex
- from .interfaces import ReflectedPrimaryKeyConstraint
- from .interfaces import ReflectedTableComment
- from .interfaces import ReflectedUniqueConstraint
- from .interfaces import TableKey
-
-_R = TypeVar("_R")
-
-
-@util.decorator
-def cache(
- fn: Callable[..., _R],
- self: Dialect,
- con: Connection,
- *args: Any,
- **kw: Any,
-) -> _R:
- info_cache = kw.get("info_cache", None)
- if info_cache is None:
- return fn(self, con, *args, **kw)
- exclude = {"info_cache", "unreflectable"}
- key = (
- fn.__name__,
- tuple(a for a in args if isinstance(a, str)),
- tuple((k, v) for k, v in kw.items() if k not in exclude),
- )
- ret: _R = info_cache.get(key)
- if ret is None:
- ret = fn(self, con, *args, **kw)
- info_cache[key] = ret
- return ret
-
-
-def flexi_cache(
- *traverse_args: Tuple[str, InternalTraversal]
-) -> Callable[[Callable[..., _R]], Callable[..., _R]]:
- @util.decorator
- def go(
- fn: Callable[..., _R],
- self: Dialect,
- con: Connection,
- *args: Any,
- **kw: Any,
- ) -> _R:
- info_cache = kw.get("info_cache", None)
- if info_cache is None:
- return fn(self, con, *args, **kw)
- key = _ad_hoc_cache_key_from_args((fn.__name__,), traverse_args, args)
- ret: _R = info_cache.get(key)
- if ret is None:
- ret = fn(self, con, *args, **kw)
- info_cache[key] = ret
- return ret
-
- return go
-
-
-@unique
-class ObjectKind(Flag):
- """Enumerator that indicates which kind of object to return when calling
- the ``get_multi`` methods.
-
- This is a Flag enum, so custom combinations can be passed. For example,
- to reflect tables and plain views ``ObjectKind.TABLE | ObjectKind.VIEW``
- may be used.
-
- .. note::
- Not all dialect may support all kind of object. If a dialect does
- not support a particular object an empty dict is returned.
- In case a dialect supports an object, but the requested method
- is not applicable for the specified kind the default value
- will be returned for each reflected object. For example reflecting
- check constraints of view return a dict with all the views with
- empty lists as values.
- """
-
- TABLE = auto()
- "Reflect table objects"
- VIEW = auto()
- "Reflect plain view objects"
- MATERIALIZED_VIEW = auto()
- "Reflect materialized view object"
-
- ANY_VIEW = VIEW | MATERIALIZED_VIEW
- "Reflect any kind of view objects"
- ANY = TABLE | VIEW | MATERIALIZED_VIEW
- "Reflect all type of objects"
-
-
-@unique
-class ObjectScope(Flag):
- """Enumerator that indicates which scope to use when calling
- the ``get_multi`` methods.
- """
-
- DEFAULT = auto()
- "Include default scope"
- TEMPORARY = auto()
- "Include only temp scope"
- ANY = DEFAULT | TEMPORARY
- "Include both default and temp scope"
-
-
-@inspection._self_inspects
-class Inspector(inspection.Inspectable["Inspector"]):
- """Performs database schema inspection.
-
- The Inspector acts as a proxy to the reflection methods of the
- :class:`~sqlalchemy.engine.interfaces.Dialect`, providing a
- consistent interface as well as caching support for previously
- fetched metadata.
-
- A :class:`_reflection.Inspector` object is usually created via the
- :func:`_sa.inspect` function, which may be passed an
- :class:`_engine.Engine`
- or a :class:`_engine.Connection`::
-
- from sqlalchemy import inspect, create_engine
- engine = create_engine('...')
- insp = inspect(engine)
-
- Where above, the :class:`~sqlalchemy.engine.interfaces.Dialect` associated
- with the engine may opt to return an :class:`_reflection.Inspector`
- subclass that
- provides additional methods specific to the dialect's target database.
-
- """
-
- bind: Union[Engine, Connection]
- engine: Engine
- _op_context_requires_connect: bool
- dialect: Dialect
- info_cache: Dict[Any, Any]
-
- @util.deprecated(
- "1.4",
- "The __init__() method on :class:`_reflection.Inspector` "
- "is deprecated and "
- "will be removed in a future release. Please use the "
- ":func:`.sqlalchemy.inspect` "
- "function on an :class:`_engine.Engine` or "
- ":class:`_engine.Connection` "
- "in order to "
- "acquire an :class:`_reflection.Inspector`.",
- )
- def __init__(self, bind: Union[Engine, Connection]):
- """Initialize a new :class:`_reflection.Inspector`.
-
- :param bind: a :class:`~sqlalchemy.engine.Connection`,
- which is typically an instance of
- :class:`~sqlalchemy.engine.Engine` or
- :class:`~sqlalchemy.engine.Connection`.
-
- For a dialect-specific instance of :class:`_reflection.Inspector`, see
- :meth:`_reflection.Inspector.from_engine`
-
- """
- self._init_legacy(bind)
-
- @classmethod
- def _construct(
- cls, init: Callable[..., Any], bind: Union[Engine, Connection]
- ) -> Inspector:
- if hasattr(bind.dialect, "inspector"):
- cls = bind.dialect.inspector
-
- self = cls.__new__(cls)
- init(self, bind)
- return self
-
- def _init_legacy(self, bind: Union[Engine, Connection]) -> None:
- if hasattr(bind, "exec_driver_sql"):
- self._init_connection(bind) # type: ignore[arg-type]
- else:
- self._init_engine(bind)
-
- def _init_engine(self, engine: Engine) -> None:
- self.bind = self.engine = engine
- engine.connect().close()
- self._op_context_requires_connect = True
- self.dialect = self.engine.dialect
- self.info_cache = {}
-
- def _init_connection(self, connection: Connection) -> None:
- self.bind = connection
- self.engine = connection.engine
- self._op_context_requires_connect = False
- self.dialect = self.engine.dialect
- self.info_cache = {}
-
- def clear_cache(self) -> None:
- """reset the cache for this :class:`.Inspector`.
-
- Inspection methods that have data cached will emit SQL queries
- when next called to get new data.
-
- .. versionadded:: 2.0
-
- """
- self.info_cache.clear()
-
- @classmethod
- @util.deprecated(
- "1.4",
- "The from_engine() method on :class:`_reflection.Inspector` "
- "is deprecated and "
- "will be removed in a future release. Please use the "
- ":func:`.sqlalchemy.inspect` "
- "function on an :class:`_engine.Engine` or "
- ":class:`_engine.Connection` "
- "in order to "
- "acquire an :class:`_reflection.Inspector`.",
- )
- def from_engine(cls, bind: Engine) -> Inspector:
- """Construct a new dialect-specific Inspector object from the given
- engine or connection.
-
- :param bind: a :class:`~sqlalchemy.engine.Connection`
- or :class:`~sqlalchemy.engine.Engine`.
-
- This method differs from direct a direct constructor call of
- :class:`_reflection.Inspector` in that the
- :class:`~sqlalchemy.engine.interfaces.Dialect` is given a chance to
- provide a dialect-specific :class:`_reflection.Inspector` instance,
- which may
- provide additional methods.
-
- See the example at :class:`_reflection.Inspector`.
-
- """
- return cls._construct(cls._init_legacy, bind)
-
- @inspection._inspects(Engine)
- def _engine_insp(bind: Engine) -> Inspector: # type: ignore[misc]
- return Inspector._construct(Inspector._init_engine, bind)
-
- @inspection._inspects(Connection)
- def _connection_insp(bind: Connection) -> Inspector: # type: ignore[misc]
- return Inspector._construct(Inspector._init_connection, bind)
-
- @contextlib.contextmanager
- def _operation_context(self) -> Generator[Connection, None, None]:
- """Return a context that optimizes for multiple operations on a single
- transaction.
-
- This essentially allows connect()/close() to be called if we detected
- that we're against an :class:`_engine.Engine` and not a
- :class:`_engine.Connection`.
-
- """
- conn: Connection
- if self._op_context_requires_connect:
- conn = self.bind.connect() # type: ignore[union-attr]
- else:
- conn = self.bind # type: ignore[assignment]
- try:
- yield conn
- finally:
- if self._op_context_requires_connect:
- conn.close()
-
- @contextlib.contextmanager
- def _inspection_context(self) -> Generator[Inspector, None, None]:
- """Return an :class:`_reflection.Inspector`
- from this one that will run all
- operations on a single connection.
-
- """
-
- with self._operation_context() as conn:
- sub_insp = self._construct(self.__class__._init_connection, conn)
- sub_insp.info_cache = self.info_cache
- yield sub_insp
-
- @property
- def default_schema_name(self) -> Optional[str]:
- """Return the default schema name presented by the dialect
- for the current engine's database user.
-
- E.g. this is typically ``public`` for PostgreSQL and ``dbo``
- for SQL Server.
-
- """
- return self.dialect.default_schema_name
-
- def get_schema_names(self, **kw: Any) -> List[str]:
- r"""Return all schema names.
-
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
- """
-
- with self._operation_context() as conn:
- return self.dialect.get_schema_names(
- conn, info_cache=self.info_cache, **kw
- )
-
- def get_table_names(
- self, schema: Optional[str] = None, **kw: Any
- ) -> List[str]:
- r"""Return all table names within a particular schema.
-
- The names are expected to be real tables only, not views.
- Views are instead returned using the
- :meth:`_reflection.Inspector.get_view_names` and/or
- :meth:`_reflection.Inspector.get_materialized_view_names`
- methods.
-
- :param schema: Schema name. If ``schema`` is left at ``None``, the
- database's default schema is
- used, else the named schema is searched. If the database does not
- support named schemas, behavior is undefined if ``schema`` is not
- passed as ``None``. For special quoting, use :class:`.quoted_name`.
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- .. seealso::
-
- :meth:`_reflection.Inspector.get_sorted_table_and_fkc_names`
-
- :attr:`_schema.MetaData.sorted_tables`
-
- """
-
- with self._operation_context() as conn:
- return self.dialect.get_table_names(
- conn, schema, info_cache=self.info_cache, **kw
- )
-
- def has_table(
- self, table_name: str, schema: Optional[str] = None, **kw: Any
- ) -> bool:
- r"""Return True if the backend has a table, view, or temporary
- table of the given name.
-
- :param table_name: name of the table 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:: 1.4 - the :meth:`.Inspector.has_table` method
- replaces the :meth:`_engine.Engine.has_table` method.
-
- .. versionchanged:: 2.0:: :meth:`.Inspector.has_table` now formally
- supports checking for additional table-like objects:
-
- * any type of views (plain or materialized)
- * temporary tables of any kind
-
- Previously, these two checks were not formally specified and
- different dialects would vary in their behavior. The dialect
- testing suite now includes tests for all of these object types
- and should be supported by all SQLAlchemy-included dialects.
- Support among third party dialects may be lagging, however.
-
- """
- with self._operation_context() as conn:
- return self.dialect.has_table(
- conn, table_name, schema, info_cache=self.info_cache, **kw
- )
-
- def has_sequence(
- self, sequence_name: str, schema: Optional[str] = None, **kw: Any
- ) -> bool:
- r"""Return True if the backend has a sequence with the given name.
-
- :param sequence_name: name of the sequence 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:: 1.4
-
- """
- with self._operation_context() as conn:
- return self.dialect.has_sequence(
- conn, sequence_name, schema, info_cache=self.info_cache, **kw
- )
-
- def has_index(
- self,
- table_name: str,
- index_name: str,
- schema: Optional[str] = None,
- **kw: Any,
- ) -> bool:
- r"""Check the existence of a particular index name in the database.
-
- :param table_name: the name of the table the index belongs to
- :param index_name: the name of the index 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.0
-
- """
- with self._operation_context() as conn:
- return self.dialect.has_index(
- conn,
- table_name,
- index_name,
- schema,
- info_cache=self.info_cache,
- **kw,
- )
-
- def has_schema(self, schema_name: str, **kw: Any) -> bool:
- r"""Return True if the backend has a schema with the given name.
-
- :param schema_name: name of the schema to check
- :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.0
-
- """
- with self._operation_context() as conn:
- return self.dialect.has_schema(
- conn, schema_name, info_cache=self.info_cache, **kw
- )
-
- def get_sorted_table_and_fkc_names(
- self,
- schema: Optional[str] = None,
- **kw: Any,
- ) -> List[Tuple[Optional[str], List[Tuple[str, Optional[str]]]]]:
- r"""Return dependency-sorted table and foreign key constraint names in
- referred to within a particular schema.
-
- This will yield 2-tuples of
- ``(tablename, [(tname, fkname), (tname, fkname), ...])``
- consisting of table names in CREATE order grouped with the foreign key
- constraint names that are not detected as belonging to a cycle.
- The final element
- will be ``(None, [(tname, fkname), (tname, fkname), ..])``
- which will consist of remaining
- foreign key constraint names that would require a separate CREATE
- step after-the-fact, based on dependencies between tables.
-
- :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.
-
- .. seealso::
-
- :meth:`_reflection.Inspector.get_table_names`
-
- :func:`.sort_tables_and_constraints` - similar method which works
- with an already-given :class:`_schema.MetaData`.
-
- """
-
- return [
- (
- table_key[1] if table_key else None,
- [(tname, fks) for (_, tname), fks in fk_collection],
- )
- for (
- table_key,
- fk_collection,
- ) in self.sort_tables_on_foreign_key_dependency(
- consider_schemas=(schema,)
- )
- ]
-
- def sort_tables_on_foreign_key_dependency(
- self,
- consider_schemas: Collection[Optional[str]] = (None,),
- **kw: Any,
- ) -> List[
- Tuple[
- Optional[Tuple[Optional[str], str]],
- List[Tuple[Tuple[Optional[str], str], Optional[str]]],
- ]
- ]:
- r"""Return dependency-sorted table and foreign key constraint names
- referred to within multiple schemas.
-
- This method may be compared to
- :meth:`.Inspector.get_sorted_table_and_fkc_names`, which
- works on one schema at a time; here, the method is a generalization
- that will consider multiple schemas at once including that it will
- resolve for cross-schema foreign keys.
-
- .. versionadded:: 2.0
-
- """
- SchemaTab = Tuple[Optional[str], str]
-
- tuples: Set[Tuple[SchemaTab, SchemaTab]] = set()
- remaining_fkcs: Set[Tuple[SchemaTab, Optional[str]]] = set()
- fknames_for_table: Dict[SchemaTab, Set[Optional[str]]] = {}
- tnames: List[SchemaTab] = []
-
- for schname in consider_schemas:
- schema_fkeys = self.get_multi_foreign_keys(schname, **kw)
- tnames.extend(schema_fkeys)
- for (_, tname), fkeys in schema_fkeys.items():
- fknames_for_table[(schname, tname)] = {
- fk["name"] for fk in fkeys
- }
- for fkey in fkeys:
- if (
- tname != fkey["referred_table"]
- or schname != fkey["referred_schema"]
- ):
- tuples.add(
- (
- (
- fkey["referred_schema"],
- fkey["referred_table"],
- ),
- (schname, tname),
- )
- )
- try:
- candidate_sort = list(topological.sort(tuples, tnames))
- except exc.CircularDependencyError as err:
- edge: Tuple[SchemaTab, SchemaTab]
- for edge in err.edges:
- tuples.remove(edge)
- remaining_fkcs.update(
- (edge[1], fkc) for fkc in fknames_for_table[edge[1]]
- )
-
- candidate_sort = list(topological.sort(tuples, tnames))
- ret: List[
- Tuple[Optional[SchemaTab], List[Tuple[SchemaTab, Optional[str]]]]
- ]
- ret = [
- (
- (schname, tname),
- [
- ((schname, tname), fk)
- for fk in fknames_for_table[(schname, tname)].difference(
- name for _, name in remaining_fkcs
- )
- ],
- )
- for (schname, tname) in candidate_sort
- ]
- return ret + [(None, list(remaining_fkcs))]
-
- def get_temp_table_names(self, **kw: Any) -> List[str]:
- r"""Return a list of temporary table names for the current bind.
-
- This method is unsupported by most dialects; currently
- only Oracle, PostgreSQL and SQLite implements it.
-
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- """
-
- with self._operation_context() as conn:
- return self.dialect.get_temp_table_names(
- conn, info_cache=self.info_cache, **kw
- )
-
- def get_temp_view_names(self, **kw: Any) -> List[str]:
- r"""Return a list of temporary view names for the current bind.
-
- This method is unsupported by most dialects; currently
- only PostgreSQL and SQLite implements it.
-
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- """
- with self._operation_context() as conn:
- return self.dialect.get_temp_view_names(
- conn, info_cache=self.info_cache, **kw
- )
-
- def get_table_options(
- self, table_name: str, schema: Optional[str] = None, **kw: Any
- ) -> Dict[str, Any]:
- r"""Return a dictionary of options specified when the table of the
- given name was created.
-
- This currently includes some options that apply to MySQL and Oracle
- tables.
-
- :param table_name: string name of the table. For special quoting,
- use :class:`.quoted_name`.
-
- :param schema: string schema name; if omitted, uses the default schema
- of the database connection. For special quoting,
- use :class:`.quoted_name`.
-
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- :return: a dict with the table options. The returned keys depend on the
- dialect in use. Each one is prefixed with the dialect name.
-
- .. seealso:: :meth:`Inspector.get_multi_table_options`
-
- """
- with self._operation_context() as conn:
- return self.dialect.get_table_options(
- conn, table_name, schema, info_cache=self.info_cache, **kw
- )
-
- def get_multi_table_options(
- self,
- schema: Optional[str] = None,
- filter_names: Optional[Sequence[str]] = None,
- kind: ObjectKind = ObjectKind.TABLE,
- scope: ObjectScope = ObjectScope.DEFAULT,
- **kw: Any,
- ) -> Dict[TableKey, Dict[str, Any]]:
- r"""Return a dictionary of options specified when the tables in the
- given schema were created.
-
- The tables can be filtered by passing the names to use to
- ``filter_names``.
-
- This currently includes some options that apply to MySQL and Oracle
- tables.
-
- :param schema: string schema name; if omitted, uses the default schema
- of the database connection. For special quoting,
- use :class:`.quoted_name`.
-
- :param filter_names: optionally return information only for the
- objects listed here.
-
- :param kind: a :class:`.ObjectKind` that specifies the type of objects
- to reflect. Defaults to ``ObjectKind.TABLE``.
-
- :param scope: a :class:`.ObjectScope` that specifies if options of
- default, temporary or any tables should be reflected.
- Defaults to ``ObjectScope.DEFAULT``.
-
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- :return: a dictionary where the keys are two-tuple schema,table-name
- and the values are dictionaries with the table options.
- The returned keys in each dict depend on the
- dialect in use. Each one is prefixed with the dialect name.
- The schema is ``None`` if no schema is provided.
-
- .. versionadded:: 2.0
-
- .. seealso:: :meth:`Inspector.get_table_options`
- """
- with self._operation_context() as conn:
- res = self.dialect.get_multi_table_options(
- conn,
- schema=schema,
- filter_names=filter_names,
- kind=kind,
- scope=scope,
- info_cache=self.info_cache,
- **kw,
- )
- return dict(res)
-
- def get_view_names(
- self, schema: Optional[str] = None, **kw: Any
- ) -> List[str]:
- r"""Return all non-materialized view names in `schema`.
-
- :param schema: Optional, retrieve names from a non-default schema.
- For special quoting, use :class:`.quoted_name`.
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
-
- .. versionchanged:: 2.0 For those dialects that previously included
- the names of materialized views in this list (currently PostgreSQL),
- this method no longer returns the names of materialized views.
- the :meth:`.Inspector.get_materialized_view_names` method should
- be used instead.
-
- .. seealso::
-
- :meth:`.Inspector.get_materialized_view_names`
-
- """
-
- with self._operation_context() as conn:
- return self.dialect.get_view_names(
- conn, schema, info_cache=self.info_cache, **kw
- )
-
- def get_materialized_view_names(
- self, schema: Optional[str] = None, **kw: Any
- ) -> List[str]:
- r"""Return all materialized view names in `schema`.
-
- :param schema: Optional, retrieve names from a non-default schema.
- For special quoting, use :class:`.quoted_name`.
- :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.0
-
- .. seealso::
-
- :meth:`.Inspector.get_view_names`
-
- """
-
- with self._operation_context() as conn:
- return self.dialect.get_materialized_view_names(
- conn, schema, info_cache=self.info_cache, **kw
- )
-
- def get_sequence_names(
- self, schema: Optional[str] = None, **kw: Any
- ) -> List[str]:
- r"""Return all sequence names in `schema`.
-
- :param schema: Optional, retrieve names from a non-default schema.
- For special quoting, use :class:`.quoted_name`.
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- """
-
- with self._operation_context() as conn:
- return self.dialect.get_sequence_names(
- conn, schema, info_cache=self.info_cache, **kw
- )
-
- def get_view_definition(
- self, view_name: str, schema: Optional[str] = None, **kw: Any
- ) -> str:
- r"""Return definition for the plain or materialized view called
- ``view_name``.
-
- :param view_name: Name of the view.
- :param schema: Optional, retrieve names from a non-default schema.
- For special quoting, use :class:`.quoted_name`.
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- """
-
- with self._operation_context() as conn:
- return self.dialect.get_view_definition(
- conn, view_name, schema, info_cache=self.info_cache, **kw
- )
-
- def get_columns(
- self, table_name: str, schema: Optional[str] = None, **kw: Any
- ) -> List[ReflectedColumn]:
- r"""Return information about columns in ``table_name``.
-
- Given a string ``table_name`` and an optional string ``schema``,
- return column information as a list of :class:`.ReflectedColumn`.
-
- :param table_name: string name of the table. For special quoting,
- use :class:`.quoted_name`.
-
- :param schema: string schema name; if omitted, uses the default schema
- of the database connection. For special quoting,
- use :class:`.quoted_name`.
-
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- :return: list of dictionaries, each representing the definition of
- a database column.
-
- .. seealso:: :meth:`Inspector.get_multi_columns`.
-
- """
-
- with self._operation_context() as conn:
- col_defs = self.dialect.get_columns(
- conn, table_name, schema, info_cache=self.info_cache, **kw
- )
- if col_defs:
- self._instantiate_types([col_defs])
- return col_defs
-
- def _instantiate_types(
- self, data: Iterable[List[ReflectedColumn]]
- ) -> None:
- # make this easy and only return instances for coltype
- for col_defs in data:
- for col_def in col_defs:
- coltype = col_def["type"]
- if not isinstance(coltype, TypeEngine):
- col_def["type"] = coltype()
-
- def get_multi_columns(
- self,
- schema: Optional[str] = None,
- filter_names: Optional[Sequence[str]] = None,
- kind: ObjectKind = ObjectKind.TABLE,
- scope: ObjectScope = ObjectScope.DEFAULT,
- **kw: Any,
- ) -> Dict[TableKey, List[ReflectedColumn]]:
- r"""Return information about columns in all objects in the given
- schema.
-
- The objects can be filtered by passing the names to use to
- ``filter_names``.
-
- For each table the value is a list of :class:`.ReflectedColumn`.
-
- :param schema: string schema name; if omitted, uses the default schema
- of the database connection. For special quoting,
- use :class:`.quoted_name`.
-
- :param filter_names: optionally return information only for the
- objects listed here.
-
- :param kind: a :class:`.ObjectKind` that specifies the type of objects
- to reflect. Defaults to ``ObjectKind.TABLE``.
-
- :param scope: a :class:`.ObjectScope` that specifies if columns of
- default, temporary or any tables should be reflected.
- Defaults to ``ObjectScope.DEFAULT``.
-
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- :return: a dictionary where the keys are two-tuple schema,table-name
- and the values are list of dictionaries, each representing the
- definition of a database column.
- The schema is ``None`` if no schema is provided.
-
- .. versionadded:: 2.0
-
- .. seealso:: :meth:`Inspector.get_columns`
- """
-
- with self._operation_context() as conn:
- table_col_defs = dict(
- self.dialect.get_multi_columns(
- conn,
- schema=schema,
- filter_names=filter_names,
- kind=kind,
- scope=scope,
- info_cache=self.info_cache,
- **kw,
- )
- )
- self._instantiate_types(table_col_defs.values())
- return table_col_defs
-
- def get_pk_constraint(
- self, table_name: str, schema: Optional[str] = None, **kw: Any
- ) -> ReflectedPrimaryKeyConstraint:
- r"""Return information about primary key constraint in ``table_name``.
-
- Given a string ``table_name``, and an optional string `schema`, return
- primary key information as a :class:`.ReflectedPrimaryKeyConstraint`.
-
- :param table_name: string name of the table. For special quoting,
- use :class:`.quoted_name`.
-
- :param schema: string schema name; if omitted, uses the default schema
- of the database connection. For special quoting,
- use :class:`.quoted_name`.
-
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- :return: a dictionary representing the definition of
- a primary key constraint.
-
- .. seealso:: :meth:`Inspector.get_multi_pk_constraint`
- """
- with self._operation_context() as conn:
- return self.dialect.get_pk_constraint(
- conn, table_name, schema, info_cache=self.info_cache, **kw
- )
-
- def get_multi_pk_constraint(
- self,
- schema: Optional[str] = None,
- filter_names: Optional[Sequence[str]] = None,
- kind: ObjectKind = ObjectKind.TABLE,
- scope: ObjectScope = ObjectScope.DEFAULT,
- **kw: Any,
- ) -> Dict[TableKey, ReflectedPrimaryKeyConstraint]:
- r"""Return information about primary key constraints in
- all tables in the given schema.
-
- The tables can be filtered by passing the names to use to
- ``filter_names``.
-
- For each table the value is a :class:`.ReflectedPrimaryKeyConstraint`.
-
- :param schema: string schema name; if omitted, uses the default schema
- of the database connection. For special quoting,
- use :class:`.quoted_name`.
-
- :param filter_names: optionally return information only for the
- objects listed here.
-
- :param kind: a :class:`.ObjectKind` that specifies the type of objects
- to reflect. Defaults to ``ObjectKind.TABLE``.
-
- :param scope: a :class:`.ObjectScope` that specifies if primary keys of
- default, temporary or any tables should be reflected.
- Defaults to ``ObjectScope.DEFAULT``.
-
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- :return: a dictionary where the keys are two-tuple schema,table-name
- and the values are dictionaries, each representing the
- definition of a primary key constraint.
- The schema is ``None`` if no schema is provided.
-
- .. versionadded:: 2.0
-
- .. seealso:: :meth:`Inspector.get_pk_constraint`
- """
- with self._operation_context() as conn:
- return dict(
- self.dialect.get_multi_pk_constraint(
- conn,
- schema=schema,
- filter_names=filter_names,
- kind=kind,
- scope=scope,
- info_cache=self.info_cache,
- **kw,
- )
- )
-
- def get_foreign_keys(
- self, table_name: str, schema: Optional[str] = None, **kw: Any
- ) -> List[ReflectedForeignKeyConstraint]:
- r"""Return information about foreign_keys in ``table_name``.
-
- Given a string ``table_name``, and an optional string `schema`, return
- foreign key information as a list of
- :class:`.ReflectedForeignKeyConstraint`.
-
- :param table_name: string name of the table. For special quoting,
- use :class:`.quoted_name`.
-
- :param schema: string schema name; if omitted, uses the default schema
- of the database connection. For special quoting,
- use :class:`.quoted_name`.
-
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- :return: a list of dictionaries, each representing the
- a foreign key definition.
-
- .. seealso:: :meth:`Inspector.get_multi_foreign_keys`
- """
-
- with self._operation_context() as conn:
- return self.dialect.get_foreign_keys(
- conn, table_name, schema, info_cache=self.info_cache, **kw
- )
-
- def get_multi_foreign_keys(
- self,
- schema: Optional[str] = None,
- filter_names: Optional[Sequence[str]] = None,
- kind: ObjectKind = ObjectKind.TABLE,
- scope: ObjectScope = ObjectScope.DEFAULT,
- **kw: Any,
- ) -> Dict[TableKey, List[ReflectedForeignKeyConstraint]]:
- r"""Return information about foreign_keys in all tables
- in the given schema.
-
- The tables can be filtered by passing the names to use to
- ``filter_names``.
-
- For each table the value is a list of
- :class:`.ReflectedForeignKeyConstraint`.
-
- :param schema: string schema name; if omitted, uses the default schema
- of the database connection. For special quoting,
- use :class:`.quoted_name`.
-
- :param filter_names: optionally return information only for the
- objects listed here.
-
- :param kind: a :class:`.ObjectKind` that specifies the type of objects
- to reflect. Defaults to ``ObjectKind.TABLE``.
-
- :param scope: a :class:`.ObjectScope` that specifies if foreign keys of
- default, temporary or any tables should be reflected.
- Defaults to ``ObjectScope.DEFAULT``.
-
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- :return: a dictionary where the keys are two-tuple schema,table-name
- and the values are list of dictionaries, each representing
- a foreign key definition.
- The schema is ``None`` if no schema is provided.
-
- .. versionadded:: 2.0
-
- .. seealso:: :meth:`Inspector.get_foreign_keys`
- """
-
- with self._operation_context() as conn:
- return dict(
- self.dialect.get_multi_foreign_keys(
- conn,
- schema=schema,
- filter_names=filter_names,
- kind=kind,
- scope=scope,
- info_cache=self.info_cache,
- **kw,
- )
- )
-
- def get_indexes(
- self, table_name: str, schema: Optional[str] = None, **kw: Any
- ) -> List[ReflectedIndex]:
- r"""Return information about indexes in ``table_name``.
-
- Given a string ``table_name`` and an optional string `schema`, return
- index information as a list of :class:`.ReflectedIndex`.
-
- :param table_name: string name of the table. For special quoting,
- use :class:`.quoted_name`.
-
- :param schema: string schema name; if omitted, uses the default schema
- of the database connection. For special quoting,
- use :class:`.quoted_name`.
-
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- :return: a list of dictionaries, each representing the
- definition of an index.
-
- .. seealso:: :meth:`Inspector.get_multi_indexes`
- """
-
- with self._operation_context() as conn:
- return self.dialect.get_indexes(
- conn, table_name, schema, info_cache=self.info_cache, **kw
- )
-
- def get_multi_indexes(
- self,
- schema: Optional[str] = None,
- filter_names: Optional[Sequence[str]] = None,
- kind: ObjectKind = ObjectKind.TABLE,
- scope: ObjectScope = ObjectScope.DEFAULT,
- **kw: Any,
- ) -> Dict[TableKey, List[ReflectedIndex]]:
- r"""Return information about indexes in in all objects
- in the given schema.
-
- The objects can be filtered by passing the names to use to
- ``filter_names``.
-
- For each table the value is a list of :class:`.ReflectedIndex`.
-
- :param schema: string schema name; if omitted, uses the default schema
- of the database connection. For special quoting,
- use :class:`.quoted_name`.
-
- :param filter_names: optionally return information only for the
- objects listed here.
-
- :param kind: a :class:`.ObjectKind` that specifies the type of objects
- to reflect. Defaults to ``ObjectKind.TABLE``.
-
- :param scope: a :class:`.ObjectScope` that specifies if indexes of
- default, temporary or any tables should be reflected.
- Defaults to ``ObjectScope.DEFAULT``.
-
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- :return: a dictionary where the keys are two-tuple schema,table-name
- and the values are list of dictionaries, each representing the
- definition of an index.
- The schema is ``None`` if no schema is provided.
-
- .. versionadded:: 2.0
-
- .. seealso:: :meth:`Inspector.get_indexes`
- """
-
- with self._operation_context() as conn:
- return dict(
- self.dialect.get_multi_indexes(
- conn,
- schema=schema,
- filter_names=filter_names,
- kind=kind,
- scope=scope,
- info_cache=self.info_cache,
- **kw,
- )
- )
-
- def get_unique_constraints(
- self, table_name: str, schema: Optional[str] = None, **kw: Any
- ) -> List[ReflectedUniqueConstraint]:
- r"""Return information about unique constraints in ``table_name``.
-
- Given a string ``table_name`` and an optional string `schema`, return
- unique constraint information as a list of
- :class:`.ReflectedUniqueConstraint`.
-
- :param table_name: string name of the table. For special quoting,
- use :class:`.quoted_name`.
-
- :param schema: string schema name; if omitted, uses the default schema
- of the database connection. For special quoting,
- use :class:`.quoted_name`.
-
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- :return: a list of dictionaries, each representing the
- definition of an unique constraint.
-
- .. seealso:: :meth:`Inspector.get_multi_unique_constraints`
- """
-
- with self._operation_context() as conn:
- return self.dialect.get_unique_constraints(
- conn, table_name, schema, info_cache=self.info_cache, **kw
- )
-
- def get_multi_unique_constraints(
- self,
- schema: Optional[str] = None,
- filter_names: Optional[Sequence[str]] = None,
- kind: ObjectKind = ObjectKind.TABLE,
- scope: ObjectScope = ObjectScope.DEFAULT,
- **kw: Any,
- ) -> Dict[TableKey, List[ReflectedUniqueConstraint]]:
- r"""Return information about unique constraints in all tables
- in the given schema.
-
- The tables can be filtered by passing the names to use to
- ``filter_names``.
-
- For each table the value is a list of
- :class:`.ReflectedUniqueConstraint`.
-
- :param schema: string schema name; if omitted, uses the default schema
- of the database connection. For special quoting,
- use :class:`.quoted_name`.
-
- :param filter_names: optionally return information only for the
- objects listed here.
-
- :param kind: a :class:`.ObjectKind` that specifies the type of objects
- to reflect. Defaults to ``ObjectKind.TABLE``.
-
- :param scope: a :class:`.ObjectScope` that specifies if constraints of
- default, temporary or any tables should be reflected.
- Defaults to ``ObjectScope.DEFAULT``.
-
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- :return: a dictionary where the keys are two-tuple schema,table-name
- and the values are list of dictionaries, each representing the
- definition of an unique constraint.
- The schema is ``None`` if no schema is provided.
-
- .. versionadded:: 2.0
-
- .. seealso:: :meth:`Inspector.get_unique_constraints`
- """
-
- with self._operation_context() as conn:
- return dict(
- self.dialect.get_multi_unique_constraints(
- conn,
- schema=schema,
- filter_names=filter_names,
- kind=kind,
- scope=scope,
- info_cache=self.info_cache,
- **kw,
- )
- )
-
- def get_table_comment(
- self, table_name: str, schema: Optional[str] = None, **kw: Any
- ) -> ReflectedTableComment:
- r"""Return information about the table comment for ``table_name``.
-
- Given a string ``table_name`` and an optional string ``schema``,
- return table comment information as a :class:`.ReflectedTableComment`.
-
- Raises ``NotImplementedError`` for a dialect that does not support
- comments.
-
- :param table_name: string name of the table. For special quoting,
- use :class:`.quoted_name`.
-
- :param schema: string schema name; if omitted, uses the default schema
- of the database connection. For special quoting,
- use :class:`.quoted_name`.
-
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- :return: a dictionary, with the table comment.
-
- .. versionadded:: 1.2
-
- .. seealso:: :meth:`Inspector.get_multi_table_comment`
- """
-
- with self._operation_context() as conn:
- return self.dialect.get_table_comment(
- conn, table_name, schema, info_cache=self.info_cache, **kw
- )
-
- def get_multi_table_comment(
- self,
- schema: Optional[str] = None,
- filter_names: Optional[Sequence[str]] = None,
- kind: ObjectKind = ObjectKind.TABLE,
- scope: ObjectScope = ObjectScope.DEFAULT,
- **kw: Any,
- ) -> Dict[TableKey, ReflectedTableComment]:
- r"""Return information about the table comment in all objects
- in the given schema.
-
- The objects can be filtered by passing the names to use to
- ``filter_names``.
-
- For each table the value is a :class:`.ReflectedTableComment`.
-
- Raises ``NotImplementedError`` for a dialect that does not support
- comments.
-
- :param schema: string schema name; if omitted, uses the default schema
- of the database connection. For special quoting,
- use :class:`.quoted_name`.
-
- :param filter_names: optionally return information only for the
- objects listed here.
-
- :param kind: a :class:`.ObjectKind` that specifies the type of objects
- to reflect. Defaults to ``ObjectKind.TABLE``.
-
- :param scope: a :class:`.ObjectScope` that specifies if comments of
- default, temporary or any tables should be reflected.
- Defaults to ``ObjectScope.DEFAULT``.
-
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- :return: a dictionary where the keys are two-tuple schema,table-name
- and the values are dictionaries, representing the
- table comments.
- The schema is ``None`` if no schema is provided.
-
- .. versionadded:: 2.0
-
- .. seealso:: :meth:`Inspector.get_table_comment`
- """
-
- with self._operation_context() as conn:
- return dict(
- self.dialect.get_multi_table_comment(
- conn,
- schema=schema,
- filter_names=filter_names,
- kind=kind,
- scope=scope,
- info_cache=self.info_cache,
- **kw,
- )
- )
-
- def get_check_constraints(
- self, table_name: str, schema: Optional[str] = None, **kw: Any
- ) -> List[ReflectedCheckConstraint]:
- r"""Return information about check constraints in ``table_name``.
-
- Given a string ``table_name`` and an optional string `schema`, return
- check constraint information as a list of
- :class:`.ReflectedCheckConstraint`.
-
- :param table_name: string name of the table. For special quoting,
- use :class:`.quoted_name`.
-
- :param schema: string schema name; if omitted, uses the default schema
- of the database connection. For special quoting,
- use :class:`.quoted_name`.
-
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- :return: a list of dictionaries, each representing the
- definition of a check constraints.
-
- .. seealso:: :meth:`Inspector.get_multi_check_constraints`
- """
-
- with self._operation_context() as conn:
- return self.dialect.get_check_constraints(
- conn, table_name, schema, info_cache=self.info_cache, **kw
- )
-
- def get_multi_check_constraints(
- self,
- schema: Optional[str] = None,
- filter_names: Optional[Sequence[str]] = None,
- kind: ObjectKind = ObjectKind.TABLE,
- scope: ObjectScope = ObjectScope.DEFAULT,
- **kw: Any,
- ) -> Dict[TableKey, List[ReflectedCheckConstraint]]:
- r"""Return information about check constraints in all tables
- in the given schema.
-
- The tables can be filtered by passing the names to use to
- ``filter_names``.
-
- For each table the value is a list of
- :class:`.ReflectedCheckConstraint`.
-
- :param schema: string schema name; if omitted, uses the default schema
- of the database connection. For special quoting,
- use :class:`.quoted_name`.
-
- :param filter_names: optionally return information only for the
- objects listed here.
-
- :param kind: a :class:`.ObjectKind` that specifies the type of objects
- to reflect. Defaults to ``ObjectKind.TABLE``.
-
- :param scope: a :class:`.ObjectScope` that specifies if constraints of
- default, temporary or any tables should be reflected.
- Defaults to ``ObjectScope.DEFAULT``.
-
- :param \**kw: Additional keyword argument to pass to the dialect
- specific implementation. See the documentation of the dialect
- in use for more information.
-
- :return: a dictionary where the keys are two-tuple schema,table-name
- and the values are list of dictionaries, each representing the
- definition of a check constraints.
- The schema is ``None`` if no schema is provided.
-
- .. versionadded:: 2.0
-
- .. seealso:: :meth:`Inspector.get_check_constraints`
- """
-
- with self._operation_context() as conn:
- return dict(
- self.dialect.get_multi_check_constraints(
- conn,
- schema=schema,
- filter_names=filter_names,
- kind=kind,
- scope=scope,
- info_cache=self.info_cache,
- **kw,
- )
- )
-
- def reflect_table(
- self,
- table: sa_schema.Table,
- include_columns: Optional[Collection[str]],
- exclude_columns: Collection[str] = (),
- resolve_fks: bool = True,
- _extend_on: Optional[Set[sa_schema.Table]] = None,
- _reflect_info: Optional[_ReflectionInfo] = None,
- ) -> None:
- """Given a :class:`_schema.Table` object, load its internal
- constructs based on introspection.
-
- This is the underlying method used by most dialects to produce
- table reflection. Direct usage is like::
-
- from sqlalchemy import create_engine, MetaData, Table
- from sqlalchemy import inspect
-
- engine = create_engine('...')
- meta = MetaData()
- user_table = Table('user', meta)
- insp = inspect(engine)
- insp.reflect_table(user_table, None)
-
- .. versionchanged:: 1.4 Renamed from ``reflecttable`` to
- ``reflect_table``
-
- :param table: a :class:`~sqlalchemy.schema.Table` instance.
- :param include_columns: a list of string column names to include
- in the reflection process. If ``None``, all columns are reflected.
-
- """
-
- if _extend_on is not None:
- if table in _extend_on:
- return
- else:
- _extend_on.add(table)
-
- dialect = self.bind.dialect
-
- with self._operation_context() as conn:
- schema = conn.schema_for_object(table)
-
- table_name = table.name
-
- # get table-level arguments that are specifically
- # intended for reflection, e.g. oracle_resolve_synonyms.
- # these are unconditionally passed to related Table
- # objects
- reflection_options = {
- k: table.dialect_kwargs.get(k)
- for k in dialect.reflection_options
- if k in table.dialect_kwargs
- }
-
- table_key = (schema, table_name)
- if _reflect_info is None or table_key not in _reflect_info.columns:
- _reflect_info = self._get_reflection_info(
- schema,
- filter_names=[table_name],
- kind=ObjectKind.ANY,
- scope=ObjectScope.ANY,
- _reflect_info=_reflect_info,
- **table.dialect_kwargs,
- )
- if table_key in _reflect_info.unreflectable:
- raise _reflect_info.unreflectable[table_key]
-
- if table_key not in _reflect_info.columns:
- raise exc.NoSuchTableError(table_name)
-
- # reflect table options, like mysql_engine
- if _reflect_info.table_options:
- tbl_opts = _reflect_info.table_options.get(table_key)
- if tbl_opts:
- # add additional kwargs to the Table if the dialect
- # returned them
- table._validate_dialect_kwargs(tbl_opts)
-
- found_table = False
- cols_by_orig_name: Dict[str, sa_schema.Column[Any]] = {}
-
- for col_d in _reflect_info.columns[table_key]:
- found_table = True
-
- self._reflect_column(
- table,
- col_d,
- include_columns,
- exclude_columns,
- cols_by_orig_name,
- )
-
- # NOTE: support tables/views with no columns
- if not found_table and not self.has_table(table_name, schema):
- raise exc.NoSuchTableError(table_name)
-
- self._reflect_pk(
- _reflect_info, table_key, table, cols_by_orig_name, exclude_columns
- )
-
- self._reflect_fk(
- _reflect_info,
- table_key,
- table,
- cols_by_orig_name,
- include_columns,
- exclude_columns,
- resolve_fks,
- _extend_on,
- reflection_options,
- )
-
- self._reflect_indexes(
- _reflect_info,
- table_key,
- table,
- cols_by_orig_name,
- include_columns,
- exclude_columns,
- reflection_options,
- )
-
- self._reflect_unique_constraints(
- _reflect_info,
- table_key,
- table,
- cols_by_orig_name,
- include_columns,
- exclude_columns,
- reflection_options,
- )
-
- self._reflect_check_constraints(
- _reflect_info,
- table_key,
- table,
- cols_by_orig_name,
- include_columns,
- exclude_columns,
- reflection_options,
- )
-
- self._reflect_table_comment(
- _reflect_info,
- table_key,
- table,
- reflection_options,
- )
-
- def _reflect_column(
- self,
- table: sa_schema.Table,
- col_d: ReflectedColumn,
- include_columns: Optional[Collection[str]],
- exclude_columns: Collection[str],
- cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
- ) -> None:
- orig_name = col_d["name"]
-
- table.metadata.dispatch.column_reflect(self, table, col_d)
- table.dispatch.column_reflect(self, table, col_d)
-
- # fetch name again as column_reflect is allowed to
- # change it
- name = col_d["name"]
- if (include_columns and name not in include_columns) or (
- exclude_columns and name in exclude_columns
- ):
- return
-
- coltype = col_d["type"]
-
- col_kw = {
- k: col_d[k] # type: ignore[literal-required]
- for k in [
- "nullable",
- "autoincrement",
- "quote",
- "info",
- "key",
- "comment",
- ]
- if k in col_d
- }
-
- if "dialect_options" in col_d:
- col_kw.update(col_d["dialect_options"])
-
- colargs = []
- default: Any
- if col_d.get("default") is not None:
- default_text = col_d["default"]
- assert default_text is not None
- if isinstance(default_text, TextClause):
- default = sa_schema.DefaultClause(
- default_text, _reflected=True
- )
- elif not isinstance(default_text, sa_schema.FetchedValue):
- default = sa_schema.DefaultClause(
- sql.text(default_text), _reflected=True
- )
- else:
- default = default_text
- colargs.append(default)
-
- if "computed" in col_d:
- computed = sa_schema.Computed(**col_d["computed"])
- colargs.append(computed)
-
- if "identity" in col_d:
- identity = sa_schema.Identity(**col_d["identity"])
- colargs.append(identity)
-
- cols_by_orig_name[orig_name] = col = sa_schema.Column(
- name, coltype, *colargs, **col_kw
- )
-
- if col.key in table.primary_key:
- col.primary_key = True
- table.append_column(col, replace_existing=True)
-
- def _reflect_pk(
- self,
- _reflect_info: _ReflectionInfo,
- table_key: TableKey,
- table: sa_schema.Table,
- cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
- exclude_columns: Collection[str],
- ) -> None:
- pk_cons = _reflect_info.pk_constraint.get(table_key)
- if pk_cons:
- pk_cols = [
- cols_by_orig_name[pk]
- for pk in pk_cons["constrained_columns"]
- if pk in cols_by_orig_name and pk not in exclude_columns
- ]
-
- # update pk constraint name and comment
- table.primary_key.name = pk_cons.get("name")
- table.primary_key.comment = pk_cons.get("comment", None)
-
- # tell the PKConstraint to re-initialize
- # its column collection
- table.primary_key._reload(pk_cols)
-
- def _reflect_fk(
- self,
- _reflect_info: _ReflectionInfo,
- table_key: TableKey,
- table: sa_schema.Table,
- cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
- include_columns: Optional[Collection[str]],
- exclude_columns: Collection[str],
- resolve_fks: bool,
- _extend_on: Optional[Set[sa_schema.Table]],
- reflection_options: Dict[str, Any],
- ) -> None:
- fkeys = _reflect_info.foreign_keys.get(table_key, [])
- for fkey_d in fkeys:
- conname = fkey_d["name"]
- # look for columns by orig name in cols_by_orig_name,
- # but support columns that are in-Python only as fallback
- constrained_columns = [
- cols_by_orig_name[c].key if c in cols_by_orig_name else c
- for c in fkey_d["constrained_columns"]
- ]
-
- if (
- exclude_columns
- and set(constrained_columns).intersection(exclude_columns)
- or (
- include_columns
- and set(constrained_columns).difference(include_columns)
- )
- ):
- continue
-
- referred_schema = fkey_d["referred_schema"]
- referred_table = fkey_d["referred_table"]
- referred_columns = fkey_d["referred_columns"]
- refspec = []
- if referred_schema is not None:
- if resolve_fks:
- sa_schema.Table(
- referred_table,
- table.metadata,
- schema=referred_schema,
- autoload_with=self.bind,
- _extend_on=_extend_on,
- _reflect_info=_reflect_info,
- **reflection_options,
- )
- for column in referred_columns:
- refspec.append(
- ".".join([referred_schema, referred_table, column])
- )
- else:
- if resolve_fks:
- sa_schema.Table(
- referred_table,
- table.metadata,
- autoload_with=self.bind,
- schema=sa_schema.BLANK_SCHEMA,
- _extend_on=_extend_on,
- _reflect_info=_reflect_info,
- **reflection_options,
- )
- for column in referred_columns:
- refspec.append(".".join([referred_table, column]))
- if "options" in fkey_d:
- options = fkey_d["options"]
- else:
- options = {}
-
- try:
- table.append_constraint(
- sa_schema.ForeignKeyConstraint(
- constrained_columns,
- refspec,
- conname,
- link_to_name=True,
- comment=fkey_d.get("comment"),
- **options,
- )
- )
- except exc.ConstraintColumnNotFoundError:
- util.warn(
- f"On reflected table {table.name}, skipping reflection of "
- "foreign key constraint "
- f"{conname}; one or more subject columns within "
- f"name(s) {', '.join(constrained_columns)} are not "
- "present in the table"
- )
-
- _index_sort_exprs = {
- "asc": operators.asc_op,
- "desc": operators.desc_op,
- "nulls_first": operators.nulls_first_op,
- "nulls_last": operators.nulls_last_op,
- }
-
- def _reflect_indexes(
- self,
- _reflect_info: _ReflectionInfo,
- table_key: TableKey,
- table: sa_schema.Table,
- cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
- include_columns: Optional[Collection[str]],
- exclude_columns: Collection[str],
- reflection_options: Dict[str, Any],
- ) -> None:
- # Indexes
- indexes = _reflect_info.indexes.get(table_key, [])
- for index_d in indexes:
- name = index_d["name"]
- columns = index_d["column_names"]
- expressions = index_d.get("expressions")
- column_sorting = index_d.get("column_sorting", {})
- unique = index_d["unique"]
- flavor = index_d.get("type", "index")
- dialect_options = index_d.get("dialect_options", {})
-
- duplicates = index_d.get("duplicates_constraint")
- if include_columns and not set(columns).issubset(include_columns):
- continue
- if duplicates:
- continue
- # look for columns by orig name in cols_by_orig_name,
- # but support columns that are in-Python only as fallback
- idx_element: Any
- idx_elements = []
- for index, c in enumerate(columns):
- if c is None:
- if not expressions:
- util.warn(
- f"Skipping {flavor} {name!r} because key "
- f"{index + 1} reflected as None but no "
- "'expressions' were returned"
- )
- break
- idx_element = sql.text(expressions[index])
- else:
- try:
- if c in cols_by_orig_name:
- idx_element = cols_by_orig_name[c]
- else:
- idx_element = table.c[c]
- except KeyError:
- util.warn(
- f"{flavor} key {c!r} was not located in "
- f"columns for table {table.name!r}"
- )
- continue
- for option in column_sorting.get(c, ()):
- if option in self._index_sort_exprs:
- op = self._index_sort_exprs[option]
- idx_element = op(idx_element)
- idx_elements.append(idx_element)
- else:
- sa_schema.Index(
- name,
- *idx_elements,
- _table=table,
- unique=unique,
- **dialect_options,
- )
-
- def _reflect_unique_constraints(
- self,
- _reflect_info: _ReflectionInfo,
- table_key: TableKey,
- table: sa_schema.Table,
- cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
- include_columns: Optional[Collection[str]],
- exclude_columns: Collection[str],
- reflection_options: Dict[str, Any],
- ) -> None:
- constraints = _reflect_info.unique_constraints.get(table_key, [])
- # Unique Constraints
- for const_d in constraints:
- conname = const_d["name"]
- columns = const_d["column_names"]
- comment = const_d.get("comment")
- duplicates = const_d.get("duplicates_index")
- dialect_options = const_d.get("dialect_options", {})
- if include_columns and not set(columns).issubset(include_columns):
- continue
- if duplicates:
- continue
- # look for columns by orig name in cols_by_orig_name,
- # but support columns that are in-Python only as fallback
- constrained_cols = []
- for c in columns:
- try:
- constrained_col = (
- cols_by_orig_name[c]
- if c in cols_by_orig_name
- else table.c[c]
- )
- except KeyError:
- util.warn(
- "unique constraint key '%s' was not located in "
- "columns for table '%s'" % (c, table.name)
- )
- else:
- constrained_cols.append(constrained_col)
- table.append_constraint(
- sa_schema.UniqueConstraint(
- *constrained_cols,
- name=conname,
- comment=comment,
- **dialect_options,
- )
- )
-
- def _reflect_check_constraints(
- self,
- _reflect_info: _ReflectionInfo,
- table_key: TableKey,
- table: sa_schema.Table,
- cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
- include_columns: Optional[Collection[str]],
- exclude_columns: Collection[str],
- reflection_options: Dict[str, Any],
- ) -> None:
- constraints = _reflect_info.check_constraints.get(table_key, [])
- for const_d in constraints:
- table.append_constraint(sa_schema.CheckConstraint(**const_d))
-
- def _reflect_table_comment(
- self,
- _reflect_info: _ReflectionInfo,
- table_key: TableKey,
- table: sa_schema.Table,
- reflection_options: Dict[str, Any],
- ) -> None:
- comment_dict = _reflect_info.table_comment.get(table_key)
- if comment_dict:
- table.comment = comment_dict["text"]
-
- def _get_reflection_info(
- self,
- schema: Optional[str] = None,
- filter_names: Optional[Collection[str]] = None,
- available: Optional[Collection[str]] = None,
- _reflect_info: Optional[_ReflectionInfo] = None,
- **kw: Any,
- ) -> _ReflectionInfo:
- kw["schema"] = schema
-
- if filter_names and available and len(filter_names) > 100:
- fraction = len(filter_names) / len(available)
- else:
- fraction = None
-
- unreflectable: Dict[TableKey, exc.UnreflectableTableError]
- kw["unreflectable"] = unreflectable = {}
-
- has_result: bool = True
-
- def run(
- meth: Any,
- *,
- optional: bool = False,
- check_filter_names_from_meth: bool = False,
- ) -> Any:
- nonlocal has_result
- # simple heuristic to improve reflection performance if a
- # dialect implements multi_reflection:
- # if more than 50% of the tables in the db are in filter_names
- # load all the tables, since it's most likely faster to avoid
- # a filter on that many tables.
- if (
- fraction is None
- or fraction <= 0.5
- or not self.dialect._overrides_default(meth.__name__)
- ):
- _fn = filter_names
- else:
- _fn = None
- try:
- if has_result:
- res = meth(filter_names=_fn, **kw)
- if check_filter_names_from_meth and not res:
- # method returned no result data.
- # skip any future call methods
- has_result = False
- else:
- res = {}
- except NotImplementedError:
- if not optional:
- raise
- res = {}
- return res
-
- info = _ReflectionInfo(
- columns=run(
- self.get_multi_columns, check_filter_names_from_meth=True
- ),
- pk_constraint=run(self.get_multi_pk_constraint),
- foreign_keys=run(self.get_multi_foreign_keys),
- indexes=run(self.get_multi_indexes),
- unique_constraints=run(
- self.get_multi_unique_constraints, optional=True
- ),
- table_comment=run(self.get_multi_table_comment, optional=True),
- check_constraints=run(
- self.get_multi_check_constraints, optional=True
- ),
- table_options=run(self.get_multi_table_options, optional=True),
- unreflectable=unreflectable,
- )
- if _reflect_info:
- _reflect_info.update(info)
- return _reflect_info
- else:
- return info
-
-
-@final
-class ReflectionDefaults:
- """provides blank default values for reflection methods."""
-
- @classmethod
- def columns(cls) -> List[ReflectedColumn]:
- return []
-
- @classmethod
- def pk_constraint(cls) -> ReflectedPrimaryKeyConstraint:
- return {
- "name": None,
- "constrained_columns": [],
- }
-
- @classmethod
- def foreign_keys(cls) -> List[ReflectedForeignKeyConstraint]:
- return []
-
- @classmethod
- def indexes(cls) -> List[ReflectedIndex]:
- return []
-
- @classmethod
- def unique_constraints(cls) -> List[ReflectedUniqueConstraint]:
- return []
-
- @classmethod
- def check_constraints(cls) -> List[ReflectedCheckConstraint]:
- return []
-
- @classmethod
- def table_options(cls) -> Dict[str, Any]:
- return {}
-
- @classmethod
- def table_comment(cls) -> ReflectedTableComment:
- return {"text": None}
-
-
-@dataclass
-class _ReflectionInfo:
- columns: Dict[TableKey, List[ReflectedColumn]]
- pk_constraint: Dict[TableKey, Optional[ReflectedPrimaryKeyConstraint]]
- foreign_keys: Dict[TableKey, List[ReflectedForeignKeyConstraint]]
- indexes: Dict[TableKey, List[ReflectedIndex]]
- # optionals
- unique_constraints: Dict[TableKey, List[ReflectedUniqueConstraint]]
- table_comment: Dict[TableKey, Optional[ReflectedTableComment]]
- check_constraints: Dict[TableKey, List[ReflectedCheckConstraint]]
- table_options: Dict[TableKey, Dict[str, Any]]
- unreflectable: Dict[TableKey, exc.UnreflectableTableError]
-
- def update(self, other: _ReflectionInfo) -> None:
- for k, v in self.__dict__.items():
- ov = getattr(other, k)
- if ov is not None:
- if v is None:
- setattr(self, k, ov)
- else:
- v.update(ov)