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 final
40from typing import Generator
41from typing import Iterable
42from typing import List
43from typing import Optional
44from typing import Sequence
45from typing import Set
46from typing import Tuple
47from typing import TYPE_CHECKING
48from typing import TypeVar
49from typing import Union
51from .base import Connection
52from .base import Engine
53from .. import exc
54from .. import inspection
55from .. import sql
56from .. import util
57from ..sql import operators
58from ..sql import schema as sa_schema
59from ..sql.cache_key import _ad_hoc_cache_key_from_args
60from ..sql.elements import quoted_name
61from ..sql.elements import TextClause
62from ..sql.type_api import TypeEngine
63from ..sql.visitors import InternalTraversal
64from ..util import topological
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 The caching behavior is dialect-specific: as a general rule single table
192 reflection methods should cache their result, while multi-table reflection
193 methods should not, but behavior may vary.
194 Following DDL operations on the database, it is recommended to either
195 create a new :class:`_reflection.Inspector` instance or to clear the cache
196 of existing instances using the :meth:`.Inspector.clear_cache` method.
198 A :class:`_reflection.Inspector` object is usually created via the
199 :func:`_sa.inspect` function, which may be passed an
200 :class:`_engine.Engine`
201 or a :class:`_engine.Connection`::
203 from sqlalchemy import inspect, create_engine
205 engine = create_engine("...")
206 insp = inspect(engine)
208 Where above, the :class:`~sqlalchemy.engine.interfaces.Dialect` associated
209 with the engine may opt to return an :class:`_reflection.Inspector`
210 subclass that
211 provides additional methods specific to the dialect's target database.
213 """
215 bind: Union[Engine, Connection]
216 engine: Engine
217 _op_context_requires_connect: bool
218 dialect: Dialect
219 info_cache: Dict[Any, Any]
221 @util.deprecated(
222 "1.4",
223 "The __init__() method on :class:`_reflection.Inspector` "
224 "is deprecated and "
225 "will be removed in a future release. Please use the "
226 ":func:`.sqlalchemy.inspect` "
227 "function on an :class:`_engine.Engine` or "
228 ":class:`_engine.Connection` "
229 "in order to "
230 "acquire an :class:`_reflection.Inspector`.",
231 )
232 def __init__(self, bind: Union[Engine, Connection]):
233 """Initialize a new :class:`_reflection.Inspector`.
235 :param bind: a :class:`~sqlalchemy.engine.Connection`,
236 which is typically an instance of
237 :class:`~sqlalchemy.engine.Engine` or
238 :class:`~sqlalchemy.engine.Connection`.
240 For a dialect-specific instance of :class:`_reflection.Inspector`, see
241 :meth:`_reflection.Inspector.from_engine`
243 """
244 self._init_legacy(bind)
246 @classmethod
247 def _construct(
248 cls, init: Callable[..., Any], bind: Union[Engine, Connection]
249 ) -> Inspector:
250 if hasattr(bind.dialect, "inspector"):
251 cls = bind.dialect.inspector
253 self = cls.__new__(cls)
254 init(self, bind)
255 return self
257 def _init_legacy(self, bind: Union[Engine, Connection]) -> None:
258 if hasattr(bind, "exec_driver_sql"):
259 self._init_connection(bind) # type: ignore[arg-type]
260 else:
261 self._init_engine(bind)
263 def _init_engine(self, engine: Engine) -> None:
264 self.bind = self.engine = engine
265 engine.connect().close()
266 self._op_context_requires_connect = True
267 self.dialect = self.engine.dialect
268 self.info_cache = {}
270 def _init_connection(self, connection: Connection) -> None:
271 self.bind = connection
272 self.engine = connection.engine
273 self._op_context_requires_connect = False
274 self.dialect = self.engine.dialect
275 self.info_cache = {}
277 def clear_cache(self) -> None:
278 """Reset the cache for this :class:`.Inspector`.
280 Inspection methods that have data cached will emit SQL queries
281 when next called to get new data.
283 .. versionadded:: 2.0
285 """
286 self.info_cache.clear()
288 @classmethod
289 @util.deprecated(
290 "1.4",
291 "The from_engine() method on :class:`_reflection.Inspector` "
292 "is deprecated and "
293 "will be removed in a future release. Please use the "
294 ":func:`.sqlalchemy.inspect` "
295 "function on an :class:`_engine.Engine` or "
296 ":class:`_engine.Connection` "
297 "in order to "
298 "acquire an :class:`_reflection.Inspector`.",
299 )
300 def from_engine(cls, bind: Engine) -> Inspector:
301 """Construct a new dialect-specific Inspector object from the given
302 engine or connection.
304 :param bind: a :class:`~sqlalchemy.engine.Connection`
305 or :class:`~sqlalchemy.engine.Engine`.
307 This method differs from direct a direct constructor call of
308 :class:`_reflection.Inspector` in that the
309 :class:`~sqlalchemy.engine.interfaces.Dialect` is given a chance to
310 provide a dialect-specific :class:`_reflection.Inspector` instance,
311 which may
312 provide additional methods.
314 See the example at :class:`_reflection.Inspector`.
316 """
317 return cls._construct(cls._init_legacy, bind)
319 @inspection._inspects(Engine)
320 def _engine_insp(bind: Engine) -> Inspector: # type: ignore[misc]
321 return Inspector._construct(Inspector._init_engine, bind)
323 @inspection._inspects(Connection)
324 def _connection_insp(bind: Connection) -> Inspector: # type: ignore[misc]
325 return Inspector._construct(Inspector._init_connection, bind)
327 @contextlib.contextmanager
328 def _operation_context(self) -> Generator[Connection, None, None]:
329 """Return a context that optimizes for multiple operations on a single
330 transaction.
332 This essentially allows connect()/close() to be called if we detected
333 that we're against an :class:`_engine.Engine` and not a
334 :class:`_engine.Connection`.
336 """
337 conn: Connection
338 if self._op_context_requires_connect:
339 conn = self.bind.connect() # type: ignore[union-attr]
340 else:
341 conn = self.bind # type: ignore[assignment]
342 try:
343 yield conn
344 finally:
345 if self._op_context_requires_connect:
346 conn.close()
348 @contextlib.contextmanager
349 def _inspection_context(self) -> Generator[Inspector, None, None]:
350 """Return an :class:`_reflection.Inspector`
351 from this one that will run all
352 operations on a single connection.
354 """
356 with self._operation_context() as conn:
357 sub_insp = self._construct(self.__class__._init_connection, conn)
358 sub_insp.info_cache = self.info_cache
359 yield sub_insp
361 @property
362 def default_schema_name(self) -> Optional[str]:
363 """Return the default schema name presented by the dialect
364 for the current engine's database user.
366 E.g. this is typically ``public`` for PostgreSQL and ``dbo``
367 for SQL Server.
369 """
370 return self.dialect.default_schema_name
372 def get_schema_names(self, **kw: Any) -> List[str]:
373 r"""Return all schema names.
375 :param \**kw: Additional keyword argument to pass to the dialect
376 specific implementation. See the documentation of the dialect
377 in use for more information.
378 """
380 with self._operation_context() as conn:
381 return self.dialect.get_schema_names(
382 conn, info_cache=self.info_cache, **kw
383 )
385 def get_table_names(
386 self, schema: Optional[str] = None, **kw: Any
387 ) -> List[str]:
388 r"""Return all table names within a particular schema.
390 The names are expected to be real tables only, not views.
391 Views are instead returned using the
392 :meth:`_reflection.Inspector.get_view_names` and/or
393 :meth:`_reflection.Inspector.get_materialized_view_names`
394 methods.
396 :param schema: Schema name. If ``schema`` is left at ``None``, the
397 database's default schema is
398 used, else the named schema is searched. If the database does not
399 support named schemas, behavior is undefined if ``schema`` is not
400 passed as ``None``. For special quoting, use :class:`.quoted_name`.
401 :param \**kw: Additional keyword argument to pass to the dialect
402 specific implementation. See the documentation of the dialect
403 in use for more information.
405 .. seealso::
407 :meth:`_reflection.Inspector.get_sorted_table_and_fkc_names`
409 :attr:`_schema.MetaData.sorted_tables`
411 """
413 with self._operation_context() as conn:
414 return self.dialect.get_table_names(
415 conn, schema, info_cache=self.info_cache, **kw
416 )
418 def has_table(
419 self, table_name: str, schema: Optional[str] = None, **kw: Any
420 ) -> bool:
421 r"""Return True if the backend has a table, view, or temporary
422 table of the given name.
424 :param table_name: name of the table to check
425 :param schema: schema name to query, if not the default schema.
426 :param \**kw: Additional keyword argument to pass to the dialect
427 specific implementation. See the documentation of the dialect
428 in use for more information.
430 .. versionadded:: 1.4 - the :meth:`.Inspector.has_table` method
431 replaces the :meth:`_engine.Engine.has_table` method.
433 .. versionchanged:: 2.0:: :meth:`.Inspector.has_table` now formally
434 supports checking for additional table-like objects:
436 * any type of views (plain or materialized)
437 * temporary tables of any kind
439 Previously, these two checks were not formally specified and
440 different dialects would vary in their behavior. The dialect
441 testing suite now includes tests for all of these object types
442 and should be supported by all SQLAlchemy-included dialects.
443 Support among third party dialects may be lagging, however.
445 """
446 with self._operation_context() as conn:
447 return self.dialect.has_table(
448 conn, table_name, schema, info_cache=self.info_cache, **kw
449 )
451 def has_multi_table(
452 self,
453 table_names: Sequence[str],
454 schema: Optional[str] = None,
455 **kw: Any,
456 ) -> Dict[TableKey, bool]:
457 r"""Return a dict indicating for each table name whether the backend
458 has a table, view, or temporary table with the given name.
460 :param table_names: sequence of table names to check.
461 :param schema: schema name to query, if not the default schema.
462 :param \**kw: Additional keyword argument to pass to the dialect
463 specific implementation. See the documentation of the dialect
464 in use for more information.
466 .. versionadded:: 2.1
468 .. seealso:: :meth:`Inspector.has_table`
470 """
471 with self._operation_context() as conn:
472 return dict(
473 self.dialect.has_multi_table(
474 conn, table_names, schema, info_cache=self.info_cache, **kw
475 )
476 )
478 def has_sequence(
479 self, sequence_name: str, schema: Optional[str] = None, **kw: Any
480 ) -> bool:
481 r"""Return True if the backend has a sequence with the given name.
483 :param sequence_name: name of the sequence to check
484 :param schema: schema name to query, if not the default schema.
485 :param \**kw: Additional keyword argument to pass to the dialect
486 specific implementation. See the documentation of the dialect
487 in use for more information.
489 .. versionadded:: 1.4
491 """
492 with self._operation_context() as conn:
493 return self.dialect.has_sequence(
494 conn, sequence_name, schema, info_cache=self.info_cache, **kw
495 )
497 def has_index(
498 self,
499 table_name: str,
500 index_name: str,
501 schema: Optional[str] = None,
502 **kw: Any,
503 ) -> bool:
504 r"""Check the existence of a particular index name in the database.
506 :param table_name: the name of the table the index belongs to
507 :param index_name: the name of the index to check
508 :param schema: schema name to query, if not the default schema.
509 :param \**kw: Additional keyword argument to pass to the dialect
510 specific implementation. See the documentation of the dialect
511 in use for more information.
513 .. versionadded:: 2.0
515 """
516 with self._operation_context() as conn:
517 return self.dialect.has_index(
518 conn,
519 table_name,
520 index_name,
521 schema,
522 info_cache=self.info_cache,
523 **kw,
524 )
526 def has_schema(self, schema_name: str, **kw: Any) -> bool:
527 r"""Return True if the backend has a schema with the given name.
529 :param schema_name: name of the schema to check
530 :param \**kw: Additional keyword argument to pass to the dialect
531 specific implementation. See the documentation of the dialect
532 in use for more information.
534 .. versionadded:: 2.0
536 """
537 with self._operation_context() as conn:
538 return self.dialect.has_schema(
539 conn, schema_name, info_cache=self.info_cache, **kw
540 )
542 def get_sorted_table_and_fkc_names(
543 self,
544 schema: Optional[str] = None,
545 **kw: Any,
546 ) -> List[Tuple[Optional[str], List[Tuple[str, Optional[str]]]]]:
547 r"""Return dependency-sorted table and foreign key constraint names in
548 referred to within a particular schema.
550 This will yield 2-tuples of
551 ``(tablename, [(tname, fkname), (tname, fkname), ...])``
552 consisting of table names in CREATE order grouped with the foreign key
553 constraint names that are not detected as belonging to a cycle.
554 The final element
555 will be ``(None, [(tname, fkname), (tname, fkname), ..])``
556 which will consist of remaining
557 foreign key constraint names that would require a separate CREATE
558 step after-the-fact, based on dependencies between tables.
560 :param schema: schema name to query, if not the default schema.
561 :param \**kw: Additional keyword argument to pass to the dialect
562 specific implementation. See the documentation of the dialect
563 in use for more information.
565 .. seealso::
567 :meth:`_reflection.Inspector.get_table_names`
569 :func:`.sort_tables_and_constraints` - similar method which works
570 with an already-given :class:`_schema.MetaData`.
572 """
574 return [
575 (
576 table_key[1] if table_key else None,
577 [(tname, fks) for (_, tname), fks in fk_collection],
578 )
579 for (
580 table_key,
581 fk_collection,
582 ) in self.sort_tables_on_foreign_key_dependency(
583 consider_schemas=(schema,)
584 )
585 ]
587 def sort_tables_on_foreign_key_dependency(
588 self,
589 consider_schemas: Collection[Optional[str]] = (None,),
590 **kw: Any,
591 ) -> List[
592 Tuple[
593 Optional[Tuple[Optional[str], str]],
594 List[Tuple[Tuple[Optional[str], str], Optional[str]]],
595 ]
596 ]:
597 r"""Return dependency-sorted table and foreign key constraint names
598 referred to within multiple schemas.
600 This method may be compared to
601 :meth:`.Inspector.get_sorted_table_and_fkc_names`, which
602 works on one schema at a time; here, the method is a generalization
603 that will consider multiple schemas at once including that it will
604 resolve for cross-schema foreign keys.
606 .. versionadded:: 2.0
608 """
609 SchemaTab = Tuple[Optional[str], str]
611 tuples: Set[Tuple[SchemaTab, SchemaTab]] = set()
612 remaining_fkcs: Set[Tuple[SchemaTab, Optional[str]]] = set()
613 fknames_for_table: Dict[SchemaTab, Set[Optional[str]]] = {}
614 tnames: List[SchemaTab] = []
616 for schname in consider_schemas:
617 schema_fkeys = self.get_multi_foreign_keys(schname, **kw)
618 tnames.extend(schema_fkeys)
619 for (_, tname), fkeys in schema_fkeys.items():
620 fknames_for_table[(schname, tname)] = {
621 fk["name"] for fk in fkeys
622 }
623 for fkey in fkeys:
624 if (
625 tname != fkey["referred_table"]
626 or schname != fkey["referred_schema"]
627 ):
628 tuples.add(
629 (
630 (
631 fkey["referred_schema"],
632 fkey["referred_table"],
633 ),
634 (schname, tname),
635 )
636 )
637 try:
638 candidate_sort = list(topological.sort(tuples, tnames))
639 except exc.CircularDependencyError as err:
640 edge: Tuple[SchemaTab, SchemaTab]
641 for edge in err.edges:
642 tuples.remove(edge)
643 remaining_fkcs.update(
644 (edge[1], fkc) for fkc in fknames_for_table[edge[1]]
645 )
647 candidate_sort = list(topological.sort(tuples, tnames))
648 ret: List[
649 Tuple[Optional[SchemaTab], List[Tuple[SchemaTab, Optional[str]]]]
650 ]
651 ret = [
652 (
653 (schname, tname),
654 [
655 ((schname, tname), fk)
656 for fk in fknames_for_table[(schname, tname)].difference(
657 name for _, name in remaining_fkcs
658 )
659 ],
660 )
661 for (schname, tname) in candidate_sort
662 ]
663 return ret + [(None, list(remaining_fkcs))]
665 def get_temp_table_names(self, **kw: Any) -> List[str]:
666 r"""Return a list of temporary table names for the current bind.
668 This method is unsupported by most dialects; currently
669 only Oracle Database, PostgreSQL and SQLite implements it.
671 :param \**kw: Additional keyword argument to pass to the dialect
672 specific implementation. See the documentation of the dialect
673 in use for more information.
675 """
677 with self._operation_context() as conn:
678 return self.dialect.get_temp_table_names(
679 conn, info_cache=self.info_cache, **kw
680 )
682 def get_temp_view_names(self, **kw: Any) -> List[str]:
683 r"""Return a list of temporary view names for the current bind.
685 This method is unsupported by most dialects; currently
686 only PostgreSQL and SQLite implements it.
688 :param \**kw: Additional keyword argument to pass to the dialect
689 specific implementation. See the documentation of the dialect
690 in use for more information.
692 """
693 with self._operation_context() as conn:
694 return self.dialect.get_temp_view_names(
695 conn, info_cache=self.info_cache, **kw
696 )
698 def get_table_options(
699 self, table_name: str, schema: Optional[str] = None, **kw: Any
700 ) -> Dict[str, Any]:
701 r"""Return a dictionary of options specified when the table of the
702 given name was created.
704 This currently includes some options that apply to MySQL and Oracle
705 Database tables.
707 :param table_name: string name of the table. For special quoting,
708 use :class:`.quoted_name`.
710 :param schema: string schema name; if omitted, uses the default schema
711 of the database connection. For special quoting,
712 use :class:`.quoted_name`.
714 :param \**kw: Additional keyword argument to pass to the dialect
715 specific implementation. See the documentation of the dialect
716 in use for more information.
718 :return: a dict with the table options. The returned keys depend on the
719 dialect in use. Each one is prefixed with the dialect name.
721 .. seealso:: :meth:`Inspector.get_multi_table_options`
723 """
724 with self._operation_context() as conn:
725 return self.dialect.get_table_options(
726 conn, table_name, schema, info_cache=self.info_cache, **kw
727 )
729 def get_multi_table_options(
730 self,
731 schema: Optional[str] = None,
732 filter_names: Optional[Sequence[str]] = None,
733 kind: ObjectKind = ObjectKind.TABLE,
734 scope: ObjectScope = ObjectScope.DEFAULT,
735 **kw: Any,
736 ) -> Dict[TableKey, Dict[str, Any]]:
737 r"""Return a dictionary of options specified when the tables in the
738 given schema were created.
740 The tables can be filtered by passing the names to use to
741 ``filter_names``.
743 This currently includes some options that apply to MySQL and Oracle
744 tables.
746 :param schema: string schema name; if omitted, uses the default schema
747 of the database connection. For special quoting,
748 use :class:`.quoted_name`.
750 :param filter_names: optionally return information only for the
751 objects listed here.
753 :param kind: a :class:`.ObjectKind` that specifies the type of objects
754 to reflect. Defaults to ``ObjectKind.TABLE``.
756 :param scope: a :class:`.ObjectScope` that specifies if options of
757 default, temporary or any tables should be reflected.
758 Defaults to ``ObjectScope.DEFAULT``.
760 :param \**kw: Additional keyword argument to pass to the dialect
761 specific implementation. See the documentation of the dialect
762 in use for more information.
764 :return: a dictionary where the keys are two-tuple schema,table-name
765 and the values are dictionaries with the table options.
766 The returned keys in each dict depend on the
767 dialect in use. Each one is prefixed with the dialect name.
768 The schema is ``None`` if no schema is provided.
770 .. versionadded:: 2.0
772 .. seealso:: :meth:`Inspector.get_table_options`
773 """
774 with self._operation_context() as conn:
775 res = self.dialect.get_multi_table_options(
776 conn,
777 schema=schema,
778 filter_names=filter_names,
779 kind=kind,
780 scope=scope,
781 info_cache=self.info_cache,
782 **kw,
783 )
784 return dict(res)
786 def get_view_names(
787 self, schema: Optional[str] = None, **kw: Any
788 ) -> List[str]:
789 r"""Return all non-materialized view names in `schema`.
791 :param schema: Optional, retrieve names from a non-default schema.
792 For special quoting, use :class:`.quoted_name`.
793 :param \**kw: Additional keyword argument to pass to the dialect
794 specific implementation. See the documentation of the dialect
795 in use for more information.
798 .. versionchanged:: 2.0 For those dialects that previously included
799 the names of materialized views in this list (currently PostgreSQL),
800 this method no longer returns the names of materialized views.
801 the :meth:`.Inspector.get_materialized_view_names` method should
802 be used instead.
804 .. seealso::
806 :meth:`.Inspector.get_materialized_view_names`
808 """
810 with self._operation_context() as conn:
811 return self.dialect.get_view_names(
812 conn, schema, info_cache=self.info_cache, **kw
813 )
815 def get_materialized_view_names(
816 self, schema: Optional[str] = None, **kw: Any
817 ) -> List[str]:
818 r"""Return all materialized view names in `schema`.
820 :param schema: Optional, retrieve names from a non-default schema.
821 For special quoting, use :class:`.quoted_name`.
822 :param \**kw: Additional keyword argument to pass to the dialect
823 specific implementation. See the documentation of the dialect
824 in use for more information.
826 .. versionadded:: 2.0
828 .. seealso::
830 :meth:`.Inspector.get_view_names`
832 """
834 with self._operation_context() as conn:
835 return self.dialect.get_materialized_view_names(
836 conn, schema, info_cache=self.info_cache, **kw
837 )
839 def get_sequence_names(
840 self, schema: Optional[str] = None, **kw: Any
841 ) -> List[str]:
842 r"""Return all sequence names in `schema`.
844 :param schema: Optional, retrieve names from a non-default schema.
845 For special quoting, use :class:`.quoted_name`.
846 :param \**kw: Additional keyword argument to pass to the dialect
847 specific implementation. See the documentation of the dialect
848 in use for more information.
850 """
852 with self._operation_context() as conn:
853 return self.dialect.get_sequence_names(
854 conn, schema, info_cache=self.info_cache, **kw
855 )
857 def get_view_definition(
858 self, view_name: str, schema: Optional[str] = None, **kw: Any
859 ) -> str:
860 r"""Return definition for the plain or materialized view called
861 ``view_name``.
863 :param view_name: Name of the view.
864 :param schema: Optional, retrieve names from a non-default schema.
865 For special quoting, use :class:`.quoted_name`.
866 :param \**kw: Additional keyword argument to pass to the dialect
867 specific implementation. See the documentation of the dialect
868 in use for more information.
870 """
872 with self._operation_context() as conn:
873 return self.dialect.get_view_definition(
874 conn, view_name, schema, info_cache=self.info_cache, **kw
875 )
877 def get_columns(
878 self, table_name: str, schema: Optional[str] = None, **kw: Any
879 ) -> List[ReflectedColumn]:
880 r"""Return information about columns in ``table_name``.
882 Given a string ``table_name`` and an optional string ``schema``,
883 return column information as a list of :class:`.ReflectedColumn`.
885 :param table_name: string name of the table. For special quoting,
886 use :class:`.quoted_name`.
888 :param schema: string schema name; if omitted, uses the default schema
889 of the database connection. For special quoting,
890 use :class:`.quoted_name`.
892 :param \**kw: Additional keyword argument to pass to the dialect
893 specific implementation. See the documentation of the dialect
894 in use for more information.
896 :return: list of dictionaries, each representing the definition of
897 a database column.
899 .. seealso:: :meth:`Inspector.get_multi_columns`.
901 """
903 with self._operation_context() as conn:
904 col_defs = self.dialect.get_columns(
905 conn, table_name, schema, info_cache=self.info_cache, **kw
906 )
907 if col_defs:
908 self._instantiate_types([col_defs])
909 return col_defs
911 def _instantiate_types(
912 self, data: Iterable[List[ReflectedColumn]]
913 ) -> None:
914 # make this easy and only return instances for coltype
915 for col_defs in data:
916 for col_def in col_defs:
917 coltype = col_def["type"]
918 if not isinstance(coltype, TypeEngine):
919 col_def["type"] = coltype()
921 def get_multi_columns(
922 self,
923 schema: Optional[str] = None,
924 filter_names: Optional[Sequence[str]] = None,
925 kind: ObjectKind = ObjectKind.TABLE,
926 scope: ObjectScope = ObjectScope.DEFAULT,
927 **kw: Any,
928 ) -> Dict[TableKey, List[ReflectedColumn]]:
929 r"""Return information about columns in all objects in the given
930 schema.
932 The objects can be filtered by passing the names to use to
933 ``filter_names``.
935 For each table the value is a list of :class:`.ReflectedColumn`.
937 :param schema: string schema name; if omitted, uses the default schema
938 of the database connection. For special quoting,
939 use :class:`.quoted_name`.
941 :param filter_names: optionally return information only for the
942 objects listed here.
944 :param kind: a :class:`.ObjectKind` that specifies the type of objects
945 to reflect. Defaults to ``ObjectKind.TABLE``.
947 :param scope: a :class:`.ObjectScope` that specifies if columns of
948 default, temporary or any tables should be reflected.
949 Defaults to ``ObjectScope.DEFAULT``.
951 :param \**kw: Additional keyword argument to pass to the dialect
952 specific implementation. See the documentation of the dialect
953 in use for more information.
955 :return: a dictionary where the keys are two-tuple schema,table-name
956 and the values are list of dictionaries, each representing the
957 definition of a database column.
958 The schema is ``None`` if no schema is provided.
960 .. versionadded:: 2.0
962 .. seealso:: :meth:`Inspector.get_columns`
963 """
965 with self._operation_context() as conn:
966 table_col_defs = dict(
967 self.dialect.get_multi_columns(
968 conn,
969 schema=schema,
970 filter_names=filter_names,
971 kind=kind,
972 scope=scope,
973 info_cache=self.info_cache,
974 **kw,
975 )
976 )
977 self._instantiate_types(table_col_defs.values())
978 return table_col_defs
980 def get_pk_constraint(
981 self, table_name: str, schema: Optional[str] = None, **kw: Any
982 ) -> ReflectedPrimaryKeyConstraint:
983 r"""Return information about primary key constraint in ``table_name``.
985 Given a string ``table_name``, and an optional string `schema`, return
986 primary key information as a :class:`.ReflectedPrimaryKeyConstraint`.
988 :param table_name: string name of the table. For special quoting,
989 use :class:`.quoted_name`.
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 \**kw: Additional keyword argument to pass to the dialect
996 specific implementation. See the documentation of the dialect
997 in use for more information.
999 :return: a dictionary representing the definition of
1000 a primary key constraint.
1002 .. seealso:: :meth:`Inspector.get_multi_pk_constraint`
1003 """
1004 with self._operation_context() as conn:
1005 return self.dialect.get_pk_constraint(
1006 conn, table_name, schema, info_cache=self.info_cache, **kw
1007 )
1009 def get_multi_pk_constraint(
1010 self,
1011 schema: Optional[str] = None,
1012 filter_names: Optional[Sequence[str]] = None,
1013 kind: ObjectKind = ObjectKind.TABLE,
1014 scope: ObjectScope = ObjectScope.DEFAULT,
1015 **kw: Any,
1016 ) -> Dict[TableKey, ReflectedPrimaryKeyConstraint]:
1017 r"""Return information about primary key constraints in
1018 all tables in the given schema.
1020 The tables can be filtered by passing the names to use to
1021 ``filter_names``.
1023 For each table the value is a :class:`.ReflectedPrimaryKeyConstraint`.
1025 :param schema: string schema name; if omitted, uses the default schema
1026 of the database connection. For special quoting,
1027 use :class:`.quoted_name`.
1029 :param filter_names: optionally return information only for the
1030 objects listed here.
1032 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1033 to reflect. Defaults to ``ObjectKind.TABLE``.
1035 :param scope: a :class:`.ObjectScope` that specifies if primary keys of
1036 default, temporary or any tables should be reflected.
1037 Defaults to ``ObjectScope.DEFAULT``.
1039 :param \**kw: Additional keyword argument to pass to the dialect
1040 specific implementation. See the documentation of the dialect
1041 in use for more information.
1043 :return: a dictionary where the keys are two-tuple schema,table-name
1044 and the values are dictionaries, each representing the
1045 definition of a primary key constraint.
1046 The schema is ``None`` if no schema is provided.
1048 .. versionadded:: 2.0
1050 .. seealso:: :meth:`Inspector.get_pk_constraint`
1051 """
1052 with self._operation_context() as conn:
1053 return dict(
1054 self.dialect.get_multi_pk_constraint(
1055 conn,
1056 schema=schema,
1057 filter_names=filter_names,
1058 kind=kind,
1059 scope=scope,
1060 info_cache=self.info_cache,
1061 **kw,
1062 )
1063 )
1065 def get_foreign_keys(
1066 self, table_name: str, schema: Optional[str] = None, **kw: Any
1067 ) -> List[ReflectedForeignKeyConstraint]:
1068 r"""Return information about foreign_keys in ``table_name``.
1070 Given a string ``table_name``, and an optional string `schema`, return
1071 foreign key information as a list of
1072 :class:`.ReflectedForeignKeyConstraint`.
1074 :param table_name: string name of the table. For special quoting,
1075 use :class:`.quoted_name`.
1077 :param schema: string schema name; if omitted, uses the default schema
1078 of the database connection. For special quoting,
1079 use :class:`.quoted_name`.
1081 :param \**kw: Additional keyword argument to pass to the dialect
1082 specific implementation. See the documentation of the dialect
1083 in use for more information.
1085 :return: a list of dictionaries, each representing the
1086 a foreign key definition.
1088 .. seealso:: :meth:`Inspector.get_multi_foreign_keys`
1089 """
1091 with self._operation_context() as conn:
1092 return self.dialect.get_foreign_keys(
1093 conn, table_name, schema, info_cache=self.info_cache, **kw
1094 )
1096 def get_multi_foreign_keys(
1097 self,
1098 schema: Optional[str] = None,
1099 filter_names: Optional[Sequence[str]] = None,
1100 kind: ObjectKind = ObjectKind.TABLE,
1101 scope: ObjectScope = ObjectScope.DEFAULT,
1102 **kw: Any,
1103 ) -> Dict[TableKey, List[ReflectedForeignKeyConstraint]]:
1104 r"""Return information about foreign_keys in all tables
1105 in the given schema.
1107 The tables can be filtered by passing the names to use to
1108 ``filter_names``.
1110 For each table the value is a list of
1111 :class:`.ReflectedForeignKeyConstraint`.
1113 :param schema: string schema name; if omitted, uses the default schema
1114 of the database connection. For special quoting,
1115 use :class:`.quoted_name`.
1117 :param filter_names: optionally return information only for the
1118 objects listed here.
1120 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1121 to reflect. Defaults to ``ObjectKind.TABLE``.
1123 :param scope: a :class:`.ObjectScope` that specifies if foreign keys of
1124 default, temporary or any tables should be reflected.
1125 Defaults to ``ObjectScope.DEFAULT``.
1127 :param \**kw: Additional keyword argument to pass to the dialect
1128 specific implementation. See the documentation of the dialect
1129 in use for more information.
1131 :return: a dictionary where the keys are two-tuple schema,table-name
1132 and the values are list of dictionaries, each representing
1133 a foreign key definition.
1134 The schema is ``None`` if no schema is provided.
1136 .. versionadded:: 2.0
1138 .. seealso:: :meth:`Inspector.get_foreign_keys`
1139 """
1141 with self._operation_context() as conn:
1142 return dict(
1143 self.dialect.get_multi_foreign_keys(
1144 conn,
1145 schema=schema,
1146 filter_names=filter_names,
1147 kind=kind,
1148 scope=scope,
1149 info_cache=self.info_cache,
1150 **kw,
1151 )
1152 )
1154 def get_indexes(
1155 self, table_name: str, schema: Optional[str] = None, **kw: Any
1156 ) -> List[ReflectedIndex]:
1157 r"""Return information about indexes in ``table_name``.
1159 Given a string ``table_name`` and an optional string `schema`, return
1160 index information as a list of :class:`.ReflectedIndex`.
1162 :param table_name: string name of the table. For special quoting,
1163 use :class:`.quoted_name`.
1165 :param schema: string schema name; if omitted, uses the default schema
1166 of the database connection. For special quoting,
1167 use :class:`.quoted_name`.
1169 :param \**kw: Additional keyword argument to pass to the dialect
1170 specific implementation. See the documentation of the dialect
1171 in use for more information.
1173 :return: a list of dictionaries, each representing the
1174 definition of an index.
1176 .. seealso:: :meth:`Inspector.get_multi_indexes`
1177 """
1179 with self._operation_context() as conn:
1180 return self.dialect.get_indexes(
1181 conn, table_name, schema, info_cache=self.info_cache, **kw
1182 )
1184 def get_multi_indexes(
1185 self,
1186 schema: Optional[str] = None,
1187 filter_names: Optional[Sequence[str]] = None,
1188 kind: ObjectKind = ObjectKind.TABLE,
1189 scope: ObjectScope = ObjectScope.DEFAULT,
1190 **kw: Any,
1191 ) -> Dict[TableKey, List[ReflectedIndex]]:
1192 r"""Return information about indexes in in all objects
1193 in the given schema.
1195 The objects can be filtered by passing the names to use to
1196 ``filter_names``.
1198 For each table the value is a list of :class:`.ReflectedIndex`.
1200 :param schema: string schema name; if omitted, uses the default schema
1201 of the database connection. For special quoting,
1202 use :class:`.quoted_name`.
1204 :param filter_names: optionally return information only for the
1205 objects listed here.
1207 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1208 to reflect. Defaults to ``ObjectKind.TABLE``.
1210 :param scope: a :class:`.ObjectScope` that specifies if indexes of
1211 default, temporary or any tables should be reflected.
1212 Defaults to ``ObjectScope.DEFAULT``.
1214 :param \**kw: Additional keyword argument to pass to the dialect
1215 specific implementation. See the documentation of the dialect
1216 in use for more information.
1218 :return: a dictionary where the keys are two-tuple schema,table-name
1219 and the values are list of dictionaries, each representing the
1220 definition of an index.
1221 The schema is ``None`` if no schema is provided.
1223 .. versionadded:: 2.0
1225 .. seealso:: :meth:`Inspector.get_indexes`
1226 """
1228 with self._operation_context() as conn:
1229 return dict(
1230 self.dialect.get_multi_indexes(
1231 conn,
1232 schema=schema,
1233 filter_names=filter_names,
1234 kind=kind,
1235 scope=scope,
1236 info_cache=self.info_cache,
1237 **kw,
1238 )
1239 )
1241 def get_unique_constraints(
1242 self, table_name: str, schema: Optional[str] = None, **kw: Any
1243 ) -> List[ReflectedUniqueConstraint]:
1244 r"""Return information about unique constraints in ``table_name``.
1246 Given a string ``table_name`` and an optional string `schema`, return
1247 unique constraint information as a list of
1248 :class:`.ReflectedUniqueConstraint`.
1250 :param table_name: string name of the table. For special quoting,
1251 use :class:`.quoted_name`.
1253 :param schema: string schema name; if omitted, uses the default schema
1254 of the database connection. For special quoting,
1255 use :class:`.quoted_name`.
1257 :param \**kw: Additional keyword argument to pass to the dialect
1258 specific implementation. See the documentation of the dialect
1259 in use for more information.
1261 :return: a list of dictionaries, each representing the
1262 definition of an unique constraint.
1264 .. seealso:: :meth:`Inspector.get_multi_unique_constraints`
1265 """
1267 with self._operation_context() as conn:
1268 return self.dialect.get_unique_constraints(
1269 conn, table_name, schema, info_cache=self.info_cache, **kw
1270 )
1272 def get_multi_unique_constraints(
1273 self,
1274 schema: Optional[str] = None,
1275 filter_names: Optional[Sequence[str]] = None,
1276 kind: ObjectKind = ObjectKind.TABLE,
1277 scope: ObjectScope = ObjectScope.DEFAULT,
1278 **kw: Any,
1279 ) -> Dict[TableKey, List[ReflectedUniqueConstraint]]:
1280 r"""Return information about unique constraints in all tables
1281 in the given schema.
1283 The tables can be filtered by passing the names to use to
1284 ``filter_names``.
1286 For each table the value is a list of
1287 :class:`.ReflectedUniqueConstraint`.
1289 :param schema: string schema name; if omitted, uses the default schema
1290 of the database connection. For special quoting,
1291 use :class:`.quoted_name`.
1293 :param filter_names: optionally return information only for the
1294 objects listed here.
1296 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1297 to reflect. Defaults to ``ObjectKind.TABLE``.
1299 :param scope: a :class:`.ObjectScope` that specifies if constraints of
1300 default, temporary or any tables should be reflected.
1301 Defaults to ``ObjectScope.DEFAULT``.
1303 :param \**kw: Additional keyword argument to pass to the dialect
1304 specific implementation. See the documentation of the dialect
1305 in use for more information.
1307 :return: a dictionary where the keys are two-tuple schema,table-name
1308 and the values are list of dictionaries, each representing the
1309 definition of an unique constraint.
1310 The schema is ``None`` if no schema is provided.
1312 .. versionadded:: 2.0
1314 .. seealso:: :meth:`Inspector.get_unique_constraints`
1315 """
1317 with self._operation_context() as conn:
1318 return dict(
1319 self.dialect.get_multi_unique_constraints(
1320 conn,
1321 schema=schema,
1322 filter_names=filter_names,
1323 kind=kind,
1324 scope=scope,
1325 info_cache=self.info_cache,
1326 **kw,
1327 )
1328 )
1330 def get_table_comment(
1331 self, table_name: str, schema: Optional[str] = None, **kw: Any
1332 ) -> ReflectedTableComment:
1333 r"""Return information about the table comment for ``table_name``.
1335 Given a string ``table_name`` and an optional string ``schema``,
1336 return table comment information as a :class:`.ReflectedTableComment`.
1338 Raises ``NotImplementedError`` for a dialect that does not support
1339 comments.
1341 :param table_name: string name of the table. For special quoting,
1342 use :class:`.quoted_name`.
1344 :param schema: string schema name; if omitted, uses the default schema
1345 of the database connection. For special quoting,
1346 use :class:`.quoted_name`.
1348 :param \**kw: Additional keyword argument to pass to the dialect
1349 specific implementation. See the documentation of the dialect
1350 in use for more information.
1352 :return: a dictionary, with the table comment.
1354 .. seealso:: :meth:`Inspector.get_multi_table_comment`
1355 """
1357 with self._operation_context() as conn:
1358 return self.dialect.get_table_comment(
1359 conn, table_name, schema, info_cache=self.info_cache, **kw
1360 )
1362 def get_multi_table_comment(
1363 self,
1364 schema: Optional[str] = None,
1365 filter_names: Optional[Sequence[str]] = None,
1366 kind: ObjectKind = ObjectKind.TABLE,
1367 scope: ObjectScope = ObjectScope.DEFAULT,
1368 **kw: Any,
1369 ) -> Dict[TableKey, ReflectedTableComment]:
1370 r"""Return information about the table comment in all objects
1371 in the given schema.
1373 The objects can be filtered by passing the names to use to
1374 ``filter_names``.
1376 For each table the value is a :class:`.ReflectedTableComment`.
1378 Raises ``NotImplementedError`` for a dialect that does not support
1379 comments.
1381 :param schema: string schema name; if omitted, uses the default schema
1382 of the database connection. For special quoting,
1383 use :class:`.quoted_name`.
1385 :param filter_names: optionally return information only for the
1386 objects listed here.
1388 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1389 to reflect. Defaults to ``ObjectKind.TABLE``.
1391 :param scope: a :class:`.ObjectScope` that specifies if comments of
1392 default, temporary or any tables should be reflected.
1393 Defaults to ``ObjectScope.DEFAULT``.
1395 :param \**kw: Additional keyword argument to pass to the dialect
1396 specific implementation. See the documentation of the dialect
1397 in use for more information.
1399 :return: a dictionary where the keys are two-tuple schema,table-name
1400 and the values are dictionaries, representing the
1401 table comments.
1402 The schema is ``None`` if no schema is provided.
1404 .. versionadded:: 2.0
1406 .. seealso:: :meth:`Inspector.get_table_comment`
1407 """
1409 with self._operation_context() as conn:
1410 return dict(
1411 self.dialect.get_multi_table_comment(
1412 conn,
1413 schema=schema,
1414 filter_names=filter_names,
1415 kind=kind,
1416 scope=scope,
1417 info_cache=self.info_cache,
1418 **kw,
1419 )
1420 )
1422 def get_check_constraints(
1423 self, table_name: str, schema: Optional[str] = None, **kw: Any
1424 ) -> List[ReflectedCheckConstraint]:
1425 r"""Return information about check constraints in ``table_name``.
1427 Given a string ``table_name`` and an optional string `schema`, return
1428 check constraint information as a list of
1429 :class:`.ReflectedCheckConstraint`.
1431 :param table_name: string name of the table. For special quoting,
1432 use :class:`.quoted_name`.
1434 :param schema: string schema name; if omitted, uses the default schema
1435 of the database connection. For special quoting,
1436 use :class:`.quoted_name`.
1438 :param \**kw: Additional keyword argument to pass to the dialect
1439 specific implementation. See the documentation of the dialect
1440 in use for more information.
1442 :return: a list of dictionaries, each representing the
1443 definition of a check constraints.
1445 .. seealso:: :meth:`Inspector.get_multi_check_constraints`
1446 """
1448 with self._operation_context() as conn:
1449 return self.dialect.get_check_constraints(
1450 conn, table_name, schema, info_cache=self.info_cache, **kw
1451 )
1453 def get_multi_check_constraints(
1454 self,
1455 schema: Optional[str] = None,
1456 filter_names: Optional[Sequence[str]] = None,
1457 kind: ObjectKind = ObjectKind.TABLE,
1458 scope: ObjectScope = ObjectScope.DEFAULT,
1459 **kw: Any,
1460 ) -> Dict[TableKey, List[ReflectedCheckConstraint]]:
1461 r"""Return information about check constraints in all tables
1462 in the given schema.
1464 The tables can be filtered by passing the names to use to
1465 ``filter_names``.
1467 For each table the value is a list of
1468 :class:`.ReflectedCheckConstraint`.
1470 :param schema: string schema name; if omitted, uses the default schema
1471 of the database connection. For special quoting,
1472 use :class:`.quoted_name`.
1474 :param filter_names: optionally return information only for the
1475 objects listed here.
1477 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1478 to reflect. Defaults to ``ObjectKind.TABLE``.
1480 :param scope: a :class:`.ObjectScope` that specifies if constraints of
1481 default, temporary or any tables should be reflected.
1482 Defaults to ``ObjectScope.DEFAULT``.
1484 :param \**kw: Additional keyword argument to pass to the dialect
1485 specific implementation. See the documentation of the dialect
1486 in use for more information.
1488 :return: a dictionary where the keys are two-tuple schema,table-name
1489 and the values are list of dictionaries, each representing the
1490 definition of a check constraints.
1491 The schema is ``None`` if no schema is provided.
1493 .. versionadded:: 2.0
1495 .. seealso:: :meth:`Inspector.get_check_constraints`
1496 """
1498 with self._operation_context() as conn:
1499 return dict(
1500 self.dialect.get_multi_check_constraints(
1501 conn,
1502 schema=schema,
1503 filter_names=filter_names,
1504 kind=kind,
1505 scope=scope,
1506 info_cache=self.info_cache,
1507 **kw,
1508 )
1509 )
1511 def reflect_table(
1512 self,
1513 table: sa_schema.Table,
1514 include_columns: Optional[Collection[str]],
1515 exclude_columns: Collection[str] = (),
1516 resolve_fks: bool = True,
1517 _extend_on: Optional[Set[sa_schema.Table]] = None,
1518 _reflect_info: Optional[_ReflectionInfo] = None,
1519 ) -> None:
1520 """Given a :class:`_schema.Table` object, load its internal
1521 constructs based on introspection.
1523 This is the underlying method used by most dialects to produce
1524 table reflection. Direct usage is like::
1526 from sqlalchemy import create_engine, MetaData, Table
1527 from sqlalchemy import inspect
1529 engine = create_engine("...")
1530 meta = MetaData()
1531 user_table = Table("user", meta)
1532 insp = inspect(engine)
1533 insp.reflect_table(user_table, None)
1535 .. versionchanged:: 1.4 Renamed from ``reflecttable`` to
1536 ``reflect_table``
1538 :param table: a :class:`~sqlalchemy.schema.Table` instance.
1539 :param include_columns: a list of string column names to include
1540 in the reflection process. If ``None``, all columns are reflected.
1542 """
1544 if _extend_on is not None:
1545 if table in _extend_on:
1546 return
1547 else:
1548 _extend_on.add(table)
1550 dialect = self.bind.dialect
1552 with self._operation_context() as conn:
1553 schema = conn.schema_for_object(table)
1555 table_name = table.name
1557 # get table-level arguments that are specifically
1558 # intended for reflection, e.g. oracle_resolve_synonyms.
1559 # these are unconditionally passed to related Table
1560 # objects
1561 reflection_options = {
1562 k: table.dialect_kwargs.get(k)
1563 for k in dialect.reflection_options
1564 if k in table.dialect_kwargs
1565 }
1567 table_key = (schema, table_name)
1568 if _reflect_info is None or table_key not in _reflect_info.columns:
1569 _reflect_info = self._get_reflection_info(
1570 schema,
1571 filter_names=[table_name],
1572 kind=ObjectKind.ANY,
1573 scope=ObjectScope.ANY,
1574 _reflect_info=_reflect_info,
1575 **table.dialect_kwargs,
1576 )
1577 if table_key in _reflect_info.unreflectable:
1578 raise _reflect_info.unreflectable[table_key]
1580 if table_key not in _reflect_info.columns:
1581 raise exc.NoSuchTableError(table_name)
1583 # reflect table options, like mysql_engine
1584 if _reflect_info.table_options:
1585 tbl_opts = _reflect_info.table_options.get(table_key)
1586 if tbl_opts:
1587 # add additional kwargs to the Table if the dialect
1588 # returned them
1589 table._validate_dialect_kwargs(tbl_opts)
1591 found_table = False
1592 cols_by_orig_name: Dict[str, sa_schema.Column[Any]] = {}
1594 for col_d in _reflect_info.columns[table_key]:
1595 found_table = True
1597 self._reflect_column(
1598 table,
1599 col_d,
1600 include_columns,
1601 exclude_columns,
1602 cols_by_orig_name,
1603 )
1605 # NOTE: support tables/views with no columns
1606 if not found_table and not self.has_table(table_name, schema):
1607 raise exc.NoSuchTableError(table_name)
1609 self._reflect_pk(
1610 _reflect_info, table_key, table, cols_by_orig_name, exclude_columns
1611 )
1613 self._reflect_fk(
1614 _reflect_info,
1615 table_key,
1616 table,
1617 cols_by_orig_name,
1618 include_columns,
1619 exclude_columns,
1620 resolve_fks,
1621 _extend_on,
1622 reflection_options,
1623 )
1625 self._reflect_indexes(
1626 _reflect_info,
1627 table_key,
1628 table,
1629 cols_by_orig_name,
1630 include_columns,
1631 exclude_columns,
1632 reflection_options,
1633 )
1635 self._reflect_unique_constraints(
1636 _reflect_info,
1637 table_key,
1638 table,
1639 cols_by_orig_name,
1640 include_columns,
1641 exclude_columns,
1642 reflection_options,
1643 )
1645 self._reflect_check_constraints(
1646 _reflect_info,
1647 table_key,
1648 table,
1649 cols_by_orig_name,
1650 include_columns,
1651 exclude_columns,
1652 reflection_options,
1653 )
1655 self._reflect_table_comment(
1656 _reflect_info,
1657 table_key,
1658 table,
1659 reflection_options,
1660 )
1662 def _reflect_column(
1663 self,
1664 table: sa_schema.Table,
1665 col_d: ReflectedColumn,
1666 include_columns: Optional[Collection[str]],
1667 exclude_columns: Collection[str],
1668 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1669 ) -> None:
1670 orig_name = col_d["name"]
1672 table.metadata.dispatch.column_reflect(self, table, col_d)
1673 table.dispatch.column_reflect(self, table, col_d)
1675 # fetch name again as column_reflect is allowed to
1676 # change it
1677 name = col_d["name"]
1678 if (include_columns and name not in include_columns) or (
1679 exclude_columns and name in exclude_columns
1680 ):
1681 return
1683 coltype = col_d["type"]
1685 col_kw = {
1686 k: col_d[k] # type: ignore[literal-required]
1687 for k in [
1688 "nullable",
1689 "autoincrement",
1690 "quote",
1691 "info",
1692 "key",
1693 "comment",
1694 ]
1695 if k in col_d
1696 }
1698 if "dialect_options" in col_d:
1699 col_kw.update(col_d["dialect_options"])
1701 colargs = []
1702 default: Any
1703 if col_d.get("default") is not None:
1704 default_text = col_d["default"]
1705 assert default_text is not None
1706 if isinstance(default_text, TextClause):
1707 default = sa_schema.DefaultClause(
1708 default_text, _reflected=True
1709 )
1710 elif not isinstance(default_text, sa_schema.FetchedValue):
1711 default = sa_schema.DefaultClause(
1712 sql.text(default_text), _reflected=True
1713 )
1714 else:
1715 default = default_text
1716 colargs.append(default)
1718 if "computed" in col_d:
1719 computed = sa_schema.Computed(**col_d["computed"])
1720 colargs.append(computed)
1722 if "identity" in col_d:
1723 identity = sa_schema.Identity(**col_d["identity"])
1724 colargs.append(identity)
1726 cols_by_orig_name[orig_name] = col = sa_schema.Column(
1727 name, coltype, *colargs, **col_kw
1728 )
1730 if col.key in table.primary_key:
1731 col.primary_key = True
1732 table.append_column(col, replace_existing=True)
1734 def _reflect_pk(
1735 self,
1736 _reflect_info: _ReflectionInfo,
1737 table_key: TableKey,
1738 table: sa_schema.Table,
1739 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1740 exclude_columns: Collection[str],
1741 ) -> None:
1742 pk_cons = _reflect_info.pk_constraint.get(table_key)
1743 if pk_cons:
1744 pk_cols = [
1745 cols_by_orig_name[pk]
1746 for pk in pk_cons["constrained_columns"]
1747 if pk in cols_by_orig_name and pk not in exclude_columns
1748 ]
1750 # update pk constraint name, comment and dialect_kwargs
1751 table.primary_key.name = pk_cons.get("name")
1752 table.primary_key.comment = pk_cons.get("comment", None)
1753 dialect_options = pk_cons.get("dialect_options")
1754 if dialect_options:
1755 table.primary_key.dialect_kwargs.update(dialect_options)
1757 # tell the PKConstraint to re-initialize
1758 # its column collection
1759 table.primary_key._reload(pk_cols)
1761 def _reflect_fk(
1762 self,
1763 _reflect_info: _ReflectionInfo,
1764 table_key: TableKey,
1765 table: sa_schema.Table,
1766 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1767 include_columns: Optional[Collection[str]],
1768 exclude_columns: Collection[str],
1769 resolve_fks: bool,
1770 _extend_on: Optional[Set[sa_schema.Table]],
1771 reflection_options: Dict[str, Any],
1772 ) -> None:
1773 fkeys = _reflect_info.foreign_keys.get(table_key, [])
1774 for fkey_d in fkeys:
1775 conname = fkey_d["name"]
1776 # look for columns by orig name in cols_by_orig_name,
1777 # but support columns that are in-Python only as fallback
1778 constrained_columns = [
1779 cols_by_orig_name[c].key if c in cols_by_orig_name else c
1780 for c in fkey_d["constrained_columns"]
1781 ]
1783 if (
1784 exclude_columns
1785 and set(constrained_columns).intersection(exclude_columns)
1786 or (
1787 include_columns
1788 and set(constrained_columns).difference(include_columns)
1789 )
1790 ):
1791 continue
1793 referred_schema = fkey_d["referred_schema"]
1794 referred_table = fkey_d["referred_table"]
1795 referred_columns = fkey_d["referred_columns"]
1796 refspec = []
1797 if referred_schema is not None:
1798 if resolve_fks:
1799 sa_schema.Table(
1800 referred_table,
1801 table.metadata,
1802 schema=referred_schema,
1803 autoload_with=self.bind,
1804 _extend_on=_extend_on,
1805 _reflect_info=_reflect_info,
1806 **reflection_options,
1807 )
1808 for column in referred_columns:
1809 refspec.append(
1810 ".".join([referred_schema, referred_table, column])
1811 )
1812 else:
1813 if resolve_fks:
1814 sa_schema.Table(
1815 referred_table,
1816 table.metadata,
1817 autoload_with=self.bind,
1818 schema=sa_schema.BLANK_SCHEMA,
1819 _extend_on=_extend_on,
1820 _reflect_info=_reflect_info,
1821 **reflection_options,
1822 )
1823 for column in referred_columns:
1824 refspec.append(".".join([referred_table, column]))
1825 if "options" in fkey_d:
1826 options = fkey_d["options"]
1827 else:
1828 options = {}
1830 try:
1831 table.append_constraint(
1832 sa_schema.ForeignKeyConstraint(
1833 constrained_columns,
1834 refspec,
1835 conname,
1836 link_to_name=True,
1837 comment=fkey_d.get("comment"),
1838 **options,
1839 )
1840 )
1841 except exc.ConstraintColumnNotFoundError:
1842 util.warn(
1843 f"On reflected table {table.name}, skipping reflection of "
1844 "foreign key constraint "
1845 f"{conname}; one or more subject columns within "
1846 f"name(s) {', '.join(constrained_columns)} are not "
1847 "present in the table"
1848 )
1850 _index_sort_exprs = {
1851 "asc": operators.asc_op,
1852 "desc": operators.desc_op,
1853 "nulls_first": operators.nulls_first_op,
1854 "nulls_last": operators.nulls_last_op,
1855 }
1857 def _reflect_indexes(
1858 self,
1859 _reflect_info: _ReflectionInfo,
1860 table_key: TableKey,
1861 table: sa_schema.Table,
1862 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1863 include_columns: Optional[Collection[str]],
1864 exclude_columns: Collection[str],
1865 reflection_options: Dict[str, Any],
1866 ) -> None:
1867 # Indexes
1868 indexes = _reflect_info.indexes.get(table_key, [])
1869 for index_d in indexes:
1870 name = index_d["name"]
1871 columns = index_d["column_names"]
1872 expressions = index_d.get("expressions")
1873 column_sorting = index_d.get("column_sorting", {})
1874 unique = index_d["unique"]
1875 flavor = index_d.get("type", "index")
1876 dialect_options = index_d.get("dialect_options", {})
1878 duplicates = index_d.get("duplicates_constraint")
1879 if include_columns and not set(columns).issubset(include_columns):
1880 continue
1881 if duplicates:
1882 continue
1883 # look for columns by orig name in cols_by_orig_name,
1884 # but support columns that are in-Python only as fallback
1885 idx_element: Any
1886 idx_elements = []
1887 for index, c in enumerate(columns):
1888 if c is None:
1889 if not expressions:
1890 util.warn(
1891 f"Skipping {flavor} {name!r} because key "
1892 f"{index + 1} reflected as None but no "
1893 "'expressions' were returned"
1894 )
1895 break
1896 idx_element = sql.text(expressions[index])
1897 else:
1898 try:
1899 if c in cols_by_orig_name:
1900 idx_element = cols_by_orig_name[c]
1901 else:
1902 idx_element = table.c[c]
1903 except KeyError:
1904 util.warn(
1905 f"{flavor} key {c!r} was not located in "
1906 f"columns for table {table.name!r}"
1907 )
1908 continue
1909 for option in column_sorting.get(c, ()):
1910 if option in self._index_sort_exprs:
1911 op = self._index_sort_exprs[option]
1912 idx_element = op(idx_element)
1913 idx_elements.append(idx_element)
1914 else:
1915 sa_schema.Index(
1916 name,
1917 *idx_elements,
1918 _table=table,
1919 unique=unique,
1920 **dialect_options,
1921 )
1923 def _reflect_unique_constraints(
1924 self,
1925 _reflect_info: _ReflectionInfo,
1926 table_key: TableKey,
1927 table: sa_schema.Table,
1928 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1929 include_columns: Optional[Collection[str]],
1930 exclude_columns: Collection[str],
1931 reflection_options: Dict[str, Any],
1932 ) -> None:
1933 constraints = _reflect_info.unique_constraints.get(table_key, [])
1934 # Unique Constraints
1935 for const_d in constraints:
1936 conname = const_d["name"]
1937 columns = const_d["column_names"]
1938 comment = const_d.get("comment")
1939 duplicates = const_d.get("duplicates_index")
1940 dialect_options = const_d.get("dialect_options", {})
1941 if include_columns and not set(columns).issubset(include_columns):
1942 continue
1943 if duplicates:
1944 continue
1945 # look for columns by orig name in cols_by_orig_name,
1946 # but support columns that are in-Python only as fallback
1947 constrained_cols = []
1948 for c in columns:
1949 try:
1950 constrained_col = (
1951 cols_by_orig_name[c]
1952 if c in cols_by_orig_name
1953 else table.c[c]
1954 )
1955 except KeyError:
1956 util.warn(
1957 "unique constraint key '%s' was not located in "
1958 "columns for table '%s'" % (c, table.name)
1959 )
1960 else:
1961 constrained_cols.append(constrained_col)
1962 table.append_constraint(
1963 sa_schema.UniqueConstraint(
1964 *constrained_cols,
1965 name=conname,
1966 comment=comment,
1967 **dialect_options,
1968 )
1969 )
1971 def _reflect_check_constraints(
1972 self,
1973 _reflect_info: _ReflectionInfo,
1974 table_key: TableKey,
1975 table: sa_schema.Table,
1976 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1977 include_columns: Optional[Collection[str]],
1978 exclude_columns: Collection[str],
1979 reflection_options: Dict[str, Any],
1980 ) -> None:
1981 constraints = _reflect_info.check_constraints.get(table_key, [])
1982 for const_d in constraints:
1983 table.append_constraint(sa_schema.CheckConstraint(**const_d))
1985 def _reflect_table_comment(
1986 self,
1987 _reflect_info: _ReflectionInfo,
1988 table_key: TableKey,
1989 table: sa_schema.Table,
1990 reflection_options: Dict[str, Any],
1991 ) -> None:
1992 comment_dict = _reflect_info.table_comment.get(table_key)
1993 if comment_dict:
1994 table.comment = comment_dict["text"]
1996 def _get_reflection_info(
1997 self,
1998 schema: Optional[str] = None,
1999 filter_names: Optional[Collection[str]] = None,
2000 available: Optional[Collection[str]] = None,
2001 _reflect_info: Optional[_ReflectionInfo] = None,
2002 **kw: Any,
2003 ) -> _ReflectionInfo:
2004 kw["schema"] = schema
2006 if filter_names and available and len(filter_names) > 100:
2007 fraction = len(filter_names) / len(available)
2008 else:
2009 fraction = None
2011 unreflectable: Dict[TableKey, exc.UnreflectableTableError]
2012 kw["unreflectable"] = unreflectable = {}
2014 has_result: bool = True
2016 def run(
2017 meth: Any,
2018 *,
2019 optional: bool = False,
2020 check_filter_names_from_meth: bool = False,
2021 ) -> Any:
2022 nonlocal has_result
2023 # simple heuristic to improve reflection performance if a
2024 # dialect implements multi_reflection:
2025 # if more than 50% of the tables in the db are in filter_names
2026 # load all the tables, since it's most likely faster to avoid
2027 # a filter on that many tables.
2028 if (
2029 fraction is None
2030 or fraction <= 0.5
2031 or not self.dialect._overrides_default(meth.__name__)
2032 ):
2033 _fn = filter_names
2034 else:
2035 _fn = None
2036 try:
2037 if has_result:
2038 res = meth(filter_names=_fn, **kw)
2039 if check_filter_names_from_meth and not res:
2040 # method returned no result data.
2041 # skip any future call methods
2042 has_result = False
2043 else:
2044 res = {}
2045 except NotImplementedError:
2046 if not optional:
2047 raise
2048 res = {}
2049 return res
2051 info = _ReflectionInfo(
2052 columns=run(
2053 self.get_multi_columns, check_filter_names_from_meth=True
2054 ),
2055 pk_constraint=run(self.get_multi_pk_constraint),
2056 foreign_keys=run(self.get_multi_foreign_keys),
2057 indexes=run(self.get_multi_indexes),
2058 unique_constraints=run(
2059 self.get_multi_unique_constraints, optional=True
2060 ),
2061 table_comment=run(self.get_multi_table_comment, optional=True),
2062 check_constraints=run(
2063 self.get_multi_check_constraints, optional=True
2064 ),
2065 table_options=run(self.get_multi_table_options, optional=True),
2066 unreflectable=unreflectable,
2067 )
2068 if _reflect_info:
2069 _reflect_info.update(info)
2070 return _reflect_info
2071 else:
2072 return info
2075@final
2076class ReflectionDefaults:
2077 """provides blank default values for reflection methods."""
2079 @classmethod
2080 def columns(cls) -> List[ReflectedColumn]:
2081 return []
2083 @classmethod
2084 def pk_constraint(cls) -> ReflectedPrimaryKeyConstraint:
2085 return {
2086 "name": None,
2087 "constrained_columns": [],
2088 }
2090 @classmethod
2091 def foreign_keys(cls) -> List[ReflectedForeignKeyConstraint]:
2092 return []
2094 @classmethod
2095 def indexes(cls) -> List[ReflectedIndex]:
2096 return []
2098 @classmethod
2099 def unique_constraints(cls) -> List[ReflectedUniqueConstraint]:
2100 return []
2102 @classmethod
2103 def check_constraints(cls) -> List[ReflectedCheckConstraint]:
2104 return []
2106 @classmethod
2107 def table_options(cls) -> Dict[str, Any]:
2108 return {}
2110 @classmethod
2111 def table_comment(cls) -> ReflectedTableComment:
2112 return {"text": None}
2115@dataclass
2116class _ReflectionInfo:
2117 columns: Dict[TableKey, List[ReflectedColumn]]
2118 pk_constraint: Dict[TableKey, Optional[ReflectedPrimaryKeyConstraint]]
2119 foreign_keys: Dict[TableKey, List[ReflectedForeignKeyConstraint]]
2120 indexes: Dict[TableKey, List[ReflectedIndex]]
2121 # optionals
2122 unique_constraints: Dict[TableKey, List[ReflectedUniqueConstraint]]
2123 table_comment: Dict[TableKey, Optional[ReflectedTableComment]]
2124 check_constraints: Dict[TableKey, List[ReflectedCheckConstraint]]
2125 table_options: Dict[TableKey, Dict[str, Any]]
2126 unreflectable: Dict[TableKey, exc.UnreflectableTableError]
2128 def update(self, other: _ReflectionInfo) -> None:
2129 for k, v in self.__dict__.items():
2130 ov = getattr(other, k)
2131 if ov is not None:
2132 if v is None:
2133 setattr(self, k, ov)
2134 else:
2135 v.update(ov)