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 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 .. seealso:: :meth:`Inspector.get_multi_table_comment`
1321 """
1323 with self._operation_context() as conn:
1324 return self.dialect.get_table_comment(
1325 conn, table_name, schema, info_cache=self.info_cache, **kw
1326 )
1328 def get_multi_table_comment(
1329 self,
1330 schema: Optional[str] = None,
1331 filter_names: Optional[Sequence[str]] = None,
1332 kind: ObjectKind = ObjectKind.TABLE,
1333 scope: ObjectScope = ObjectScope.DEFAULT,
1334 **kw: Any,
1335 ) -> Dict[TableKey, ReflectedTableComment]:
1336 r"""Return information about the table comment in all objects
1337 in the given schema.
1339 The objects can be filtered by passing the names to use to
1340 ``filter_names``.
1342 For each table the value is a :class:`.ReflectedTableComment`.
1344 Raises ``NotImplementedError`` for a dialect that does not support
1345 comments.
1347 :param schema: string schema name; if omitted, uses the default schema
1348 of the database connection. For special quoting,
1349 use :class:`.quoted_name`.
1351 :param filter_names: optionally return information only for the
1352 objects listed here.
1354 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1355 to reflect. Defaults to ``ObjectKind.TABLE``.
1357 :param scope: a :class:`.ObjectScope` that specifies if comments of
1358 default, temporary or any tables should be reflected.
1359 Defaults to ``ObjectScope.DEFAULT``.
1361 :param \**kw: Additional keyword argument to pass to the dialect
1362 specific implementation. See the documentation of the dialect
1363 in use for more information.
1365 :return: a dictionary where the keys are two-tuple schema,table-name
1366 and the values are dictionaries, representing the
1367 table comments.
1368 The schema is ``None`` if no schema is provided.
1370 .. versionadded:: 2.0
1372 .. seealso:: :meth:`Inspector.get_table_comment`
1373 """
1375 with self._operation_context() as conn:
1376 return dict(
1377 self.dialect.get_multi_table_comment(
1378 conn,
1379 schema=schema,
1380 filter_names=filter_names,
1381 kind=kind,
1382 scope=scope,
1383 info_cache=self.info_cache,
1384 **kw,
1385 )
1386 )
1388 def get_check_constraints(
1389 self, table_name: str, schema: Optional[str] = None, **kw: Any
1390 ) -> List[ReflectedCheckConstraint]:
1391 r"""Return information about check constraints in ``table_name``.
1393 Given a string ``table_name`` and an optional string `schema`, return
1394 check constraint information as a list of
1395 :class:`.ReflectedCheckConstraint`.
1397 :param table_name: string name of the table. For special quoting,
1398 use :class:`.quoted_name`.
1400 :param schema: string schema name; if omitted, uses the default schema
1401 of the database connection. For special quoting,
1402 use :class:`.quoted_name`.
1404 :param \**kw: Additional keyword argument to pass to the dialect
1405 specific implementation. See the documentation of the dialect
1406 in use for more information.
1408 :return: a list of dictionaries, each representing the
1409 definition of a check constraints.
1411 .. seealso:: :meth:`Inspector.get_multi_check_constraints`
1412 """
1414 with self._operation_context() as conn:
1415 return self.dialect.get_check_constraints(
1416 conn, table_name, schema, info_cache=self.info_cache, **kw
1417 )
1419 def get_multi_check_constraints(
1420 self,
1421 schema: Optional[str] = None,
1422 filter_names: Optional[Sequence[str]] = None,
1423 kind: ObjectKind = ObjectKind.TABLE,
1424 scope: ObjectScope = ObjectScope.DEFAULT,
1425 **kw: Any,
1426 ) -> Dict[TableKey, List[ReflectedCheckConstraint]]:
1427 r"""Return information about check constraints in all tables
1428 in the given schema.
1430 The tables can be filtered by passing the names to use to
1431 ``filter_names``.
1433 For each table the value is a list of
1434 :class:`.ReflectedCheckConstraint`.
1436 :param schema: string schema name; if omitted, uses the default schema
1437 of the database connection. For special quoting,
1438 use :class:`.quoted_name`.
1440 :param filter_names: optionally return information only for the
1441 objects listed here.
1443 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1444 to reflect. Defaults to ``ObjectKind.TABLE``.
1446 :param scope: a :class:`.ObjectScope` that specifies if constraints of
1447 default, temporary or any tables should be reflected.
1448 Defaults to ``ObjectScope.DEFAULT``.
1450 :param \**kw: Additional keyword argument to pass to the dialect
1451 specific implementation. See the documentation of the dialect
1452 in use for more information.
1454 :return: a dictionary where the keys are two-tuple schema,table-name
1455 and the values are list of dictionaries, each representing the
1456 definition of a check constraints.
1457 The schema is ``None`` if no schema is provided.
1459 .. versionadded:: 2.0
1461 .. seealso:: :meth:`Inspector.get_check_constraints`
1462 """
1464 with self._operation_context() as conn:
1465 return dict(
1466 self.dialect.get_multi_check_constraints(
1467 conn,
1468 schema=schema,
1469 filter_names=filter_names,
1470 kind=kind,
1471 scope=scope,
1472 info_cache=self.info_cache,
1473 **kw,
1474 )
1475 )
1477 def reflect_table(
1478 self,
1479 table: sa_schema.Table,
1480 include_columns: Optional[Collection[str]],
1481 exclude_columns: Collection[str] = (),
1482 resolve_fks: bool = True,
1483 _extend_on: Optional[Set[sa_schema.Table]] = None,
1484 _reflect_info: Optional[_ReflectionInfo] = None,
1485 ) -> None:
1486 """Given a :class:`_schema.Table` object, load its internal
1487 constructs based on introspection.
1489 This is the underlying method used by most dialects to produce
1490 table reflection. Direct usage is like::
1492 from sqlalchemy import create_engine, MetaData, Table
1493 from sqlalchemy import inspect
1495 engine = create_engine("...")
1496 meta = MetaData()
1497 user_table = Table("user", meta)
1498 insp = inspect(engine)
1499 insp.reflect_table(user_table, None)
1501 .. versionchanged:: 1.4 Renamed from ``reflecttable`` to
1502 ``reflect_table``
1504 :param table: a :class:`~sqlalchemy.schema.Table` instance.
1505 :param include_columns: a list of string column names to include
1506 in the reflection process. If ``None``, all columns are reflected.
1508 """
1510 if _extend_on is not None:
1511 if table in _extend_on:
1512 return
1513 else:
1514 _extend_on.add(table)
1516 dialect = self.bind.dialect
1518 with self._operation_context() as conn:
1519 schema = conn.schema_for_object(table)
1521 table_name = table.name
1523 # get table-level arguments that are specifically
1524 # intended for reflection, e.g. oracle_resolve_synonyms.
1525 # these are unconditionally passed to related Table
1526 # objects
1527 reflection_options = {
1528 k: table.dialect_kwargs.get(k)
1529 for k in dialect.reflection_options
1530 if k in table.dialect_kwargs
1531 }
1533 table_key = (schema, table_name)
1534 if _reflect_info is None or table_key not in _reflect_info.columns:
1535 _reflect_info = self._get_reflection_info(
1536 schema,
1537 filter_names=[table_name],
1538 kind=ObjectKind.ANY,
1539 scope=ObjectScope.ANY,
1540 _reflect_info=_reflect_info,
1541 **table.dialect_kwargs,
1542 )
1543 if table_key in _reflect_info.unreflectable:
1544 raise _reflect_info.unreflectable[table_key]
1546 if table_key not in _reflect_info.columns:
1547 raise exc.NoSuchTableError(table_name)
1549 # reflect table options, like mysql_engine
1550 if _reflect_info.table_options:
1551 tbl_opts = _reflect_info.table_options.get(table_key)
1552 if tbl_opts:
1553 # add additional kwargs to the Table if the dialect
1554 # returned them
1555 table._validate_dialect_kwargs(tbl_opts)
1557 found_table = False
1558 cols_by_orig_name: Dict[str, sa_schema.Column[Any]] = {}
1560 for col_d in _reflect_info.columns[table_key]:
1561 found_table = True
1563 self._reflect_column(
1564 table,
1565 col_d,
1566 include_columns,
1567 exclude_columns,
1568 cols_by_orig_name,
1569 )
1571 # NOTE: support tables/views with no columns
1572 if not found_table and not self.has_table(table_name, schema):
1573 raise exc.NoSuchTableError(table_name)
1575 self._reflect_pk(
1576 _reflect_info, table_key, table, cols_by_orig_name, exclude_columns
1577 )
1579 self._reflect_fk(
1580 _reflect_info,
1581 table_key,
1582 table,
1583 cols_by_orig_name,
1584 include_columns,
1585 exclude_columns,
1586 resolve_fks,
1587 _extend_on,
1588 reflection_options,
1589 )
1591 self._reflect_indexes(
1592 _reflect_info,
1593 table_key,
1594 table,
1595 cols_by_orig_name,
1596 include_columns,
1597 exclude_columns,
1598 reflection_options,
1599 )
1601 self._reflect_unique_constraints(
1602 _reflect_info,
1603 table_key,
1604 table,
1605 cols_by_orig_name,
1606 include_columns,
1607 exclude_columns,
1608 reflection_options,
1609 )
1611 self._reflect_check_constraints(
1612 _reflect_info,
1613 table_key,
1614 table,
1615 cols_by_orig_name,
1616 include_columns,
1617 exclude_columns,
1618 reflection_options,
1619 )
1621 self._reflect_table_comment(
1622 _reflect_info,
1623 table_key,
1624 table,
1625 reflection_options,
1626 )
1628 def _reflect_column(
1629 self,
1630 table: sa_schema.Table,
1631 col_d: ReflectedColumn,
1632 include_columns: Optional[Collection[str]],
1633 exclude_columns: Collection[str],
1634 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1635 ) -> None:
1636 orig_name = col_d["name"]
1638 table.metadata.dispatch.column_reflect(self, table, col_d)
1639 table.dispatch.column_reflect(self, table, col_d)
1641 # fetch name again as column_reflect is allowed to
1642 # change it
1643 name = col_d["name"]
1644 if (include_columns and name not in include_columns) or (
1645 exclude_columns and name in exclude_columns
1646 ):
1647 return
1649 coltype = col_d["type"]
1651 col_kw = {
1652 k: col_d[k] # type: ignore[literal-required]
1653 for k in [
1654 "nullable",
1655 "autoincrement",
1656 "quote",
1657 "info",
1658 "key",
1659 "comment",
1660 ]
1661 if k in col_d
1662 }
1664 if "dialect_options" in col_d:
1665 col_kw.update(col_d["dialect_options"])
1667 colargs = []
1668 default: Any
1669 if col_d.get("default") is not None:
1670 default_text = col_d["default"]
1671 assert default_text is not None
1672 if isinstance(default_text, TextClause):
1673 default = sa_schema.DefaultClause(
1674 default_text, _reflected=True
1675 )
1676 elif not isinstance(default_text, sa_schema.FetchedValue):
1677 default = sa_schema.DefaultClause(
1678 sql.text(default_text), _reflected=True
1679 )
1680 else:
1681 default = default_text
1682 colargs.append(default)
1684 if "computed" in col_d:
1685 computed = sa_schema.Computed(**col_d["computed"])
1686 colargs.append(computed)
1688 if "identity" in col_d:
1689 identity = sa_schema.Identity(**col_d["identity"])
1690 colargs.append(identity)
1692 cols_by_orig_name[orig_name] = col = sa_schema.Column(
1693 name, coltype, *colargs, **col_kw
1694 )
1696 if col.key in table.primary_key:
1697 col.primary_key = True
1698 table.append_column(col, replace_existing=True)
1700 def _reflect_pk(
1701 self,
1702 _reflect_info: _ReflectionInfo,
1703 table_key: TableKey,
1704 table: sa_schema.Table,
1705 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1706 exclude_columns: Collection[str],
1707 ) -> None:
1708 pk_cons = _reflect_info.pk_constraint.get(table_key)
1709 if pk_cons:
1710 pk_cols = [
1711 cols_by_orig_name[pk]
1712 for pk in pk_cons["constrained_columns"]
1713 if pk in cols_by_orig_name and pk not in exclude_columns
1714 ]
1716 # update pk constraint name, comment and dialect_kwargs
1717 table.primary_key.name = pk_cons.get("name")
1718 table.primary_key.comment = pk_cons.get("comment", None)
1719 dialect_options = pk_cons.get("dialect_options")
1720 if dialect_options:
1721 table.primary_key.dialect_kwargs.update(dialect_options)
1723 # tell the PKConstraint to re-initialize
1724 # its column collection
1725 table.primary_key._reload(pk_cols)
1727 def _reflect_fk(
1728 self,
1729 _reflect_info: _ReflectionInfo,
1730 table_key: TableKey,
1731 table: sa_schema.Table,
1732 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1733 include_columns: Optional[Collection[str]],
1734 exclude_columns: Collection[str],
1735 resolve_fks: bool,
1736 _extend_on: Optional[Set[sa_schema.Table]],
1737 reflection_options: Dict[str, Any],
1738 ) -> None:
1739 fkeys = _reflect_info.foreign_keys.get(table_key, [])
1740 for fkey_d in fkeys:
1741 conname = fkey_d["name"]
1742 # look for columns by orig name in cols_by_orig_name,
1743 # but support columns that are in-Python only as fallback
1744 constrained_columns = [
1745 cols_by_orig_name[c].key if c in cols_by_orig_name else c
1746 for c in fkey_d["constrained_columns"]
1747 ]
1749 if (
1750 exclude_columns
1751 and set(constrained_columns).intersection(exclude_columns)
1752 or (
1753 include_columns
1754 and set(constrained_columns).difference(include_columns)
1755 )
1756 ):
1757 continue
1759 referred_schema = fkey_d["referred_schema"]
1760 referred_table = fkey_d["referred_table"]
1761 referred_columns = fkey_d["referred_columns"]
1762 refspec = []
1763 if referred_schema is not None:
1764 if resolve_fks:
1765 sa_schema.Table(
1766 referred_table,
1767 table.metadata,
1768 schema=referred_schema,
1769 autoload_with=self.bind,
1770 _extend_on=_extend_on,
1771 _reflect_info=_reflect_info,
1772 **reflection_options,
1773 )
1774 for column in referred_columns:
1775 refspec.append(
1776 ".".join([referred_schema, referred_table, column])
1777 )
1778 else:
1779 if resolve_fks:
1780 sa_schema.Table(
1781 referred_table,
1782 table.metadata,
1783 autoload_with=self.bind,
1784 schema=sa_schema.BLANK_SCHEMA,
1785 _extend_on=_extend_on,
1786 _reflect_info=_reflect_info,
1787 **reflection_options,
1788 )
1789 for column in referred_columns:
1790 refspec.append(".".join([referred_table, column]))
1791 if "options" in fkey_d:
1792 options = fkey_d["options"]
1793 else:
1794 options = {}
1796 try:
1797 table.append_constraint(
1798 sa_schema.ForeignKeyConstraint(
1799 constrained_columns,
1800 refspec,
1801 conname,
1802 link_to_name=True,
1803 comment=fkey_d.get("comment"),
1804 **options,
1805 )
1806 )
1807 except exc.ConstraintColumnNotFoundError:
1808 util.warn(
1809 f"On reflected table {table.name}, skipping reflection of "
1810 "foreign key constraint "
1811 f"{conname}; one or more subject columns within "
1812 f"name(s) {', '.join(constrained_columns)} are not "
1813 "present in the table"
1814 )
1816 _index_sort_exprs = {
1817 "asc": operators.asc_op,
1818 "desc": operators.desc_op,
1819 "nulls_first": operators.nulls_first_op,
1820 "nulls_last": operators.nulls_last_op,
1821 }
1823 def _reflect_indexes(
1824 self,
1825 _reflect_info: _ReflectionInfo,
1826 table_key: TableKey,
1827 table: sa_schema.Table,
1828 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1829 include_columns: Optional[Collection[str]],
1830 exclude_columns: Collection[str],
1831 reflection_options: Dict[str, Any],
1832 ) -> None:
1833 # Indexes
1834 indexes = _reflect_info.indexes.get(table_key, [])
1835 for index_d in indexes:
1836 name = index_d["name"]
1837 columns = index_d["column_names"]
1838 expressions = index_d.get("expressions")
1839 column_sorting = index_d.get("column_sorting", {})
1840 unique = index_d["unique"]
1841 flavor = index_d.get("type", "index")
1842 dialect_options = index_d.get("dialect_options", {})
1844 duplicates = index_d.get("duplicates_constraint")
1845 if include_columns and not set(columns).issubset(include_columns):
1846 continue
1847 if duplicates:
1848 continue
1849 # look for columns by orig name in cols_by_orig_name,
1850 # but support columns that are in-Python only as fallback
1851 idx_element: Any
1852 idx_elements = []
1853 for index, c in enumerate(columns):
1854 if c is None:
1855 if not expressions:
1856 util.warn(
1857 f"Skipping {flavor} {name!r} because key "
1858 f"{index + 1} reflected as None but no "
1859 "'expressions' were returned"
1860 )
1861 break
1862 idx_element = sql.text(expressions[index])
1863 else:
1864 try:
1865 if c in cols_by_orig_name:
1866 idx_element = cols_by_orig_name[c]
1867 else:
1868 idx_element = table.c[c]
1869 except KeyError:
1870 util.warn(
1871 f"{flavor} key {c!r} was not located in "
1872 f"columns for table {table.name!r}"
1873 )
1874 continue
1875 for option in column_sorting.get(c, ()):
1876 if option in self._index_sort_exprs:
1877 op = self._index_sort_exprs[option]
1878 idx_element = op(idx_element)
1879 idx_elements.append(idx_element)
1880 else:
1881 sa_schema.Index(
1882 name,
1883 *idx_elements,
1884 _table=table,
1885 unique=unique,
1886 **dialect_options,
1887 )
1889 def _reflect_unique_constraints(
1890 self,
1891 _reflect_info: _ReflectionInfo,
1892 table_key: TableKey,
1893 table: sa_schema.Table,
1894 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1895 include_columns: Optional[Collection[str]],
1896 exclude_columns: Collection[str],
1897 reflection_options: Dict[str, Any],
1898 ) -> None:
1899 constraints = _reflect_info.unique_constraints.get(table_key, [])
1900 # Unique Constraints
1901 for const_d in constraints:
1902 conname = const_d["name"]
1903 columns = const_d["column_names"]
1904 comment = const_d.get("comment")
1905 duplicates = const_d.get("duplicates_index")
1906 dialect_options = const_d.get("dialect_options", {})
1907 if include_columns and not set(columns).issubset(include_columns):
1908 continue
1909 if duplicates:
1910 continue
1911 # look for columns by orig name in cols_by_orig_name,
1912 # but support columns that are in-Python only as fallback
1913 constrained_cols = []
1914 for c in columns:
1915 try:
1916 constrained_col = (
1917 cols_by_orig_name[c]
1918 if c in cols_by_orig_name
1919 else table.c[c]
1920 )
1921 except KeyError:
1922 util.warn(
1923 "unique constraint key '%s' was not located in "
1924 "columns for table '%s'" % (c, table.name)
1925 )
1926 else:
1927 constrained_cols.append(constrained_col)
1928 table.append_constraint(
1929 sa_schema.UniqueConstraint(
1930 *constrained_cols,
1931 name=conname,
1932 comment=comment,
1933 **dialect_options,
1934 )
1935 )
1937 def _reflect_check_constraints(
1938 self,
1939 _reflect_info: _ReflectionInfo,
1940 table_key: TableKey,
1941 table: sa_schema.Table,
1942 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1943 include_columns: Optional[Collection[str]],
1944 exclude_columns: Collection[str],
1945 reflection_options: Dict[str, Any],
1946 ) -> None:
1947 constraints = _reflect_info.check_constraints.get(table_key, [])
1948 for const_d in constraints:
1949 table.append_constraint(sa_schema.CheckConstraint(**const_d))
1951 def _reflect_table_comment(
1952 self,
1953 _reflect_info: _ReflectionInfo,
1954 table_key: TableKey,
1955 table: sa_schema.Table,
1956 reflection_options: Dict[str, Any],
1957 ) -> None:
1958 comment_dict = _reflect_info.table_comment.get(table_key)
1959 if comment_dict:
1960 table.comment = comment_dict["text"]
1962 def _get_reflection_info(
1963 self,
1964 schema: Optional[str] = None,
1965 filter_names: Optional[Collection[str]] = None,
1966 available: Optional[Collection[str]] = None,
1967 _reflect_info: Optional[_ReflectionInfo] = None,
1968 **kw: Any,
1969 ) -> _ReflectionInfo:
1970 kw["schema"] = schema
1972 if filter_names and available and len(filter_names) > 100:
1973 fraction = len(filter_names) / len(available)
1974 else:
1975 fraction = None
1977 unreflectable: Dict[TableKey, exc.UnreflectableTableError]
1978 kw["unreflectable"] = unreflectable = {}
1980 has_result: bool = True
1982 def run(
1983 meth: Any,
1984 *,
1985 optional: bool = False,
1986 check_filter_names_from_meth: bool = False,
1987 ) -> Any:
1988 nonlocal has_result
1989 # simple heuristic to improve reflection performance if a
1990 # dialect implements multi_reflection:
1991 # if more than 50% of the tables in the db are in filter_names
1992 # load all the tables, since it's most likely faster to avoid
1993 # a filter on that many tables.
1994 if (
1995 fraction is None
1996 or fraction <= 0.5
1997 or not self.dialect._overrides_default(meth.__name__)
1998 ):
1999 _fn = filter_names
2000 else:
2001 _fn = None
2002 try:
2003 if has_result:
2004 res = meth(filter_names=_fn, **kw)
2005 if check_filter_names_from_meth and not res:
2006 # method returned no result data.
2007 # skip any future call methods
2008 has_result = False
2009 else:
2010 res = {}
2011 except NotImplementedError:
2012 if not optional:
2013 raise
2014 res = {}
2015 return res
2017 info = _ReflectionInfo(
2018 columns=run(
2019 self.get_multi_columns, check_filter_names_from_meth=True
2020 ),
2021 pk_constraint=run(self.get_multi_pk_constraint),
2022 foreign_keys=run(self.get_multi_foreign_keys),
2023 indexes=run(self.get_multi_indexes),
2024 unique_constraints=run(
2025 self.get_multi_unique_constraints, optional=True
2026 ),
2027 table_comment=run(self.get_multi_table_comment, optional=True),
2028 check_constraints=run(
2029 self.get_multi_check_constraints, optional=True
2030 ),
2031 table_options=run(self.get_multi_table_options, optional=True),
2032 unreflectable=unreflectable,
2033 )
2034 if _reflect_info:
2035 _reflect_info.update(info)
2036 return _reflect_info
2037 else:
2038 return info
2041@final
2042class ReflectionDefaults:
2043 """provides blank default values for reflection methods."""
2045 @classmethod
2046 def columns(cls) -> List[ReflectedColumn]:
2047 return []
2049 @classmethod
2050 def pk_constraint(cls) -> ReflectedPrimaryKeyConstraint:
2051 return {
2052 "name": None,
2053 "constrained_columns": [],
2054 }
2056 @classmethod
2057 def foreign_keys(cls) -> List[ReflectedForeignKeyConstraint]:
2058 return []
2060 @classmethod
2061 def indexes(cls) -> List[ReflectedIndex]:
2062 return []
2064 @classmethod
2065 def unique_constraints(cls) -> List[ReflectedUniqueConstraint]:
2066 return []
2068 @classmethod
2069 def check_constraints(cls) -> List[ReflectedCheckConstraint]:
2070 return []
2072 @classmethod
2073 def table_options(cls) -> Dict[str, Any]:
2074 return {}
2076 @classmethod
2077 def table_comment(cls) -> ReflectedTableComment:
2078 return {"text": None}
2081@dataclass
2082class _ReflectionInfo:
2083 columns: Dict[TableKey, List[ReflectedColumn]]
2084 pk_constraint: Dict[TableKey, Optional[ReflectedPrimaryKeyConstraint]]
2085 foreign_keys: Dict[TableKey, List[ReflectedForeignKeyConstraint]]
2086 indexes: Dict[TableKey, List[ReflectedIndex]]
2087 # optionals
2088 unique_constraints: Dict[TableKey, List[ReflectedUniqueConstraint]]
2089 table_comment: Dict[TableKey, Optional[ReflectedTableComment]]
2090 check_constraints: Dict[TableKey, List[ReflectedCheckConstraint]]
2091 table_options: Dict[TableKey, Dict[str, Any]]
2092 unreflectable: Dict[TableKey, exc.UnreflectableTableError]
2094 def update(self, other: _ReflectionInfo) -> None:
2095 for k, v in self.__dict__.items():
2096 ov = getattr(other, k)
2097 if ov is not None:
2098 if v is None:
2099 setattr(self, k, ov)
2100 else:
2101 v.update(ov)