Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# engine/reflection.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
8"""Provides an abstraction for obtaining database schema information.
10Usage Notes:
12Here are some general conventions when accessing the low level inspector
13methods such as get_table_names, get_columns, etc.
151. Inspector methods return lists of dicts in most cases for the following
16 reasons:
18 * They're both standard types that can be serialized.
19 * Using a dict instead of a tuple allows easy expansion of attributes.
20 * Using a list for the outer structure maintains order and is easy to work
21 with (e.g. list comprehension [d['name'] for d in cols]).
232. Records that contain a name, such as the column name in a column record
24 use the key 'name'. So for most return values, each record will have a
25 'name' attribute..
26"""
28from __future__ import annotations
30import contextlib
31from dataclasses import dataclass
32from enum import auto
33from enum import Flag
34from enum import unique
35from typing import Any
36from typing import Callable
37from typing import Collection
38from typing import Dict
39from typing import Generator
40from typing import Iterable
41from typing import List
42from typing import Optional
43from typing import Sequence
44from typing import Set
45from typing import Tuple
46from typing import TYPE_CHECKING
47from typing import TypeVar
48from typing import Union
50from .base import Connection
51from .base import Engine
52from .. import exc
53from .. import inspection
54from .. import sql
55from .. import util
56from ..sql import operators
57from ..sql import schema as sa_schema
58from ..sql.cache_key import _ad_hoc_cache_key_from_args
59from ..sql.elements import quoted_name
60from ..sql.elements import TextClause
61from ..sql.type_api import TypeEngine
62from ..sql.visitors import InternalTraversal
63from ..util import topological
64from ..util.typing import final
66if TYPE_CHECKING:
67 from .interfaces import Dialect
68 from .interfaces import ReflectedCheckConstraint
69 from .interfaces import ReflectedColumn
70 from .interfaces import ReflectedForeignKeyConstraint
71 from .interfaces import ReflectedIndex
72 from .interfaces import ReflectedPrimaryKeyConstraint
73 from .interfaces import ReflectedTableComment
74 from .interfaces import ReflectedUniqueConstraint
75 from .interfaces import TableKey
77_R = TypeVar("_R")
80@util.decorator
81def cache(
82 fn: Callable[..., _R],
83 self: Dialect,
84 con: Connection,
85 *args: Any,
86 **kw: Any,
87) -> _R:
88 info_cache = kw.get("info_cache", None)
89 if info_cache is None:
90 return fn(self, con, *args, **kw)
91 exclude = {"info_cache", "unreflectable"}
92 key = (
93 fn.__name__,
94 tuple(
95 (str(a), a.quote) if isinstance(a, quoted_name) else a
96 for a in args
97 if isinstance(a, str)
98 ),
99 tuple(
100 (k, (str(v), v.quote) if isinstance(v, quoted_name) else v)
101 for k, v in kw.items()
102 if k not in exclude
103 ),
104 )
105 ret: _R = info_cache.get(key)
106 if ret is None:
107 ret = fn(self, con, *args, **kw)
108 info_cache[key] = ret
109 return ret
112def flexi_cache(
113 *traverse_args: Tuple[str, InternalTraversal]
114) -> Callable[[Callable[..., _R]], Callable[..., _R]]:
115 @util.decorator
116 def go(
117 fn: Callable[..., _R],
118 self: Dialect,
119 con: Connection,
120 *args: Any,
121 **kw: Any,
122 ) -> _R:
123 info_cache = kw.get("info_cache", None)
124 if info_cache is None:
125 return fn(self, con, *args, **kw)
126 key = _ad_hoc_cache_key_from_args((fn.__name__,), traverse_args, args)
127 ret: _R = info_cache.get(key)
128 if ret is None:
129 ret = fn(self, con, *args, **kw)
130 info_cache[key] = ret
131 return ret
133 return go
136@unique
137class ObjectKind(Flag):
138 """Enumerator that indicates which kind of object to return when calling
139 the ``get_multi`` methods.
141 This is a Flag enum, so custom combinations can be passed. For example,
142 to reflect tables and plain views ``ObjectKind.TABLE | ObjectKind.VIEW``
143 may be used.
145 .. note::
146 Not all dialect may support all kind of object. If a dialect does
147 not support a particular object an empty dict is returned.
148 In case a dialect supports an object, but the requested method
149 is not applicable for the specified kind the default value
150 will be returned for each reflected object. For example reflecting
151 check constraints of view return a dict with all the views with
152 empty lists as values.
153 """
155 TABLE = auto()
156 "Reflect table objects"
157 VIEW = auto()
158 "Reflect plain view objects"
159 MATERIALIZED_VIEW = auto()
160 "Reflect materialized view object"
162 ANY_VIEW = VIEW | MATERIALIZED_VIEW
163 "Reflect any kind of view objects"
164 ANY = TABLE | VIEW | MATERIALIZED_VIEW
165 "Reflect all type of objects"
168@unique
169class ObjectScope(Flag):
170 """Enumerator that indicates which scope to use when calling
171 the ``get_multi`` methods.
172 """
174 DEFAULT = auto()
175 "Include default scope"
176 TEMPORARY = auto()
177 "Include only temp scope"
178 ANY = DEFAULT | TEMPORARY
179 "Include both default and temp scope"
182@inspection._self_inspects
183class Inspector(inspection.Inspectable["Inspector"]):
184 """Performs database schema inspection.
186 The Inspector acts as a proxy to the reflection methods of the
187 :class:`~sqlalchemy.engine.interfaces.Dialect`, providing a
188 consistent interface as well as caching support for previously
189 fetched metadata.
191 A :class:`_reflection.Inspector` object is usually created via the
192 :func:`_sa.inspect` function, which may be passed an
193 :class:`_engine.Engine`
194 or a :class:`_engine.Connection`::
196 from sqlalchemy import inspect, create_engine
198 engine = create_engine("...")
199 insp = inspect(engine)
201 Where above, the :class:`~sqlalchemy.engine.interfaces.Dialect` associated
202 with the engine may opt to return an :class:`_reflection.Inspector`
203 subclass that
204 provides additional methods specific to the dialect's target database.
206 """
208 bind: Union[Engine, Connection]
209 engine: Engine
210 _op_context_requires_connect: bool
211 dialect: Dialect
212 info_cache: Dict[Any, Any]
214 @util.deprecated(
215 "1.4",
216 "The __init__() method on :class:`_reflection.Inspector` "
217 "is deprecated and "
218 "will be removed in a future release. Please use the "
219 ":func:`.sqlalchemy.inspect` "
220 "function on an :class:`_engine.Engine` or "
221 ":class:`_engine.Connection` "
222 "in order to "
223 "acquire an :class:`_reflection.Inspector`.",
224 )
225 def __init__(self, bind: Union[Engine, Connection]):
226 """Initialize a new :class:`_reflection.Inspector`.
228 :param bind: a :class:`~sqlalchemy.engine.Connection`,
229 which is typically an instance of
230 :class:`~sqlalchemy.engine.Engine` or
231 :class:`~sqlalchemy.engine.Connection`.
233 For a dialect-specific instance of :class:`_reflection.Inspector`, see
234 :meth:`_reflection.Inspector.from_engine`
236 """
237 self._init_legacy(bind)
239 @classmethod
240 def _construct(
241 cls, init: Callable[..., Any], bind: Union[Engine, Connection]
242 ) -> Inspector:
243 if hasattr(bind.dialect, "inspector"):
244 cls = bind.dialect.inspector
246 self = cls.__new__(cls)
247 init(self, bind)
248 return self
250 def _init_legacy(self, bind: Union[Engine, Connection]) -> None:
251 if hasattr(bind, "exec_driver_sql"):
252 self._init_connection(bind) # type: ignore[arg-type]
253 else:
254 self._init_engine(bind)
256 def _init_engine(self, engine: Engine) -> None:
257 self.bind = self.engine = engine
258 engine.connect().close()
259 self._op_context_requires_connect = True
260 self.dialect = self.engine.dialect
261 self.info_cache = {}
263 def _init_connection(self, connection: Connection) -> None:
264 self.bind = connection
265 self.engine = connection.engine
266 self._op_context_requires_connect = False
267 self.dialect = self.engine.dialect
268 self.info_cache = {}
270 def clear_cache(self) -> None:
271 """reset the cache for this :class:`.Inspector`.
273 Inspection methods that have data cached will emit SQL queries
274 when next called to get new data.
276 .. versionadded:: 2.0
278 """
279 self.info_cache.clear()
281 @classmethod
282 @util.deprecated(
283 "1.4",
284 "The from_engine() method on :class:`_reflection.Inspector` "
285 "is deprecated and "
286 "will be removed in a future release. Please use the "
287 ":func:`.sqlalchemy.inspect` "
288 "function on an :class:`_engine.Engine` or "
289 ":class:`_engine.Connection` "
290 "in order to "
291 "acquire an :class:`_reflection.Inspector`.",
292 )
293 def from_engine(cls, bind: Engine) -> Inspector:
294 """Construct a new dialect-specific Inspector object from the given
295 engine or connection.
297 :param bind: a :class:`~sqlalchemy.engine.Connection`
298 or :class:`~sqlalchemy.engine.Engine`.
300 This method differs from direct a direct constructor call of
301 :class:`_reflection.Inspector` in that the
302 :class:`~sqlalchemy.engine.interfaces.Dialect` is given a chance to
303 provide a dialect-specific :class:`_reflection.Inspector` instance,
304 which may
305 provide additional methods.
307 See the example at :class:`_reflection.Inspector`.
309 """
310 return cls._construct(cls._init_legacy, bind)
312 @inspection._inspects(Engine)
313 def _engine_insp(bind: Engine) -> Inspector: # type: ignore[misc]
314 return Inspector._construct(Inspector._init_engine, bind)
316 @inspection._inspects(Connection)
317 def _connection_insp(bind: Connection) -> Inspector: # type: ignore[misc]
318 return Inspector._construct(Inspector._init_connection, bind)
320 @contextlib.contextmanager
321 def _operation_context(self) -> Generator[Connection, None, None]:
322 """Return a context that optimizes for multiple operations on a single
323 transaction.
325 This essentially allows connect()/close() to be called if we detected
326 that we're against an :class:`_engine.Engine` and not a
327 :class:`_engine.Connection`.
329 """
330 conn: Connection
331 if self._op_context_requires_connect:
332 conn = self.bind.connect() # type: ignore[union-attr]
333 else:
334 conn = self.bind # type: ignore[assignment]
335 try:
336 yield conn
337 finally:
338 if self._op_context_requires_connect:
339 conn.close()
341 @contextlib.contextmanager
342 def _inspection_context(self) -> Generator[Inspector, None, None]:
343 """Return an :class:`_reflection.Inspector`
344 from this one that will run all
345 operations on a single connection.
347 """
349 with self._operation_context() as conn:
350 sub_insp = self._construct(self.__class__._init_connection, conn)
351 sub_insp.info_cache = self.info_cache
352 yield sub_insp
354 @property
355 def default_schema_name(self) -> Optional[str]:
356 """Return the default schema name presented by the dialect
357 for the current engine's database user.
359 E.g. this is typically ``public`` for PostgreSQL and ``dbo``
360 for SQL Server.
362 """
363 return self.dialect.default_schema_name
365 def get_schema_names(self, **kw: Any) -> List[str]:
366 r"""Return all schema names.
368 :param \**kw: Additional keyword argument to pass to the dialect
369 specific implementation. See the documentation of the dialect
370 in use for more information.
371 """
373 with self._operation_context() as conn:
374 return self.dialect.get_schema_names(
375 conn, info_cache=self.info_cache, **kw
376 )
378 def get_table_names(
379 self, schema: Optional[str] = None, **kw: Any
380 ) -> List[str]:
381 r"""Return all table names within a particular schema.
383 The names are expected to be real tables only, not views.
384 Views are instead returned using the
385 :meth:`_reflection.Inspector.get_view_names` and/or
386 :meth:`_reflection.Inspector.get_materialized_view_names`
387 methods.
389 :param schema: Schema name. If ``schema`` is left at ``None``, the
390 database's default schema is
391 used, else the named schema is searched. If the database does not
392 support named schemas, behavior is undefined if ``schema`` is not
393 passed as ``None``. For special quoting, use :class:`.quoted_name`.
394 :param \**kw: Additional keyword argument to pass to the dialect
395 specific implementation. See the documentation of the dialect
396 in use for more information.
398 .. seealso::
400 :meth:`_reflection.Inspector.get_sorted_table_and_fkc_names`
402 :attr:`_schema.MetaData.sorted_tables`
404 """
406 with self._operation_context() as conn:
407 return self.dialect.get_table_names(
408 conn, schema, info_cache=self.info_cache, **kw
409 )
411 def has_table(
412 self, table_name: str, schema: Optional[str] = None, **kw: Any
413 ) -> bool:
414 r"""Return True if the backend has a table, view, or temporary
415 table of the given name.
417 :param table_name: name of the table to check
418 :param schema: schema name to query, if not the default schema.
419 :param \**kw: Additional keyword argument to pass to the dialect
420 specific implementation. See the documentation of the dialect
421 in use for more information.
423 .. versionadded:: 1.4 - the :meth:`.Inspector.has_table` method
424 replaces the :meth:`_engine.Engine.has_table` method.
426 .. versionchanged:: 2.0:: :meth:`.Inspector.has_table` now formally
427 supports checking for additional table-like objects:
429 * any type of views (plain or materialized)
430 * temporary tables of any kind
432 Previously, these two checks were not formally specified and
433 different dialects would vary in their behavior. The dialect
434 testing suite now includes tests for all of these object types
435 and should be supported by all SQLAlchemy-included dialects.
436 Support among third party dialects may be lagging, however.
438 """
439 with self._operation_context() as conn:
440 return self.dialect.has_table(
441 conn, table_name, schema, info_cache=self.info_cache, **kw
442 )
444 def has_sequence(
445 self, sequence_name: str, schema: Optional[str] = None, **kw: Any
446 ) -> bool:
447 r"""Return True if the backend has a sequence with the given name.
449 :param sequence_name: name of the sequence to check
450 :param schema: schema name to query, if not the default schema.
451 :param \**kw: Additional keyword argument to pass to the dialect
452 specific implementation. See the documentation of the dialect
453 in use for more information.
455 .. versionadded:: 1.4
457 """
458 with self._operation_context() as conn:
459 return self.dialect.has_sequence(
460 conn, sequence_name, schema, info_cache=self.info_cache, **kw
461 )
463 def has_index(
464 self,
465 table_name: str,
466 index_name: str,
467 schema: Optional[str] = None,
468 **kw: Any,
469 ) -> bool:
470 r"""Check the existence of a particular index name in the database.
472 :param table_name: the name of the table the index belongs to
473 :param index_name: the name of the index to check
474 :param schema: schema name to query, if not the default schema.
475 :param \**kw: Additional keyword argument to pass to the dialect
476 specific implementation. See the documentation of the dialect
477 in use for more information.
479 .. versionadded:: 2.0
481 """
482 with self._operation_context() as conn:
483 return self.dialect.has_index(
484 conn,
485 table_name,
486 index_name,
487 schema,
488 info_cache=self.info_cache,
489 **kw,
490 )
492 def has_schema(self, schema_name: str, **kw: Any) -> bool:
493 r"""Return True if the backend has a schema with the given name.
495 :param schema_name: name of the schema to check
496 :param \**kw: Additional keyword argument to pass to the dialect
497 specific implementation. See the documentation of the dialect
498 in use for more information.
500 .. versionadded:: 2.0
502 """
503 with self._operation_context() as conn:
504 return self.dialect.has_schema(
505 conn, schema_name, info_cache=self.info_cache, **kw
506 )
508 def get_sorted_table_and_fkc_names(
509 self,
510 schema: Optional[str] = None,
511 **kw: Any,
512 ) -> List[Tuple[Optional[str], List[Tuple[str, Optional[str]]]]]:
513 r"""Return dependency-sorted table and foreign key constraint names in
514 referred to within a particular schema.
516 This will yield 2-tuples of
517 ``(tablename, [(tname, fkname), (tname, fkname), ...])``
518 consisting of table names in CREATE order grouped with the foreign key
519 constraint names that are not detected as belonging to a cycle.
520 The final element
521 will be ``(None, [(tname, fkname), (tname, fkname), ..])``
522 which will consist of remaining
523 foreign key constraint names that would require a separate CREATE
524 step after-the-fact, based on dependencies between tables.
526 :param schema: schema name to query, if not the default schema.
527 :param \**kw: Additional keyword argument to pass to the dialect
528 specific implementation. See the documentation of the dialect
529 in use for more information.
531 .. seealso::
533 :meth:`_reflection.Inspector.get_table_names`
535 :func:`.sort_tables_and_constraints` - similar method which works
536 with an already-given :class:`_schema.MetaData`.
538 """
540 return [
541 (
542 table_key[1] if table_key else None,
543 [(tname, fks) for (_, tname), fks in fk_collection],
544 )
545 for (
546 table_key,
547 fk_collection,
548 ) in self.sort_tables_on_foreign_key_dependency(
549 consider_schemas=(schema,)
550 )
551 ]
553 def sort_tables_on_foreign_key_dependency(
554 self,
555 consider_schemas: Collection[Optional[str]] = (None,),
556 **kw: Any,
557 ) -> List[
558 Tuple[
559 Optional[Tuple[Optional[str], str]],
560 List[Tuple[Tuple[Optional[str], str], Optional[str]]],
561 ]
562 ]:
563 r"""Return dependency-sorted table and foreign key constraint names
564 referred to within multiple schemas.
566 This method may be compared to
567 :meth:`.Inspector.get_sorted_table_and_fkc_names`, which
568 works on one schema at a time; here, the method is a generalization
569 that will consider multiple schemas at once including that it will
570 resolve for cross-schema foreign keys.
572 .. versionadded:: 2.0
574 """
575 SchemaTab = Tuple[Optional[str], str]
577 tuples: Set[Tuple[SchemaTab, SchemaTab]] = set()
578 remaining_fkcs: Set[Tuple[SchemaTab, Optional[str]]] = set()
579 fknames_for_table: Dict[SchemaTab, Set[Optional[str]]] = {}
580 tnames: List[SchemaTab] = []
582 for schname in consider_schemas:
583 schema_fkeys = self.get_multi_foreign_keys(schname, **kw)
584 tnames.extend(schema_fkeys)
585 for (_, tname), fkeys in schema_fkeys.items():
586 fknames_for_table[(schname, tname)] = {
587 fk["name"] for fk in fkeys
588 }
589 for fkey in fkeys:
590 if (
591 tname != fkey["referred_table"]
592 or schname != fkey["referred_schema"]
593 ):
594 tuples.add(
595 (
596 (
597 fkey["referred_schema"],
598 fkey["referred_table"],
599 ),
600 (schname, tname),
601 )
602 )
603 try:
604 candidate_sort = list(topological.sort(tuples, tnames))
605 except exc.CircularDependencyError as err:
606 edge: Tuple[SchemaTab, SchemaTab]
607 for edge in err.edges:
608 tuples.remove(edge)
609 remaining_fkcs.update(
610 (edge[1], fkc) for fkc in fknames_for_table[edge[1]]
611 )
613 candidate_sort = list(topological.sort(tuples, tnames))
614 ret: List[
615 Tuple[Optional[SchemaTab], List[Tuple[SchemaTab, Optional[str]]]]
616 ]
617 ret = [
618 (
619 (schname, tname),
620 [
621 ((schname, tname), fk)
622 for fk in fknames_for_table[(schname, tname)].difference(
623 name for _, name in remaining_fkcs
624 )
625 ],
626 )
627 for (schname, tname) in candidate_sort
628 ]
629 return ret + [(None, list(remaining_fkcs))]
631 def get_temp_table_names(self, **kw: Any) -> List[str]:
632 r"""Return a list of temporary table names for the current bind.
634 This method is unsupported by most dialects; currently
635 only Oracle Database, PostgreSQL and SQLite implements it.
637 :param \**kw: Additional keyword argument to pass to the dialect
638 specific implementation. See the documentation of the dialect
639 in use for more information.
641 """
643 with self._operation_context() as conn:
644 return self.dialect.get_temp_table_names(
645 conn, info_cache=self.info_cache, **kw
646 )
648 def get_temp_view_names(self, **kw: Any) -> List[str]:
649 r"""Return a list of temporary view names for the current bind.
651 This method is unsupported by most dialects; currently
652 only PostgreSQL and SQLite implements it.
654 :param \**kw: Additional keyword argument to pass to the dialect
655 specific implementation. See the documentation of the dialect
656 in use for more information.
658 """
659 with self._operation_context() as conn:
660 return self.dialect.get_temp_view_names(
661 conn, info_cache=self.info_cache, **kw
662 )
664 def get_table_options(
665 self, table_name: str, schema: Optional[str] = None, **kw: Any
666 ) -> Dict[str, Any]:
667 r"""Return a dictionary of options specified when the table of the
668 given name was created.
670 This currently includes some options that apply to MySQL and Oracle
671 Database tables.
673 :param table_name: string name of the table. For special quoting,
674 use :class:`.quoted_name`.
676 :param schema: string schema name; if omitted, uses the default schema
677 of the database connection. For special quoting,
678 use :class:`.quoted_name`.
680 :param \**kw: Additional keyword argument to pass to the dialect
681 specific implementation. See the documentation of the dialect
682 in use for more information.
684 :return: a dict with the table options. The returned keys depend on the
685 dialect in use. Each one is prefixed with the dialect name.
687 .. seealso:: :meth:`Inspector.get_multi_table_options`
689 """
690 with self._operation_context() as conn:
691 return self.dialect.get_table_options(
692 conn, table_name, schema, info_cache=self.info_cache, **kw
693 )
695 def get_multi_table_options(
696 self,
697 schema: Optional[str] = None,
698 filter_names: Optional[Sequence[str]] = None,
699 kind: ObjectKind = ObjectKind.TABLE,
700 scope: ObjectScope = ObjectScope.DEFAULT,
701 **kw: Any,
702 ) -> Dict[TableKey, Dict[str, Any]]:
703 r"""Return a dictionary of options specified when the tables in the
704 given schema were created.
706 The tables can be filtered by passing the names to use to
707 ``filter_names``.
709 This currently includes some options that apply to MySQL and Oracle
710 tables.
712 :param schema: string schema name; if omitted, uses the default schema
713 of the database connection. For special quoting,
714 use :class:`.quoted_name`.
716 :param filter_names: optionally return information only for the
717 objects listed here.
719 :param kind: a :class:`.ObjectKind` that specifies the type of objects
720 to reflect. Defaults to ``ObjectKind.TABLE``.
722 :param scope: a :class:`.ObjectScope` that specifies if options of
723 default, temporary or any tables should be reflected.
724 Defaults to ``ObjectScope.DEFAULT``.
726 :param \**kw: Additional keyword argument to pass to the dialect
727 specific implementation. See the documentation of the dialect
728 in use for more information.
730 :return: a dictionary where the keys are two-tuple schema,table-name
731 and the values are dictionaries with the table options.
732 The returned keys in each dict depend on the
733 dialect in use. Each one is prefixed with the dialect name.
734 The schema is ``None`` if no schema is provided.
736 .. versionadded:: 2.0
738 .. seealso:: :meth:`Inspector.get_table_options`
739 """
740 with self._operation_context() as conn:
741 res = self.dialect.get_multi_table_options(
742 conn,
743 schema=schema,
744 filter_names=filter_names,
745 kind=kind,
746 scope=scope,
747 info_cache=self.info_cache,
748 **kw,
749 )
750 return dict(res)
752 def get_view_names(
753 self, schema: Optional[str] = None, **kw: Any
754 ) -> List[str]:
755 r"""Return all non-materialized view names in `schema`.
757 :param schema: Optional, retrieve names from a non-default schema.
758 For special quoting, use :class:`.quoted_name`.
759 :param \**kw: Additional keyword argument to pass to the dialect
760 specific implementation. See the documentation of the dialect
761 in use for more information.
764 .. versionchanged:: 2.0 For those dialects that previously included
765 the names of materialized views in this list (currently PostgreSQL),
766 this method no longer returns the names of materialized views.
767 the :meth:`.Inspector.get_materialized_view_names` method should
768 be used instead.
770 .. seealso::
772 :meth:`.Inspector.get_materialized_view_names`
774 """
776 with self._operation_context() as conn:
777 return self.dialect.get_view_names(
778 conn, schema, info_cache=self.info_cache, **kw
779 )
781 def get_materialized_view_names(
782 self, schema: Optional[str] = None, **kw: Any
783 ) -> List[str]:
784 r"""Return all materialized view names in `schema`.
786 :param schema: Optional, retrieve names from a non-default schema.
787 For special quoting, use :class:`.quoted_name`.
788 :param \**kw: Additional keyword argument to pass to the dialect
789 specific implementation. See the documentation of the dialect
790 in use for more information.
792 .. versionadded:: 2.0
794 .. seealso::
796 :meth:`.Inspector.get_view_names`
798 """
800 with self._operation_context() as conn:
801 return self.dialect.get_materialized_view_names(
802 conn, schema, info_cache=self.info_cache, **kw
803 )
805 def get_sequence_names(
806 self, schema: Optional[str] = None, **kw: Any
807 ) -> List[str]:
808 r"""Return all sequence names in `schema`.
810 :param schema: Optional, retrieve names from a non-default schema.
811 For special quoting, use :class:`.quoted_name`.
812 :param \**kw: Additional keyword argument to pass to the dialect
813 specific implementation. See the documentation of the dialect
814 in use for more information.
816 """
818 with self._operation_context() as conn:
819 return self.dialect.get_sequence_names(
820 conn, schema, info_cache=self.info_cache, **kw
821 )
823 def get_view_definition(
824 self, view_name: str, schema: Optional[str] = None, **kw: Any
825 ) -> str:
826 r"""Return definition for the plain or materialized view called
827 ``view_name``.
829 :param view_name: Name of the view.
830 :param schema: Optional, retrieve names from a non-default schema.
831 For special quoting, use :class:`.quoted_name`.
832 :param \**kw: Additional keyword argument to pass to the dialect
833 specific implementation. See the documentation of the dialect
834 in use for more information.
836 """
838 with self._operation_context() as conn:
839 return self.dialect.get_view_definition(
840 conn, view_name, schema, info_cache=self.info_cache, **kw
841 )
843 def get_columns(
844 self, table_name: str, schema: Optional[str] = None, **kw: Any
845 ) -> List[ReflectedColumn]:
846 r"""Return information about columns in ``table_name``.
848 Given a string ``table_name`` and an optional string ``schema``,
849 return column information as a list of :class:`.ReflectedColumn`.
851 :param table_name: string name of the table. For special quoting,
852 use :class:`.quoted_name`.
854 :param schema: string schema name; if omitted, uses the default schema
855 of the database connection. For special quoting,
856 use :class:`.quoted_name`.
858 :param \**kw: Additional keyword argument to pass to the dialect
859 specific implementation. See the documentation of the dialect
860 in use for more information.
862 :return: list of dictionaries, each representing the definition of
863 a database column.
865 .. seealso:: :meth:`Inspector.get_multi_columns`.
867 """
869 with self._operation_context() as conn:
870 col_defs = self.dialect.get_columns(
871 conn, table_name, schema, info_cache=self.info_cache, **kw
872 )
873 if col_defs:
874 self._instantiate_types([col_defs])
875 return col_defs
877 def _instantiate_types(
878 self, data: Iterable[List[ReflectedColumn]]
879 ) -> None:
880 # make this easy and only return instances for coltype
881 for col_defs in data:
882 for col_def in col_defs:
883 coltype = col_def["type"]
884 if not isinstance(coltype, TypeEngine):
885 col_def["type"] = coltype()
887 def get_multi_columns(
888 self,
889 schema: Optional[str] = None,
890 filter_names: Optional[Sequence[str]] = None,
891 kind: ObjectKind = ObjectKind.TABLE,
892 scope: ObjectScope = ObjectScope.DEFAULT,
893 **kw: Any,
894 ) -> Dict[TableKey, List[ReflectedColumn]]:
895 r"""Return information about columns in all objects in the given
896 schema.
898 The objects can be filtered by passing the names to use to
899 ``filter_names``.
901 For each table the value is a list of :class:`.ReflectedColumn`.
903 :param schema: string schema name; if omitted, uses the default schema
904 of the database connection. For special quoting,
905 use :class:`.quoted_name`.
907 :param filter_names: optionally return information only for the
908 objects listed here.
910 :param kind: a :class:`.ObjectKind` that specifies the type of objects
911 to reflect. Defaults to ``ObjectKind.TABLE``.
913 :param scope: a :class:`.ObjectScope` that specifies if columns of
914 default, temporary or any tables should be reflected.
915 Defaults to ``ObjectScope.DEFAULT``.
917 :param \**kw: Additional keyword argument to pass to the dialect
918 specific implementation. See the documentation of the dialect
919 in use for more information.
921 :return: a dictionary where the keys are two-tuple schema,table-name
922 and the values are list of dictionaries, each representing the
923 definition of a database column.
924 The schema is ``None`` if no schema is provided.
926 .. versionadded:: 2.0
928 .. seealso:: :meth:`Inspector.get_columns`
929 """
931 with self._operation_context() as conn:
932 table_col_defs = dict(
933 self.dialect.get_multi_columns(
934 conn,
935 schema=schema,
936 filter_names=filter_names,
937 kind=kind,
938 scope=scope,
939 info_cache=self.info_cache,
940 **kw,
941 )
942 )
943 self._instantiate_types(table_col_defs.values())
944 return table_col_defs
946 def get_pk_constraint(
947 self, table_name: str, schema: Optional[str] = None, **kw: Any
948 ) -> ReflectedPrimaryKeyConstraint:
949 r"""Return information about primary key constraint in ``table_name``.
951 Given a string ``table_name``, and an optional string `schema`, return
952 primary key information as a :class:`.ReflectedPrimaryKeyConstraint`.
954 :param table_name: string name of the table. For special quoting,
955 use :class:`.quoted_name`.
957 :param schema: string schema name; if omitted, uses the default schema
958 of the database connection. For special quoting,
959 use :class:`.quoted_name`.
961 :param \**kw: Additional keyword argument to pass to the dialect
962 specific implementation. See the documentation of the dialect
963 in use for more information.
965 :return: a dictionary representing the definition of
966 a primary key constraint.
968 .. seealso:: :meth:`Inspector.get_multi_pk_constraint`
969 """
970 with self._operation_context() as conn:
971 return self.dialect.get_pk_constraint(
972 conn, table_name, schema, info_cache=self.info_cache, **kw
973 )
975 def get_multi_pk_constraint(
976 self,
977 schema: Optional[str] = None,
978 filter_names: Optional[Sequence[str]] = None,
979 kind: ObjectKind = ObjectKind.TABLE,
980 scope: ObjectScope = ObjectScope.DEFAULT,
981 **kw: Any,
982 ) -> Dict[TableKey, ReflectedPrimaryKeyConstraint]:
983 r"""Return information about primary key constraints in
984 all tables in the given schema.
986 The tables can be filtered by passing the names to use to
987 ``filter_names``.
989 For each table the value is a :class:`.ReflectedPrimaryKeyConstraint`.
991 :param schema: string schema name; if omitted, uses the default schema
992 of the database connection. For special quoting,
993 use :class:`.quoted_name`.
995 :param filter_names: optionally return information only for the
996 objects listed here.
998 :param kind: a :class:`.ObjectKind` that specifies the type of objects
999 to reflect. Defaults to ``ObjectKind.TABLE``.
1001 :param scope: a :class:`.ObjectScope` that specifies if primary keys of
1002 default, temporary or any tables should be reflected.
1003 Defaults to ``ObjectScope.DEFAULT``.
1005 :param \**kw: Additional keyword argument to pass to the dialect
1006 specific implementation. See the documentation of the dialect
1007 in use for more information.
1009 :return: a dictionary where the keys are two-tuple schema,table-name
1010 and the values are dictionaries, each representing the
1011 definition of a primary key constraint.
1012 The schema is ``None`` if no schema is provided.
1014 .. versionadded:: 2.0
1016 .. seealso:: :meth:`Inspector.get_pk_constraint`
1017 """
1018 with self._operation_context() as conn:
1019 return dict(
1020 self.dialect.get_multi_pk_constraint(
1021 conn,
1022 schema=schema,
1023 filter_names=filter_names,
1024 kind=kind,
1025 scope=scope,
1026 info_cache=self.info_cache,
1027 **kw,
1028 )
1029 )
1031 def get_foreign_keys(
1032 self, table_name: str, schema: Optional[str] = None, **kw: Any
1033 ) -> List[ReflectedForeignKeyConstraint]:
1034 r"""Return information about foreign_keys in ``table_name``.
1036 Given a string ``table_name``, and an optional string `schema`, return
1037 foreign key information as a list of
1038 :class:`.ReflectedForeignKeyConstraint`.
1040 :param table_name: string name of the table. For special quoting,
1041 use :class:`.quoted_name`.
1043 :param schema: string schema name; if omitted, uses the default schema
1044 of the database connection. For special quoting,
1045 use :class:`.quoted_name`.
1047 :param \**kw: Additional keyword argument to pass to the dialect
1048 specific implementation. See the documentation of the dialect
1049 in use for more information.
1051 :return: a list of dictionaries, each representing the
1052 a foreign key definition.
1054 .. seealso:: :meth:`Inspector.get_multi_foreign_keys`
1055 """
1057 with self._operation_context() as conn:
1058 return self.dialect.get_foreign_keys(
1059 conn, table_name, schema, info_cache=self.info_cache, **kw
1060 )
1062 def get_multi_foreign_keys(
1063 self,
1064 schema: Optional[str] = None,
1065 filter_names: Optional[Sequence[str]] = None,
1066 kind: ObjectKind = ObjectKind.TABLE,
1067 scope: ObjectScope = ObjectScope.DEFAULT,
1068 **kw: Any,
1069 ) -> Dict[TableKey, List[ReflectedForeignKeyConstraint]]:
1070 r"""Return information about foreign_keys in all tables
1071 in the given schema.
1073 The tables can be filtered by passing the names to use to
1074 ``filter_names``.
1076 For each table the value is a list of
1077 :class:`.ReflectedForeignKeyConstraint`.
1079 :param schema: string schema name; if omitted, uses the default schema
1080 of the database connection. For special quoting,
1081 use :class:`.quoted_name`.
1083 :param filter_names: optionally return information only for the
1084 objects listed here.
1086 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1087 to reflect. Defaults to ``ObjectKind.TABLE``.
1089 :param scope: a :class:`.ObjectScope` that specifies if foreign keys of
1090 default, temporary or any tables should be reflected.
1091 Defaults to ``ObjectScope.DEFAULT``.
1093 :param \**kw: Additional keyword argument to pass to the dialect
1094 specific implementation. See the documentation of the dialect
1095 in use for more information.
1097 :return: a dictionary where the keys are two-tuple schema,table-name
1098 and the values are list of dictionaries, each representing
1099 a foreign key definition.
1100 The schema is ``None`` if no schema is provided.
1102 .. versionadded:: 2.0
1104 .. seealso:: :meth:`Inspector.get_foreign_keys`
1105 """
1107 with self._operation_context() as conn:
1108 return dict(
1109 self.dialect.get_multi_foreign_keys(
1110 conn,
1111 schema=schema,
1112 filter_names=filter_names,
1113 kind=kind,
1114 scope=scope,
1115 info_cache=self.info_cache,
1116 **kw,
1117 )
1118 )
1120 def get_indexes(
1121 self, table_name: str, schema: Optional[str] = None, **kw: Any
1122 ) -> List[ReflectedIndex]:
1123 r"""Return information about indexes in ``table_name``.
1125 Given a string ``table_name`` and an optional string `schema`, return
1126 index information as a list of :class:`.ReflectedIndex`.
1128 :param table_name: string name of the table. For special quoting,
1129 use :class:`.quoted_name`.
1131 :param schema: string schema name; if omitted, uses the default schema
1132 of the database connection. For special quoting,
1133 use :class:`.quoted_name`.
1135 :param \**kw: Additional keyword argument to pass to the dialect
1136 specific implementation. See the documentation of the dialect
1137 in use for more information.
1139 :return: a list of dictionaries, each representing the
1140 definition of an index.
1142 .. seealso:: :meth:`Inspector.get_multi_indexes`
1143 """
1145 with self._operation_context() as conn:
1146 return self.dialect.get_indexes(
1147 conn, table_name, schema, info_cache=self.info_cache, **kw
1148 )
1150 def get_multi_indexes(
1151 self,
1152 schema: Optional[str] = None,
1153 filter_names: Optional[Sequence[str]] = None,
1154 kind: ObjectKind = ObjectKind.TABLE,
1155 scope: ObjectScope = ObjectScope.DEFAULT,
1156 **kw: Any,
1157 ) -> Dict[TableKey, List[ReflectedIndex]]:
1158 r"""Return information about indexes in in all objects
1159 in the given schema.
1161 The objects can be filtered by passing the names to use to
1162 ``filter_names``.
1164 For each table the value is a list of :class:`.ReflectedIndex`.
1166 :param schema: string schema name; if omitted, uses the default schema
1167 of the database connection. For special quoting,
1168 use :class:`.quoted_name`.
1170 :param filter_names: optionally return information only for the
1171 objects listed here.
1173 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1174 to reflect. Defaults to ``ObjectKind.TABLE``.
1176 :param scope: a :class:`.ObjectScope` that specifies if indexes of
1177 default, temporary or any tables should be reflected.
1178 Defaults to ``ObjectScope.DEFAULT``.
1180 :param \**kw: Additional keyword argument to pass to the dialect
1181 specific implementation. See the documentation of the dialect
1182 in use for more information.
1184 :return: a dictionary where the keys are two-tuple schema,table-name
1185 and the values are list of dictionaries, each representing the
1186 definition of an index.
1187 The schema is ``None`` if no schema is provided.
1189 .. versionadded:: 2.0
1191 .. seealso:: :meth:`Inspector.get_indexes`
1192 """
1194 with self._operation_context() as conn:
1195 return dict(
1196 self.dialect.get_multi_indexes(
1197 conn,
1198 schema=schema,
1199 filter_names=filter_names,
1200 kind=kind,
1201 scope=scope,
1202 info_cache=self.info_cache,
1203 **kw,
1204 )
1205 )
1207 def get_unique_constraints(
1208 self, table_name: str, schema: Optional[str] = None, **kw: Any
1209 ) -> List[ReflectedUniqueConstraint]:
1210 r"""Return information about unique constraints in ``table_name``.
1212 Given a string ``table_name`` and an optional string `schema`, return
1213 unique constraint information as a list of
1214 :class:`.ReflectedUniqueConstraint`.
1216 :param table_name: string name of the table. For special quoting,
1217 use :class:`.quoted_name`.
1219 :param schema: string schema name; if omitted, uses the default schema
1220 of the database connection. For special quoting,
1221 use :class:`.quoted_name`.
1223 :param \**kw: Additional keyword argument to pass to the dialect
1224 specific implementation. See the documentation of the dialect
1225 in use for more information.
1227 :return: a list of dictionaries, each representing the
1228 definition of an unique constraint.
1230 .. seealso:: :meth:`Inspector.get_multi_unique_constraints`
1231 """
1233 with self._operation_context() as conn:
1234 return self.dialect.get_unique_constraints(
1235 conn, table_name, schema, info_cache=self.info_cache, **kw
1236 )
1238 def get_multi_unique_constraints(
1239 self,
1240 schema: Optional[str] = None,
1241 filter_names: Optional[Sequence[str]] = None,
1242 kind: ObjectKind = ObjectKind.TABLE,
1243 scope: ObjectScope = ObjectScope.DEFAULT,
1244 **kw: Any,
1245 ) -> Dict[TableKey, List[ReflectedUniqueConstraint]]:
1246 r"""Return information about unique constraints in all tables
1247 in the given schema.
1249 The tables can be filtered by passing the names to use to
1250 ``filter_names``.
1252 For each table the value is a list of
1253 :class:`.ReflectedUniqueConstraint`.
1255 :param schema: string schema name; if omitted, uses the default schema
1256 of the database connection. For special quoting,
1257 use :class:`.quoted_name`.
1259 :param filter_names: optionally return information only for the
1260 objects listed here.
1262 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1263 to reflect. Defaults to ``ObjectKind.TABLE``.
1265 :param scope: a :class:`.ObjectScope` that specifies if constraints of
1266 default, temporary or any tables should be reflected.
1267 Defaults to ``ObjectScope.DEFAULT``.
1269 :param \**kw: Additional keyword argument to pass to the dialect
1270 specific implementation. See the documentation of the dialect
1271 in use for more information.
1273 :return: a dictionary where the keys are two-tuple schema,table-name
1274 and the values are list of dictionaries, each representing the
1275 definition of an unique constraint.
1276 The schema is ``None`` if no schema is provided.
1278 .. versionadded:: 2.0
1280 .. seealso:: :meth:`Inspector.get_unique_constraints`
1281 """
1283 with self._operation_context() as conn:
1284 return dict(
1285 self.dialect.get_multi_unique_constraints(
1286 conn,
1287 schema=schema,
1288 filter_names=filter_names,
1289 kind=kind,
1290 scope=scope,
1291 info_cache=self.info_cache,
1292 **kw,
1293 )
1294 )
1296 def get_table_comment(
1297 self, table_name: str, schema: Optional[str] = None, **kw: Any
1298 ) -> ReflectedTableComment:
1299 r"""Return information about the table comment for ``table_name``.
1301 Given a string ``table_name`` and an optional string ``schema``,
1302 return table comment information as a :class:`.ReflectedTableComment`.
1304 Raises ``NotImplementedError`` for a dialect that does not support
1305 comments.
1307 :param table_name: string name of the table. For special quoting,
1308 use :class:`.quoted_name`.
1310 :param schema: string schema name; if omitted, uses the default schema
1311 of the database connection. For special quoting,
1312 use :class:`.quoted_name`.
1314 :param \**kw: Additional keyword argument to pass to the dialect
1315 specific implementation. See the documentation of the dialect
1316 in use for more information.
1318 :return: a dictionary, with the table comment.
1320 .. versionadded:: 1.2
1322 .. seealso:: :meth:`Inspector.get_multi_table_comment`
1323 """
1325 with self._operation_context() as conn:
1326 return self.dialect.get_table_comment(
1327 conn, table_name, schema, info_cache=self.info_cache, **kw
1328 )
1330 def get_multi_table_comment(
1331 self,
1332 schema: Optional[str] = None,
1333 filter_names: Optional[Sequence[str]] = None,
1334 kind: ObjectKind = ObjectKind.TABLE,
1335 scope: ObjectScope = ObjectScope.DEFAULT,
1336 **kw: Any,
1337 ) -> Dict[TableKey, ReflectedTableComment]:
1338 r"""Return information about the table comment in all objects
1339 in the given schema.
1341 The objects can be filtered by passing the names to use to
1342 ``filter_names``.
1344 For each table the value is a :class:`.ReflectedTableComment`.
1346 Raises ``NotImplementedError`` for a dialect that does not support
1347 comments.
1349 :param schema: string schema name; if omitted, uses the default schema
1350 of the database connection. For special quoting,
1351 use :class:`.quoted_name`.
1353 :param filter_names: optionally return information only for the
1354 objects listed here.
1356 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1357 to reflect. Defaults to ``ObjectKind.TABLE``.
1359 :param scope: a :class:`.ObjectScope` that specifies if comments of
1360 default, temporary or any tables should be reflected.
1361 Defaults to ``ObjectScope.DEFAULT``.
1363 :param \**kw: Additional keyword argument to pass to the dialect
1364 specific implementation. See the documentation of the dialect
1365 in use for more information.
1367 :return: a dictionary where the keys are two-tuple schema,table-name
1368 and the values are dictionaries, representing the
1369 table comments.
1370 The schema is ``None`` if no schema is provided.
1372 .. versionadded:: 2.0
1374 .. seealso:: :meth:`Inspector.get_table_comment`
1375 """
1377 with self._operation_context() as conn:
1378 return dict(
1379 self.dialect.get_multi_table_comment(
1380 conn,
1381 schema=schema,
1382 filter_names=filter_names,
1383 kind=kind,
1384 scope=scope,
1385 info_cache=self.info_cache,
1386 **kw,
1387 )
1388 )
1390 def get_check_constraints(
1391 self, table_name: str, schema: Optional[str] = None, **kw: Any
1392 ) -> List[ReflectedCheckConstraint]:
1393 r"""Return information about check constraints in ``table_name``.
1395 Given a string ``table_name`` and an optional string `schema`, return
1396 check constraint information as a list of
1397 :class:`.ReflectedCheckConstraint`.
1399 :param table_name: string name of the table. For special quoting,
1400 use :class:`.quoted_name`.
1402 :param schema: string schema name; if omitted, uses the default schema
1403 of the database connection. For special quoting,
1404 use :class:`.quoted_name`.
1406 :param \**kw: Additional keyword argument to pass to the dialect
1407 specific implementation. See the documentation of the dialect
1408 in use for more information.
1410 :return: a list of dictionaries, each representing the
1411 definition of a check constraints.
1413 .. seealso:: :meth:`Inspector.get_multi_check_constraints`
1414 """
1416 with self._operation_context() as conn:
1417 return self.dialect.get_check_constraints(
1418 conn, table_name, schema, info_cache=self.info_cache, **kw
1419 )
1421 def get_multi_check_constraints(
1422 self,
1423 schema: Optional[str] = None,
1424 filter_names: Optional[Sequence[str]] = None,
1425 kind: ObjectKind = ObjectKind.TABLE,
1426 scope: ObjectScope = ObjectScope.DEFAULT,
1427 **kw: Any,
1428 ) -> Dict[TableKey, List[ReflectedCheckConstraint]]:
1429 r"""Return information about check constraints in all tables
1430 in the given schema.
1432 The tables can be filtered by passing the names to use to
1433 ``filter_names``.
1435 For each table the value is a list of
1436 :class:`.ReflectedCheckConstraint`.
1438 :param schema: string schema name; if omitted, uses the default schema
1439 of the database connection. For special quoting,
1440 use :class:`.quoted_name`.
1442 :param filter_names: optionally return information only for the
1443 objects listed here.
1445 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1446 to reflect. Defaults to ``ObjectKind.TABLE``.
1448 :param scope: a :class:`.ObjectScope` that specifies if constraints of
1449 default, temporary or any tables should be reflected.
1450 Defaults to ``ObjectScope.DEFAULT``.
1452 :param \**kw: Additional keyword argument to pass to the dialect
1453 specific implementation. See the documentation of the dialect
1454 in use for more information.
1456 :return: a dictionary where the keys are two-tuple schema,table-name
1457 and the values are list of dictionaries, each representing the
1458 definition of a check constraints.
1459 The schema is ``None`` if no schema is provided.
1461 .. versionadded:: 2.0
1463 .. seealso:: :meth:`Inspector.get_check_constraints`
1464 """
1466 with self._operation_context() as conn:
1467 return dict(
1468 self.dialect.get_multi_check_constraints(
1469 conn,
1470 schema=schema,
1471 filter_names=filter_names,
1472 kind=kind,
1473 scope=scope,
1474 info_cache=self.info_cache,
1475 **kw,
1476 )
1477 )
1479 def reflect_table(
1480 self,
1481 table: sa_schema.Table,
1482 include_columns: Optional[Collection[str]],
1483 exclude_columns: Collection[str] = (),
1484 resolve_fks: bool = True,
1485 _extend_on: Optional[Set[sa_schema.Table]] = None,
1486 _reflect_info: Optional[_ReflectionInfo] = None,
1487 ) -> None:
1488 """Given a :class:`_schema.Table` object, load its internal
1489 constructs based on introspection.
1491 This is the underlying method used by most dialects to produce
1492 table reflection. Direct usage is like::
1494 from sqlalchemy import create_engine, MetaData, Table
1495 from sqlalchemy import inspect
1497 engine = create_engine("...")
1498 meta = MetaData()
1499 user_table = Table("user", meta)
1500 insp = inspect(engine)
1501 insp.reflect_table(user_table, None)
1503 .. versionchanged:: 1.4 Renamed from ``reflecttable`` to
1504 ``reflect_table``
1506 :param table: a :class:`~sqlalchemy.schema.Table` instance.
1507 :param include_columns: a list of string column names to include
1508 in the reflection process. If ``None``, all columns are reflected.
1510 """
1512 if _extend_on is not None:
1513 if table in _extend_on:
1514 return
1515 else:
1516 _extend_on.add(table)
1518 dialect = self.bind.dialect
1520 with self._operation_context() as conn:
1521 schema = conn.schema_for_object(table)
1523 table_name = table.name
1525 # get table-level arguments that are specifically
1526 # intended for reflection, e.g. oracle_resolve_synonyms.
1527 # these are unconditionally passed to related Table
1528 # objects
1529 reflection_options = {
1530 k: table.dialect_kwargs.get(k)
1531 for k in dialect.reflection_options
1532 if k in table.dialect_kwargs
1533 }
1535 table_key = (schema, table_name)
1536 if _reflect_info is None or table_key not in _reflect_info.columns:
1537 _reflect_info = self._get_reflection_info(
1538 schema,
1539 filter_names=[table_name],
1540 kind=ObjectKind.ANY,
1541 scope=ObjectScope.ANY,
1542 _reflect_info=_reflect_info,
1543 **table.dialect_kwargs,
1544 )
1545 if table_key in _reflect_info.unreflectable:
1546 raise _reflect_info.unreflectable[table_key]
1548 if table_key not in _reflect_info.columns:
1549 raise exc.NoSuchTableError(table_name)
1551 # reflect table options, like mysql_engine
1552 if _reflect_info.table_options:
1553 tbl_opts = _reflect_info.table_options.get(table_key)
1554 if tbl_opts:
1555 # add additional kwargs to the Table if the dialect
1556 # returned them
1557 table._validate_dialect_kwargs(tbl_opts)
1559 found_table = False
1560 cols_by_orig_name: Dict[str, sa_schema.Column[Any]] = {}
1562 for col_d in _reflect_info.columns[table_key]:
1563 found_table = True
1565 self._reflect_column(
1566 table,
1567 col_d,
1568 include_columns,
1569 exclude_columns,
1570 cols_by_orig_name,
1571 )
1573 # NOTE: support tables/views with no columns
1574 if not found_table and not self.has_table(table_name, schema):
1575 raise exc.NoSuchTableError(table_name)
1577 self._reflect_pk(
1578 _reflect_info, table_key, table, cols_by_orig_name, exclude_columns
1579 )
1581 self._reflect_fk(
1582 _reflect_info,
1583 table_key,
1584 table,
1585 cols_by_orig_name,
1586 include_columns,
1587 exclude_columns,
1588 resolve_fks,
1589 _extend_on,
1590 reflection_options,
1591 )
1593 self._reflect_indexes(
1594 _reflect_info,
1595 table_key,
1596 table,
1597 cols_by_orig_name,
1598 include_columns,
1599 exclude_columns,
1600 reflection_options,
1601 )
1603 self._reflect_unique_constraints(
1604 _reflect_info,
1605 table_key,
1606 table,
1607 cols_by_orig_name,
1608 include_columns,
1609 exclude_columns,
1610 reflection_options,
1611 )
1613 self._reflect_check_constraints(
1614 _reflect_info,
1615 table_key,
1616 table,
1617 cols_by_orig_name,
1618 include_columns,
1619 exclude_columns,
1620 reflection_options,
1621 )
1623 self._reflect_table_comment(
1624 _reflect_info,
1625 table_key,
1626 table,
1627 reflection_options,
1628 )
1630 def _reflect_column(
1631 self,
1632 table: sa_schema.Table,
1633 col_d: ReflectedColumn,
1634 include_columns: Optional[Collection[str]],
1635 exclude_columns: Collection[str],
1636 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1637 ) -> None:
1638 orig_name = col_d["name"]
1640 table.metadata.dispatch.column_reflect(self, table, col_d)
1641 table.dispatch.column_reflect(self, table, col_d)
1643 # fetch name again as column_reflect is allowed to
1644 # change it
1645 name = col_d["name"]
1646 if (include_columns and name not in include_columns) or (
1647 exclude_columns and name in exclude_columns
1648 ):
1649 return
1651 coltype = col_d["type"]
1653 col_kw = {
1654 k: col_d[k] # type: ignore[literal-required]
1655 for k in [
1656 "nullable",
1657 "autoincrement",
1658 "quote",
1659 "info",
1660 "key",
1661 "comment",
1662 ]
1663 if k in col_d
1664 }
1666 if "dialect_options" in col_d:
1667 col_kw.update(col_d["dialect_options"])
1669 colargs = []
1670 default: Any
1671 if col_d.get("default") is not None:
1672 default_text = col_d["default"]
1673 assert default_text is not None
1674 if isinstance(default_text, TextClause):
1675 default = sa_schema.DefaultClause(
1676 default_text, _reflected=True
1677 )
1678 elif not isinstance(default_text, sa_schema.FetchedValue):
1679 default = sa_schema.DefaultClause(
1680 sql.text(default_text), _reflected=True
1681 )
1682 else:
1683 default = default_text
1684 colargs.append(default)
1686 if "computed" in col_d:
1687 computed = sa_schema.Computed(**col_d["computed"])
1688 colargs.append(computed)
1690 if "identity" in col_d:
1691 identity = sa_schema.Identity(**col_d["identity"])
1692 colargs.append(identity)
1694 cols_by_orig_name[orig_name] = col = sa_schema.Column(
1695 name, coltype, *colargs, **col_kw
1696 )
1698 if col.key in table.primary_key:
1699 col.primary_key = True
1700 table.append_column(col, replace_existing=True)
1702 def _reflect_pk(
1703 self,
1704 _reflect_info: _ReflectionInfo,
1705 table_key: TableKey,
1706 table: sa_schema.Table,
1707 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1708 exclude_columns: Collection[str],
1709 ) -> None:
1710 pk_cons = _reflect_info.pk_constraint.get(table_key)
1711 if pk_cons:
1712 pk_cols = [
1713 cols_by_orig_name[pk]
1714 for pk in pk_cons["constrained_columns"]
1715 if pk in cols_by_orig_name and pk not in exclude_columns
1716 ]
1718 # update pk constraint name, comment and dialect_kwargs
1719 table.primary_key.name = pk_cons.get("name")
1720 table.primary_key.comment = pk_cons.get("comment", None)
1721 dialect_options = pk_cons.get("dialect_options")
1722 if dialect_options:
1723 table.primary_key.dialect_kwargs.update(dialect_options)
1725 # tell the PKConstraint to re-initialize
1726 # its column collection
1727 table.primary_key._reload(pk_cols)
1729 def _reflect_fk(
1730 self,
1731 _reflect_info: _ReflectionInfo,
1732 table_key: TableKey,
1733 table: sa_schema.Table,
1734 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1735 include_columns: Optional[Collection[str]],
1736 exclude_columns: Collection[str],
1737 resolve_fks: bool,
1738 _extend_on: Optional[Set[sa_schema.Table]],
1739 reflection_options: Dict[str, Any],
1740 ) -> None:
1741 fkeys = _reflect_info.foreign_keys.get(table_key, [])
1742 for fkey_d in fkeys:
1743 conname = fkey_d["name"]
1744 # look for columns by orig name in cols_by_orig_name,
1745 # but support columns that are in-Python only as fallback
1746 constrained_columns = [
1747 cols_by_orig_name[c].key if c in cols_by_orig_name else c
1748 for c in fkey_d["constrained_columns"]
1749 ]
1751 if (
1752 exclude_columns
1753 and set(constrained_columns).intersection(exclude_columns)
1754 or (
1755 include_columns
1756 and set(constrained_columns).difference(include_columns)
1757 )
1758 ):
1759 continue
1761 referred_schema = fkey_d["referred_schema"]
1762 referred_table = fkey_d["referred_table"]
1763 referred_columns = fkey_d["referred_columns"]
1764 refspec = []
1765 if referred_schema is not None:
1766 if resolve_fks:
1767 sa_schema.Table(
1768 referred_table,
1769 table.metadata,
1770 schema=referred_schema,
1771 autoload_with=self.bind,
1772 _extend_on=_extend_on,
1773 _reflect_info=_reflect_info,
1774 **reflection_options,
1775 )
1776 for column in referred_columns:
1777 refspec.append(
1778 ".".join([referred_schema, referred_table, column])
1779 )
1780 else:
1781 if resolve_fks:
1782 sa_schema.Table(
1783 referred_table,
1784 table.metadata,
1785 autoload_with=self.bind,
1786 schema=sa_schema.BLANK_SCHEMA,
1787 _extend_on=_extend_on,
1788 _reflect_info=_reflect_info,
1789 **reflection_options,
1790 )
1791 for column in referred_columns:
1792 refspec.append(".".join([referred_table, column]))
1793 if "options" in fkey_d:
1794 options = fkey_d["options"]
1795 else:
1796 options = {}
1798 try:
1799 table.append_constraint(
1800 sa_schema.ForeignKeyConstraint(
1801 constrained_columns,
1802 refspec,
1803 conname,
1804 link_to_name=True,
1805 comment=fkey_d.get("comment"),
1806 **options,
1807 )
1808 )
1809 except exc.ConstraintColumnNotFoundError:
1810 util.warn(
1811 f"On reflected table {table.name}, skipping reflection of "
1812 "foreign key constraint "
1813 f"{conname}; one or more subject columns within "
1814 f"name(s) {', '.join(constrained_columns)} are not "
1815 "present in the table"
1816 )
1818 _index_sort_exprs = {
1819 "asc": operators.asc_op,
1820 "desc": operators.desc_op,
1821 "nulls_first": operators.nulls_first_op,
1822 "nulls_last": operators.nulls_last_op,
1823 }
1825 def _reflect_indexes(
1826 self,
1827 _reflect_info: _ReflectionInfo,
1828 table_key: TableKey,
1829 table: sa_schema.Table,
1830 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1831 include_columns: Optional[Collection[str]],
1832 exclude_columns: Collection[str],
1833 reflection_options: Dict[str, Any],
1834 ) -> None:
1835 # Indexes
1836 indexes = _reflect_info.indexes.get(table_key, [])
1837 for index_d in indexes:
1838 name = index_d["name"]
1839 columns = index_d["column_names"]
1840 expressions = index_d.get("expressions")
1841 column_sorting = index_d.get("column_sorting", {})
1842 unique = index_d["unique"]
1843 flavor = index_d.get("type", "index")
1844 dialect_options = index_d.get("dialect_options", {})
1846 duplicates = index_d.get("duplicates_constraint")
1847 if include_columns and not set(columns).issubset(include_columns):
1848 continue
1849 if duplicates:
1850 continue
1851 # look for columns by orig name in cols_by_orig_name,
1852 # but support columns that are in-Python only as fallback
1853 idx_element: Any
1854 idx_elements = []
1855 for index, c in enumerate(columns):
1856 if c is None:
1857 if not expressions:
1858 util.warn(
1859 f"Skipping {flavor} {name!r} because key "
1860 f"{index + 1} reflected as None but no "
1861 "'expressions' were returned"
1862 )
1863 break
1864 idx_element = sql.text(expressions[index])
1865 else:
1866 try:
1867 if c in cols_by_orig_name:
1868 idx_element = cols_by_orig_name[c]
1869 else:
1870 idx_element = table.c[c]
1871 except KeyError:
1872 util.warn(
1873 f"{flavor} key {c!r} was not located in "
1874 f"columns for table {table.name!r}"
1875 )
1876 continue
1877 for option in column_sorting.get(c, ()):
1878 if option in self._index_sort_exprs:
1879 op = self._index_sort_exprs[option]
1880 idx_element = op(idx_element)
1881 idx_elements.append(idx_element)
1882 else:
1883 sa_schema.Index(
1884 name,
1885 *idx_elements,
1886 _table=table,
1887 unique=unique,
1888 **dialect_options,
1889 )
1891 def _reflect_unique_constraints(
1892 self,
1893 _reflect_info: _ReflectionInfo,
1894 table_key: TableKey,
1895 table: sa_schema.Table,
1896 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1897 include_columns: Optional[Collection[str]],
1898 exclude_columns: Collection[str],
1899 reflection_options: Dict[str, Any],
1900 ) -> None:
1901 constraints = _reflect_info.unique_constraints.get(table_key, [])
1902 # Unique Constraints
1903 for const_d in constraints:
1904 conname = const_d["name"]
1905 columns = const_d["column_names"]
1906 comment = const_d.get("comment")
1907 duplicates = const_d.get("duplicates_index")
1908 dialect_options = const_d.get("dialect_options", {})
1909 if include_columns and not set(columns).issubset(include_columns):
1910 continue
1911 if duplicates:
1912 continue
1913 # look for columns by orig name in cols_by_orig_name,
1914 # but support columns that are in-Python only as fallback
1915 constrained_cols = []
1916 for c in columns:
1917 try:
1918 constrained_col = (
1919 cols_by_orig_name[c]
1920 if c in cols_by_orig_name
1921 else table.c[c]
1922 )
1923 except KeyError:
1924 util.warn(
1925 "unique constraint key '%s' was not located in "
1926 "columns for table '%s'" % (c, table.name)
1927 )
1928 else:
1929 constrained_cols.append(constrained_col)
1930 table.append_constraint(
1931 sa_schema.UniqueConstraint(
1932 *constrained_cols,
1933 name=conname,
1934 comment=comment,
1935 **dialect_options,
1936 )
1937 )
1939 def _reflect_check_constraints(
1940 self,
1941 _reflect_info: _ReflectionInfo,
1942 table_key: TableKey,
1943 table: sa_schema.Table,
1944 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1945 include_columns: Optional[Collection[str]],
1946 exclude_columns: Collection[str],
1947 reflection_options: Dict[str, Any],
1948 ) -> None:
1949 constraints = _reflect_info.check_constraints.get(table_key, [])
1950 for const_d in constraints:
1951 table.append_constraint(sa_schema.CheckConstraint(**const_d))
1953 def _reflect_table_comment(
1954 self,
1955 _reflect_info: _ReflectionInfo,
1956 table_key: TableKey,
1957 table: sa_schema.Table,
1958 reflection_options: Dict[str, Any],
1959 ) -> None:
1960 comment_dict = _reflect_info.table_comment.get(table_key)
1961 if comment_dict:
1962 table.comment = comment_dict["text"]
1964 def _get_reflection_info(
1965 self,
1966 schema: Optional[str] = None,
1967 filter_names: Optional[Collection[str]] = None,
1968 available: Optional[Collection[str]] = None,
1969 _reflect_info: Optional[_ReflectionInfo] = None,
1970 **kw: Any,
1971 ) -> _ReflectionInfo:
1972 kw["schema"] = schema
1974 if filter_names and available and len(filter_names) > 100:
1975 fraction = len(filter_names) / len(available)
1976 else:
1977 fraction = None
1979 unreflectable: Dict[TableKey, exc.UnreflectableTableError]
1980 kw["unreflectable"] = unreflectable = {}
1982 has_result: bool = True
1984 def run(
1985 meth: Any,
1986 *,
1987 optional: bool = False,
1988 check_filter_names_from_meth: bool = False,
1989 ) -> Any:
1990 nonlocal has_result
1991 # simple heuristic to improve reflection performance if a
1992 # dialect implements multi_reflection:
1993 # if more than 50% of the tables in the db are in filter_names
1994 # load all the tables, since it's most likely faster to avoid
1995 # a filter on that many tables.
1996 if (
1997 fraction is None
1998 or fraction <= 0.5
1999 or not self.dialect._overrides_default(meth.__name__)
2000 ):
2001 _fn = filter_names
2002 else:
2003 _fn = None
2004 try:
2005 if has_result:
2006 res = meth(filter_names=_fn, **kw)
2007 if check_filter_names_from_meth and not res:
2008 # method returned no result data.
2009 # skip any future call methods
2010 has_result = False
2011 else:
2012 res = {}
2013 except NotImplementedError:
2014 if not optional:
2015 raise
2016 res = {}
2017 return res
2019 info = _ReflectionInfo(
2020 columns=run(
2021 self.get_multi_columns, check_filter_names_from_meth=True
2022 ),
2023 pk_constraint=run(self.get_multi_pk_constraint),
2024 foreign_keys=run(self.get_multi_foreign_keys),
2025 indexes=run(self.get_multi_indexes),
2026 unique_constraints=run(
2027 self.get_multi_unique_constraints, optional=True
2028 ),
2029 table_comment=run(self.get_multi_table_comment, optional=True),
2030 check_constraints=run(
2031 self.get_multi_check_constraints, optional=True
2032 ),
2033 table_options=run(self.get_multi_table_options, optional=True),
2034 unreflectable=unreflectable,
2035 )
2036 if _reflect_info:
2037 _reflect_info.update(info)
2038 return _reflect_info
2039 else:
2040 return info
2043@final
2044class ReflectionDefaults:
2045 """provides blank default values for reflection methods."""
2047 @classmethod
2048 def columns(cls) -> List[ReflectedColumn]:
2049 return []
2051 @classmethod
2052 def pk_constraint(cls) -> ReflectedPrimaryKeyConstraint:
2053 return {
2054 "name": None,
2055 "constrained_columns": [],
2056 }
2058 @classmethod
2059 def foreign_keys(cls) -> List[ReflectedForeignKeyConstraint]:
2060 return []
2062 @classmethod
2063 def indexes(cls) -> List[ReflectedIndex]:
2064 return []
2066 @classmethod
2067 def unique_constraints(cls) -> List[ReflectedUniqueConstraint]:
2068 return []
2070 @classmethod
2071 def check_constraints(cls) -> List[ReflectedCheckConstraint]:
2072 return []
2074 @classmethod
2075 def table_options(cls) -> Dict[str, Any]:
2076 return {}
2078 @classmethod
2079 def table_comment(cls) -> ReflectedTableComment:
2080 return {"text": None}
2083@dataclass
2084class _ReflectionInfo:
2085 columns: Dict[TableKey, List[ReflectedColumn]]
2086 pk_constraint: Dict[TableKey, Optional[ReflectedPrimaryKeyConstraint]]
2087 foreign_keys: Dict[TableKey, List[ReflectedForeignKeyConstraint]]
2088 indexes: Dict[TableKey, List[ReflectedIndex]]
2089 # optionals
2090 unique_constraints: Dict[TableKey, List[ReflectedUniqueConstraint]]
2091 table_comment: Dict[TableKey, Optional[ReflectedTableComment]]
2092 check_constraints: Dict[TableKey, List[ReflectedCheckConstraint]]
2093 table_options: Dict[TableKey, Dict[str, Any]]
2094 unreflectable: Dict[TableKey, exc.UnreflectableTableError]
2096 def update(self, other: _ReflectionInfo) -> None:
2097 for k, v in self.__dict__.items():
2098 ov = getattr(other, k)
2099 if ov is not None:
2100 if v is None:
2101 setattr(self, k, ov)
2102 else:
2103 v.update(ov)