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-2025 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"""
27from __future__ import annotations
29import contextlib
30from dataclasses import dataclass
31from enum import auto
32from enum import Flag
33from enum import unique
34from typing import Any
35from typing import Callable
36from typing import Collection
37from typing import Dict
38from typing import final
39from typing import Generator
40from typing import Iterable
41from typing import List
42from typing import Optional
43from typing import Sequence
44from typing import Set
45from typing import Tuple
46from typing import TYPE_CHECKING
47from typing import TypeVar
48from typing import Union
50from .base import Connection
51from .base import Engine
52from .. import exc
53from .. import inspection
54from .. import sql
55from .. import util
56from ..sql import operators
57from ..sql import schema as sa_schema
58from ..sql.cache_key import _ad_hoc_cache_key_from_args
59from ..sql.elements import quoted_name
60from ..sql.elements import TextClause
61from ..sql.type_api import TypeEngine
62from ..sql.visitors import InternalTraversal
63from ..util import topological
65if TYPE_CHECKING:
66 from .interfaces import Dialect
67 from .interfaces import ReflectedCheckConstraint
68 from .interfaces import ReflectedColumn
69 from .interfaces import ReflectedForeignKeyConstraint
70 from .interfaces import ReflectedIndex
71 from .interfaces import ReflectedPrimaryKeyConstraint
72 from .interfaces import ReflectedTableComment
73 from .interfaces import ReflectedUniqueConstraint
74 from .interfaces import TableKey
76_R = TypeVar("_R")
79@util.decorator
80def cache(
81 fn: Callable[..., _R],
82 self: Dialect,
83 con: Connection,
84 *args: Any,
85 **kw: Any,
86) -> _R:
87 info_cache = kw.get("info_cache", None)
88 if info_cache is None:
89 return fn(self, con, *args, **kw)
90 exclude = {"info_cache", "unreflectable"}
91 key = (
92 fn.__name__,
93 tuple(
94 (str(a), a.quote) if isinstance(a, quoted_name) else a
95 for a in args
96 if isinstance(a, str)
97 ),
98 tuple(
99 (k, (str(v), v.quote) if isinstance(v, quoted_name) else v)
100 for k, v in kw.items()
101 if k not in exclude
102 ),
103 )
104 ret: _R = info_cache.get(key)
105 if ret is None:
106 ret = fn(self, con, *args, **kw)
107 info_cache[key] = ret
108 return ret
111def flexi_cache(
112 *traverse_args: Tuple[str, InternalTraversal]
113) -> Callable[[Callable[..., _R]], Callable[..., _R]]:
114 @util.decorator
115 def go(
116 fn: Callable[..., _R],
117 self: Dialect,
118 con: Connection,
119 *args: Any,
120 **kw: Any,
121 ) -> _R:
122 info_cache = kw.get("info_cache", None)
123 if info_cache is None:
124 return fn(self, con, *args, **kw)
125 key = _ad_hoc_cache_key_from_args((fn.__name__,), traverse_args, args)
126 ret: _R = info_cache.get(key)
127 if ret is None:
128 ret = fn(self, con, *args, **kw)
129 info_cache[key] = ret
130 return ret
132 return go
135@unique
136class ObjectKind(Flag):
137 """Enumerator that indicates which kind of object to return when calling
138 the ``get_multi`` methods.
140 This is a Flag enum, so custom combinations can be passed. For example,
141 to reflect tables and plain views ``ObjectKind.TABLE | ObjectKind.VIEW``
142 may be used.
144 .. note::
145 Not all dialect may support all kind of object. If a dialect does
146 not support a particular object an empty dict is returned.
147 In case a dialect supports an object, but the requested method
148 is not applicable for the specified kind the default value
149 will be returned for each reflected object. For example reflecting
150 check constraints of view return a dict with all the views with
151 empty lists as values.
152 """
154 TABLE = auto()
155 "Reflect table objects"
156 VIEW = auto()
157 "Reflect plain view objects"
158 MATERIALIZED_VIEW = auto()
159 "Reflect materialized view object"
161 ANY_VIEW = VIEW | MATERIALIZED_VIEW
162 "Reflect any kind of view objects"
163 ANY = TABLE | VIEW | MATERIALIZED_VIEW
164 "Reflect all type of objects"
167@unique
168class ObjectScope(Flag):
169 """Enumerator that indicates which scope to use when calling
170 the ``get_multi`` methods.
171 """
173 DEFAULT = auto()
174 "Include default scope"
175 TEMPORARY = auto()
176 "Include only temp scope"
177 ANY = DEFAULT | TEMPORARY
178 "Include both default and temp scope"
181@inspection._self_inspects
182class Inspector(inspection.Inspectable["Inspector"]):
183 """Performs database schema inspection.
185 The Inspector acts as a proxy to the reflection methods of the
186 :class:`~sqlalchemy.engine.interfaces.Dialect`, providing a
187 consistent interface as well as caching support for previously
188 fetched metadata.
190 A :class:`_reflection.Inspector` object is usually created via the
191 :func:`_sa.inspect` function, which may be passed an
192 :class:`_engine.Engine`
193 or a :class:`_engine.Connection`::
195 from sqlalchemy import inspect, create_engine
197 engine = create_engine("...")
198 insp = inspect(engine)
200 Where above, the :class:`~sqlalchemy.engine.interfaces.Dialect` associated
201 with the engine may opt to return an :class:`_reflection.Inspector`
202 subclass that
203 provides additional methods specific to the dialect's target database.
205 """
207 bind: Union[Engine, Connection]
208 engine: Engine
209 _op_context_requires_connect: bool
210 dialect: Dialect
211 info_cache: Dict[Any, Any]
213 @util.deprecated(
214 "1.4",
215 "The __init__() method on :class:`_reflection.Inspector` "
216 "is deprecated and "
217 "will be removed in a future release. Please use the "
218 ":func:`.sqlalchemy.inspect` "
219 "function on an :class:`_engine.Engine` or "
220 ":class:`_engine.Connection` "
221 "in order to "
222 "acquire an :class:`_reflection.Inspector`.",
223 )
224 def __init__(self, bind: Union[Engine, Connection]):
225 """Initialize a new :class:`_reflection.Inspector`.
227 :param bind: a :class:`~sqlalchemy.engine.Connection`,
228 which is typically an instance of
229 :class:`~sqlalchemy.engine.Engine` or
230 :class:`~sqlalchemy.engine.Connection`.
232 For a dialect-specific instance of :class:`_reflection.Inspector`, see
233 :meth:`_reflection.Inspector.from_engine`
235 """
236 self._init_legacy(bind)
238 @classmethod
239 def _construct(
240 cls, init: Callable[..., Any], bind: Union[Engine, Connection]
241 ) -> Inspector:
242 if hasattr(bind.dialect, "inspector"):
243 cls = bind.dialect.inspector
245 self = cls.__new__(cls)
246 init(self, bind)
247 return self
249 def _init_legacy(self, bind: Union[Engine, Connection]) -> None:
250 if hasattr(bind, "exec_driver_sql"):
251 self._init_connection(bind) # type: ignore[arg-type]
252 else:
253 self._init_engine(bind)
255 def _init_engine(self, engine: Engine) -> None:
256 self.bind = self.engine = engine
257 engine.connect().close()
258 self._op_context_requires_connect = True
259 self.dialect = self.engine.dialect
260 self.info_cache = {}
262 def _init_connection(self, connection: Connection) -> None:
263 self.bind = connection
264 self.engine = connection.engine
265 self._op_context_requires_connect = False
266 self.dialect = self.engine.dialect
267 self.info_cache = {}
269 def clear_cache(self) -> None:
270 """reset the cache for this :class:`.Inspector`.
272 Inspection methods that have data cached will emit SQL queries
273 when next called to get new data.
275 .. versionadded:: 2.0
277 """
278 self.info_cache.clear()
280 @classmethod
281 @util.deprecated(
282 "1.4",
283 "The from_engine() method on :class:`_reflection.Inspector` "
284 "is deprecated and "
285 "will be removed in a future release. Please use the "
286 ":func:`.sqlalchemy.inspect` "
287 "function on an :class:`_engine.Engine` or "
288 ":class:`_engine.Connection` "
289 "in order to "
290 "acquire an :class:`_reflection.Inspector`.",
291 )
292 def from_engine(cls, bind: Engine) -> Inspector:
293 """Construct a new dialect-specific Inspector object from the given
294 engine or connection.
296 :param bind: a :class:`~sqlalchemy.engine.Connection`
297 or :class:`~sqlalchemy.engine.Engine`.
299 This method differs from direct a direct constructor call of
300 :class:`_reflection.Inspector` in that the
301 :class:`~sqlalchemy.engine.interfaces.Dialect` is given a chance to
302 provide a dialect-specific :class:`_reflection.Inspector` instance,
303 which may
304 provide additional methods.
306 See the example at :class:`_reflection.Inspector`.
308 """
309 return cls._construct(cls._init_legacy, bind)
311 @inspection._inspects(Engine)
312 def _engine_insp(bind: Engine) -> Inspector: # type: ignore[misc]
313 return Inspector._construct(Inspector._init_engine, bind)
315 @inspection._inspects(Connection)
316 def _connection_insp(bind: Connection) -> Inspector: # type: ignore[misc]
317 return Inspector._construct(Inspector._init_connection, bind)
319 @contextlib.contextmanager
320 def _operation_context(self) -> Generator[Connection, None, None]:
321 """Return a context that optimizes for multiple operations on a single
322 transaction.
324 This essentially allows connect()/close() to be called if we detected
325 that we're against an :class:`_engine.Engine` and not a
326 :class:`_engine.Connection`.
328 """
329 conn: Connection
330 if self._op_context_requires_connect:
331 conn = self.bind.connect() # type: ignore[union-attr]
332 else:
333 conn = self.bind # type: ignore[assignment]
334 try:
335 yield conn
336 finally:
337 if self._op_context_requires_connect:
338 conn.close()
340 @contextlib.contextmanager
341 def _inspection_context(self) -> Generator[Inspector, None, None]:
342 """Return an :class:`_reflection.Inspector`
343 from this one that will run all
344 operations on a single connection.
346 """
348 with self._operation_context() as conn:
349 sub_insp = self._construct(self.__class__._init_connection, conn)
350 sub_insp.info_cache = self.info_cache
351 yield sub_insp
353 @property
354 def default_schema_name(self) -> Optional[str]:
355 """Return the default schema name presented by the dialect
356 for the current engine's database user.
358 E.g. this is typically ``public`` for PostgreSQL and ``dbo``
359 for SQL Server.
361 """
362 return self.dialect.default_schema_name
364 def get_schema_names(self, **kw: Any) -> List[str]:
365 r"""Return all schema names.
367 :param \**kw: Additional keyword argument to pass to the dialect
368 specific implementation. See the documentation of the dialect
369 in use for more information.
370 """
372 with self._operation_context() as conn:
373 return self.dialect.get_schema_names(
374 conn, info_cache=self.info_cache, **kw
375 )
377 def get_table_names(
378 self, schema: Optional[str] = None, **kw: Any
379 ) -> List[str]:
380 r"""Return all table names within a particular schema.
382 The names are expected to be real tables only, not views.
383 Views are instead returned using the
384 :meth:`_reflection.Inspector.get_view_names` and/or
385 :meth:`_reflection.Inspector.get_materialized_view_names`
386 methods.
388 :param schema: Schema name. If ``schema`` is left at ``None``, the
389 database's default schema is
390 used, else the named schema is searched. If the database does not
391 support named schemas, behavior is undefined if ``schema`` is not
392 passed as ``None``. For special quoting, use :class:`.quoted_name`.
393 :param \**kw: Additional keyword argument to pass to the dialect
394 specific implementation. See the documentation of the dialect
395 in use for more information.
397 .. seealso::
399 :meth:`_reflection.Inspector.get_sorted_table_and_fkc_names`
401 :attr:`_schema.MetaData.sorted_tables`
403 """
405 with self._operation_context() as conn:
406 return self.dialect.get_table_names(
407 conn, schema, info_cache=self.info_cache, **kw
408 )
410 def has_table(
411 self, table_name: str, schema: Optional[str] = None, **kw: Any
412 ) -> bool:
413 r"""Return True if the backend has a table, view, or temporary
414 table of the given name.
416 :param table_name: name of the table to check
417 :param schema: schema name to query, if not the default schema.
418 :param \**kw: Additional keyword argument to pass to the dialect
419 specific implementation. See the documentation of the dialect
420 in use for more information.
422 .. versionadded:: 1.4 - the :meth:`.Inspector.has_table` method
423 replaces the :meth:`_engine.Engine.has_table` method.
425 .. versionchanged:: 2.0:: :meth:`.Inspector.has_table` now formally
426 supports checking for additional table-like objects:
428 * any type of views (plain or materialized)
429 * temporary tables of any kind
431 Previously, these two checks were not formally specified and
432 different dialects would vary in their behavior. The dialect
433 testing suite now includes tests for all of these object types
434 and should be supported by all SQLAlchemy-included dialects.
435 Support among third party dialects may be lagging, however.
437 """
438 with self._operation_context() as conn:
439 return self.dialect.has_table(
440 conn, table_name, schema, info_cache=self.info_cache, **kw
441 )
443 def has_sequence(
444 self, sequence_name: str, schema: Optional[str] = None, **kw: Any
445 ) -> bool:
446 r"""Return True if the backend has a sequence with the given name.
448 :param sequence_name: name of the sequence to check
449 :param schema: schema name to query, if not the default schema.
450 :param \**kw: Additional keyword argument to pass to the dialect
451 specific implementation. See the documentation of the dialect
452 in use for more information.
454 .. versionadded:: 1.4
456 """
457 with self._operation_context() as conn:
458 return self.dialect.has_sequence(
459 conn, sequence_name, schema, info_cache=self.info_cache, **kw
460 )
462 def has_index(
463 self,
464 table_name: str,
465 index_name: str,
466 schema: Optional[str] = None,
467 **kw: Any,
468 ) -> bool:
469 r"""Check the existence of a particular index name in the database.
471 :param table_name: the name of the table the index belongs to
472 :param index_name: the name of the index to check
473 :param schema: schema name to query, if not the default schema.
474 :param \**kw: Additional keyword argument to pass to the dialect
475 specific implementation. See the documentation of the dialect
476 in use for more information.
478 .. versionadded:: 2.0
480 """
481 with self._operation_context() as conn:
482 return self.dialect.has_index(
483 conn,
484 table_name,
485 index_name,
486 schema,
487 info_cache=self.info_cache,
488 **kw,
489 )
491 def has_schema(self, schema_name: str, **kw: Any) -> bool:
492 r"""Return True if the backend has a schema with the given name.
494 :param schema_name: name of the schema to check
495 :param \**kw: Additional keyword argument to pass to the dialect
496 specific implementation. See the documentation of the dialect
497 in use for more information.
499 .. versionadded:: 2.0
501 """
502 with self._operation_context() as conn:
503 return self.dialect.has_schema(
504 conn, schema_name, info_cache=self.info_cache, **kw
505 )
507 def get_sorted_table_and_fkc_names(
508 self,
509 schema: Optional[str] = None,
510 **kw: Any,
511 ) -> List[Tuple[Optional[str], List[Tuple[str, Optional[str]]]]]:
512 r"""Return dependency-sorted table and foreign key constraint names in
513 referred to within a particular schema.
515 This will yield 2-tuples of
516 ``(tablename, [(tname, fkname), (tname, fkname), ...])``
517 consisting of table names in CREATE order grouped with the foreign key
518 constraint names that are not detected as belonging to a cycle.
519 The final element
520 will be ``(None, [(tname, fkname), (tname, fkname), ..])``
521 which will consist of remaining
522 foreign key constraint names that would require a separate CREATE
523 step after-the-fact, based on dependencies between tables.
525 :param schema: schema name to query, if not the default schema.
526 :param \**kw: Additional keyword argument to pass to the dialect
527 specific implementation. See the documentation of the dialect
528 in use for more information.
530 .. seealso::
532 :meth:`_reflection.Inspector.get_table_names`
534 :func:`.sort_tables_and_constraints` - similar method which works
535 with an already-given :class:`_schema.MetaData`.
537 """
539 return [
540 (
541 table_key[1] if table_key else None,
542 [(tname, fks) for (_, tname), fks in fk_collection],
543 )
544 for (
545 table_key,
546 fk_collection,
547 ) in self.sort_tables_on_foreign_key_dependency(
548 consider_schemas=(schema,)
549 )
550 ]
552 def sort_tables_on_foreign_key_dependency(
553 self,
554 consider_schemas: Collection[Optional[str]] = (None,),
555 **kw: Any,
556 ) -> List[
557 Tuple[
558 Optional[Tuple[Optional[str], str]],
559 List[Tuple[Tuple[Optional[str], str], Optional[str]]],
560 ]
561 ]:
562 r"""Return dependency-sorted table and foreign key constraint names
563 referred to within multiple schemas.
565 This method may be compared to
566 :meth:`.Inspector.get_sorted_table_and_fkc_names`, which
567 works on one schema at a time; here, the method is a generalization
568 that will consider multiple schemas at once including that it will
569 resolve for cross-schema foreign keys.
571 .. versionadded:: 2.0
573 """
574 SchemaTab = Tuple[Optional[str], str]
576 tuples: Set[Tuple[SchemaTab, SchemaTab]] = set()
577 remaining_fkcs: Set[Tuple[SchemaTab, Optional[str]]] = set()
578 fknames_for_table: Dict[SchemaTab, Set[Optional[str]]] = {}
579 tnames: List[SchemaTab] = []
581 for schname in consider_schemas:
582 schema_fkeys = self.get_multi_foreign_keys(schname, **kw)
583 tnames.extend(schema_fkeys)
584 for (_, tname), fkeys in schema_fkeys.items():
585 fknames_for_table[(schname, tname)] = {
586 fk["name"] for fk in fkeys
587 }
588 for fkey in fkeys:
589 if (
590 tname != fkey["referred_table"]
591 or schname != fkey["referred_schema"]
592 ):
593 tuples.add(
594 (
595 (
596 fkey["referred_schema"],
597 fkey["referred_table"],
598 ),
599 (schname, tname),
600 )
601 )
602 try:
603 candidate_sort = list(topological.sort(tuples, tnames))
604 except exc.CircularDependencyError as err:
605 edge: Tuple[SchemaTab, SchemaTab]
606 for edge in err.edges:
607 tuples.remove(edge)
608 remaining_fkcs.update(
609 (edge[1], fkc) for fkc in fknames_for_table[edge[1]]
610 )
612 candidate_sort = list(topological.sort(tuples, tnames))
613 ret: List[
614 Tuple[Optional[SchemaTab], List[Tuple[SchemaTab, Optional[str]]]]
615 ]
616 ret = [
617 (
618 (schname, tname),
619 [
620 ((schname, tname), fk)
621 for fk in fknames_for_table[(schname, tname)].difference(
622 name for _, name in remaining_fkcs
623 )
624 ],
625 )
626 for (schname, tname) in candidate_sort
627 ]
628 return ret + [(None, list(remaining_fkcs))]
630 def get_temp_table_names(self, **kw: Any) -> List[str]:
631 r"""Return a list of temporary table names for the current bind.
633 This method is unsupported by most dialects; currently
634 only Oracle Database, PostgreSQL and SQLite implements it.
636 :param \**kw: Additional keyword argument to pass to the dialect
637 specific implementation. See the documentation of the dialect
638 in use for more information.
640 """
642 with self._operation_context() as conn:
643 return self.dialect.get_temp_table_names(
644 conn, info_cache=self.info_cache, **kw
645 )
647 def get_temp_view_names(self, **kw: Any) -> List[str]:
648 r"""Return a list of temporary view names for the current bind.
650 This method is unsupported by most dialects; currently
651 only PostgreSQL and SQLite implements it.
653 :param \**kw: Additional keyword argument to pass to the dialect
654 specific implementation. See the documentation of the dialect
655 in use for more information.
657 """
658 with self._operation_context() as conn:
659 return self.dialect.get_temp_view_names(
660 conn, info_cache=self.info_cache, **kw
661 )
663 def get_table_options(
664 self, table_name: str, schema: Optional[str] = None, **kw: Any
665 ) -> Dict[str, Any]:
666 r"""Return a dictionary of options specified when the table of the
667 given name was created.
669 This currently includes some options that apply to MySQL and Oracle
670 Database tables.
672 :param table_name: string name of the table. For special quoting,
673 use :class:`.quoted_name`.
675 :param schema: string schema name; if omitted, uses the default schema
676 of the database connection. For special quoting,
677 use :class:`.quoted_name`.
679 :param \**kw: Additional keyword argument to pass to the dialect
680 specific implementation. See the documentation of the dialect
681 in use for more information.
683 :return: a dict with the table options. The returned keys depend on the
684 dialect in use. Each one is prefixed with the dialect name.
686 .. seealso:: :meth:`Inspector.get_multi_table_options`
688 """
689 with self._operation_context() as conn:
690 return self.dialect.get_table_options(
691 conn, table_name, schema, info_cache=self.info_cache, **kw
692 )
694 def get_multi_table_options(
695 self,
696 schema: Optional[str] = None,
697 filter_names: Optional[Sequence[str]] = None,
698 kind: ObjectKind = ObjectKind.TABLE,
699 scope: ObjectScope = ObjectScope.DEFAULT,
700 **kw: Any,
701 ) -> Dict[TableKey, Dict[str, Any]]:
702 r"""Return a dictionary of options specified when the tables in the
703 given schema were created.
705 The tables can be filtered by passing the names to use to
706 ``filter_names``.
708 This currently includes some options that apply to MySQL and Oracle
709 tables.
711 :param schema: string schema name; if omitted, uses the default schema
712 of the database connection. For special quoting,
713 use :class:`.quoted_name`.
715 :param filter_names: optionally return information only for the
716 objects listed here.
718 :param kind: a :class:`.ObjectKind` that specifies the type of objects
719 to reflect. Defaults to ``ObjectKind.TABLE``.
721 :param scope: a :class:`.ObjectScope` that specifies if options of
722 default, temporary or any tables should be reflected.
723 Defaults to ``ObjectScope.DEFAULT``.
725 :param \**kw: Additional keyword argument to pass to the dialect
726 specific implementation. See the documentation of the dialect
727 in use for more information.
729 :return: a dictionary where the keys are two-tuple schema,table-name
730 and the values are dictionaries with the table options.
731 The returned keys in each dict depend on the
732 dialect in use. Each one is prefixed with the dialect name.
733 The schema is ``None`` if no schema is provided.
735 .. versionadded:: 2.0
737 .. seealso:: :meth:`Inspector.get_table_options`
738 """
739 with self._operation_context() as conn:
740 res = self.dialect.get_multi_table_options(
741 conn,
742 schema=schema,
743 filter_names=filter_names,
744 kind=kind,
745 scope=scope,
746 info_cache=self.info_cache,
747 **kw,
748 )
749 return dict(res)
751 def get_view_names(
752 self, schema: Optional[str] = None, **kw: Any
753 ) -> List[str]:
754 r"""Return all non-materialized view names in `schema`.
756 :param schema: Optional, retrieve names from a non-default schema.
757 For special quoting, use :class:`.quoted_name`.
758 :param \**kw: Additional keyword argument to pass to the dialect
759 specific implementation. See the documentation of the dialect
760 in use for more information.
763 .. versionchanged:: 2.0 For those dialects that previously included
764 the names of materialized views in this list (currently PostgreSQL),
765 this method no longer returns the names of materialized views.
766 the :meth:`.Inspector.get_materialized_view_names` method should
767 be used instead.
769 .. seealso::
771 :meth:`.Inspector.get_materialized_view_names`
773 """
775 with self._operation_context() as conn:
776 return self.dialect.get_view_names(
777 conn, schema, info_cache=self.info_cache, **kw
778 )
780 def get_materialized_view_names(
781 self, schema: Optional[str] = None, **kw: Any
782 ) -> List[str]:
783 r"""Return all materialized view names in `schema`.
785 :param schema: Optional, retrieve names from a non-default schema.
786 For special quoting, use :class:`.quoted_name`.
787 :param \**kw: Additional keyword argument to pass to the dialect
788 specific implementation. See the documentation of the dialect
789 in use for more information.
791 .. versionadded:: 2.0
793 .. seealso::
795 :meth:`.Inspector.get_view_names`
797 """
799 with self._operation_context() as conn:
800 return self.dialect.get_materialized_view_names(
801 conn, schema, info_cache=self.info_cache, **kw
802 )
804 def get_sequence_names(
805 self, schema: Optional[str] = None, **kw: Any
806 ) -> List[str]:
807 r"""Return all sequence names in `schema`.
809 :param schema: Optional, retrieve names from a non-default schema.
810 For special quoting, use :class:`.quoted_name`.
811 :param \**kw: Additional keyword argument to pass to the dialect
812 specific implementation. See the documentation of the dialect
813 in use for more information.
815 """
817 with self._operation_context() as conn:
818 return self.dialect.get_sequence_names(
819 conn, schema, info_cache=self.info_cache, **kw
820 )
822 def get_view_definition(
823 self, view_name: str, schema: Optional[str] = None, **kw: Any
824 ) -> str:
825 r"""Return definition for the plain or materialized view called
826 ``view_name``.
828 :param view_name: Name of the view.
829 :param schema: Optional, retrieve names from a non-default schema.
830 For special quoting, use :class:`.quoted_name`.
831 :param \**kw: Additional keyword argument to pass to the dialect
832 specific implementation. See the documentation of the dialect
833 in use for more information.
835 """
837 with self._operation_context() as conn:
838 return self.dialect.get_view_definition(
839 conn, view_name, schema, info_cache=self.info_cache, **kw
840 )
842 def get_columns(
843 self, table_name: str, schema: Optional[str] = None, **kw: Any
844 ) -> List[ReflectedColumn]:
845 r"""Return information about columns in ``table_name``.
847 Given a string ``table_name`` and an optional string ``schema``,
848 return column information as a list of :class:`.ReflectedColumn`.
850 :param table_name: string name of the table. For special quoting,
851 use :class:`.quoted_name`.
853 :param schema: string schema name; if omitted, uses the default schema
854 of the database connection. For special quoting,
855 use :class:`.quoted_name`.
857 :param \**kw: Additional keyword argument to pass to the dialect
858 specific implementation. See the documentation of the dialect
859 in use for more information.
861 :return: list of dictionaries, each representing the definition of
862 a database column.
864 .. seealso:: :meth:`Inspector.get_multi_columns`.
866 """
868 with self._operation_context() as conn:
869 col_defs = self.dialect.get_columns(
870 conn, table_name, schema, info_cache=self.info_cache, **kw
871 )
872 if col_defs:
873 self._instantiate_types([col_defs])
874 return col_defs
876 def _instantiate_types(
877 self, data: Iterable[List[ReflectedColumn]]
878 ) -> None:
879 # make this easy and only return instances for coltype
880 for col_defs in data:
881 for col_def in col_defs:
882 coltype = col_def["type"]
883 if not isinstance(coltype, TypeEngine):
884 col_def["type"] = coltype()
886 def get_multi_columns(
887 self,
888 schema: Optional[str] = None,
889 filter_names: Optional[Sequence[str]] = None,
890 kind: ObjectKind = ObjectKind.TABLE,
891 scope: ObjectScope = ObjectScope.DEFAULT,
892 **kw: Any,
893 ) -> Dict[TableKey, List[ReflectedColumn]]:
894 r"""Return information about columns in all objects in the given
895 schema.
897 The objects can be filtered by passing the names to use to
898 ``filter_names``.
900 For each table the value is a list of :class:`.ReflectedColumn`.
902 :param schema: string schema name; if omitted, uses the default schema
903 of the database connection. For special quoting,
904 use :class:`.quoted_name`.
906 :param filter_names: optionally return information only for the
907 objects listed here.
909 :param kind: a :class:`.ObjectKind` that specifies the type of objects
910 to reflect. Defaults to ``ObjectKind.TABLE``.
912 :param scope: a :class:`.ObjectScope` that specifies if columns of
913 default, temporary or any tables should be reflected.
914 Defaults to ``ObjectScope.DEFAULT``.
916 :param \**kw: Additional keyword argument to pass to the dialect
917 specific implementation. See the documentation of the dialect
918 in use for more information.
920 :return: a dictionary where the keys are two-tuple schema,table-name
921 and the values are list of dictionaries, each representing the
922 definition of a database column.
923 The schema is ``None`` if no schema is provided.
925 .. versionadded:: 2.0
927 .. seealso:: :meth:`Inspector.get_columns`
928 """
930 with self._operation_context() as conn:
931 table_col_defs = dict(
932 self.dialect.get_multi_columns(
933 conn,
934 schema=schema,
935 filter_names=filter_names,
936 kind=kind,
937 scope=scope,
938 info_cache=self.info_cache,
939 **kw,
940 )
941 )
942 self._instantiate_types(table_col_defs.values())
943 return table_col_defs
945 def get_pk_constraint(
946 self, table_name: str, schema: Optional[str] = None, **kw: Any
947 ) -> ReflectedPrimaryKeyConstraint:
948 r"""Return information about primary key constraint in ``table_name``.
950 Given a string ``table_name``, and an optional string `schema`, return
951 primary key information as a :class:`.ReflectedPrimaryKeyConstraint`.
953 :param table_name: string name of the table. For special quoting,
954 use :class:`.quoted_name`.
956 :param schema: string schema name; if omitted, uses the default schema
957 of the database connection. For special quoting,
958 use :class:`.quoted_name`.
960 :param \**kw: Additional keyword argument to pass to the dialect
961 specific implementation. See the documentation of the dialect
962 in use for more information.
964 :return: a dictionary representing the definition of
965 a primary key constraint.
967 .. seealso:: :meth:`Inspector.get_multi_pk_constraint`
968 """
969 with self._operation_context() as conn:
970 return self.dialect.get_pk_constraint(
971 conn, table_name, schema, info_cache=self.info_cache, **kw
972 )
974 def get_multi_pk_constraint(
975 self,
976 schema: Optional[str] = None,
977 filter_names: Optional[Sequence[str]] = None,
978 kind: ObjectKind = ObjectKind.TABLE,
979 scope: ObjectScope = ObjectScope.DEFAULT,
980 **kw: Any,
981 ) -> Dict[TableKey, ReflectedPrimaryKeyConstraint]:
982 r"""Return information about primary key constraints in
983 all tables in the given schema.
985 The tables can be filtered by passing the names to use to
986 ``filter_names``.
988 For each table the value is a :class:`.ReflectedPrimaryKeyConstraint`.
990 :param schema: string schema name; if omitted, uses the default schema
991 of the database connection. For special quoting,
992 use :class:`.quoted_name`.
994 :param filter_names: optionally return information only for the
995 objects listed here.
997 :param kind: a :class:`.ObjectKind` that specifies the type of objects
998 to reflect. Defaults to ``ObjectKind.TABLE``.
1000 :param scope: a :class:`.ObjectScope` that specifies if primary keys of
1001 default, temporary or any tables should be reflected.
1002 Defaults to ``ObjectScope.DEFAULT``.
1004 :param \**kw: Additional keyword argument to pass to the dialect
1005 specific implementation. See the documentation of the dialect
1006 in use for more information.
1008 :return: a dictionary where the keys are two-tuple schema,table-name
1009 and the values are dictionaries, each representing the
1010 definition of a primary key constraint.
1011 The schema is ``None`` if no schema is provided.
1013 .. versionadded:: 2.0
1015 .. seealso:: :meth:`Inspector.get_pk_constraint`
1016 """
1017 with self._operation_context() as conn:
1018 return dict(
1019 self.dialect.get_multi_pk_constraint(
1020 conn,
1021 schema=schema,
1022 filter_names=filter_names,
1023 kind=kind,
1024 scope=scope,
1025 info_cache=self.info_cache,
1026 **kw,
1027 )
1028 )
1030 def get_foreign_keys(
1031 self, table_name: str, schema: Optional[str] = None, **kw: Any
1032 ) -> List[ReflectedForeignKeyConstraint]:
1033 r"""Return information about foreign_keys in ``table_name``.
1035 Given a string ``table_name``, and an optional string `schema`, return
1036 foreign key information as a list of
1037 :class:`.ReflectedForeignKeyConstraint`.
1039 :param table_name: string name of the table. For special quoting,
1040 use :class:`.quoted_name`.
1042 :param schema: string schema name; if omitted, uses the default schema
1043 of the database connection. For special quoting,
1044 use :class:`.quoted_name`.
1046 :param \**kw: Additional keyword argument to pass to the dialect
1047 specific implementation. See the documentation of the dialect
1048 in use for more information.
1050 :return: a list of dictionaries, each representing the
1051 a foreign key definition.
1053 .. seealso:: :meth:`Inspector.get_multi_foreign_keys`
1054 """
1056 with self._operation_context() as conn:
1057 return self.dialect.get_foreign_keys(
1058 conn, table_name, schema, info_cache=self.info_cache, **kw
1059 )
1061 def get_multi_foreign_keys(
1062 self,
1063 schema: Optional[str] = None,
1064 filter_names: Optional[Sequence[str]] = None,
1065 kind: ObjectKind = ObjectKind.TABLE,
1066 scope: ObjectScope = ObjectScope.DEFAULT,
1067 **kw: Any,
1068 ) -> Dict[TableKey, List[ReflectedForeignKeyConstraint]]:
1069 r"""Return information about foreign_keys in all tables
1070 in the given schema.
1072 The tables can be filtered by passing the names to use to
1073 ``filter_names``.
1075 For each table the value is a list of
1076 :class:`.ReflectedForeignKeyConstraint`.
1078 :param schema: string schema name; if omitted, uses the default schema
1079 of the database connection. For special quoting,
1080 use :class:`.quoted_name`.
1082 :param filter_names: optionally return information only for the
1083 objects listed here.
1085 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1086 to reflect. Defaults to ``ObjectKind.TABLE``.
1088 :param scope: a :class:`.ObjectScope` that specifies if foreign keys of
1089 default, temporary or any tables should be reflected.
1090 Defaults to ``ObjectScope.DEFAULT``.
1092 :param \**kw: Additional keyword argument to pass to the dialect
1093 specific implementation. See the documentation of the dialect
1094 in use for more information.
1096 :return: a dictionary where the keys are two-tuple schema,table-name
1097 and the values are list of dictionaries, each representing
1098 a foreign key definition.
1099 The schema is ``None`` if no schema is provided.
1101 .. versionadded:: 2.0
1103 .. seealso:: :meth:`Inspector.get_foreign_keys`
1104 """
1106 with self._operation_context() as conn:
1107 return dict(
1108 self.dialect.get_multi_foreign_keys(
1109 conn,
1110 schema=schema,
1111 filter_names=filter_names,
1112 kind=kind,
1113 scope=scope,
1114 info_cache=self.info_cache,
1115 **kw,
1116 )
1117 )
1119 def get_indexes(
1120 self, table_name: str, schema: Optional[str] = None, **kw: Any
1121 ) -> List[ReflectedIndex]:
1122 r"""Return information about indexes in ``table_name``.
1124 Given a string ``table_name`` and an optional string `schema`, return
1125 index information as a list of :class:`.ReflectedIndex`.
1127 :param table_name: string name of the table. For special quoting,
1128 use :class:`.quoted_name`.
1130 :param schema: string schema name; if omitted, uses the default schema
1131 of the database connection. For special quoting,
1132 use :class:`.quoted_name`.
1134 :param \**kw: Additional keyword argument to pass to the dialect
1135 specific implementation. See the documentation of the dialect
1136 in use for more information.
1138 :return: a list of dictionaries, each representing the
1139 definition of an index.
1141 .. seealso:: :meth:`Inspector.get_multi_indexes`
1142 """
1144 with self._operation_context() as conn:
1145 return self.dialect.get_indexes(
1146 conn, table_name, schema, info_cache=self.info_cache, **kw
1147 )
1149 def get_multi_indexes(
1150 self,
1151 schema: Optional[str] = None,
1152 filter_names: Optional[Sequence[str]] = None,
1153 kind: ObjectKind = ObjectKind.TABLE,
1154 scope: ObjectScope = ObjectScope.DEFAULT,
1155 **kw: Any,
1156 ) -> Dict[TableKey, List[ReflectedIndex]]:
1157 r"""Return information about indexes in in all objects
1158 in the given schema.
1160 The objects can be filtered by passing the names to use to
1161 ``filter_names``.
1163 For each table the value is a list of :class:`.ReflectedIndex`.
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 filter_names: optionally return information only for the
1170 objects listed here.
1172 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1173 to reflect. Defaults to ``ObjectKind.TABLE``.
1175 :param scope: a :class:`.ObjectScope` that specifies if indexes of
1176 default, temporary or any tables should be reflected.
1177 Defaults to ``ObjectScope.DEFAULT``.
1179 :param \**kw: Additional keyword argument to pass to the dialect
1180 specific implementation. See the documentation of the dialect
1181 in use for more information.
1183 :return: a dictionary where the keys are two-tuple schema,table-name
1184 and the values are list of dictionaries, each representing the
1185 definition of an index.
1186 The schema is ``None`` if no schema is provided.
1188 .. versionadded:: 2.0
1190 .. seealso:: :meth:`Inspector.get_indexes`
1191 """
1193 with self._operation_context() as conn:
1194 return dict(
1195 self.dialect.get_multi_indexes(
1196 conn,
1197 schema=schema,
1198 filter_names=filter_names,
1199 kind=kind,
1200 scope=scope,
1201 info_cache=self.info_cache,
1202 **kw,
1203 )
1204 )
1206 def get_unique_constraints(
1207 self, table_name: str, schema: Optional[str] = None, **kw: Any
1208 ) -> List[ReflectedUniqueConstraint]:
1209 r"""Return information about unique constraints in ``table_name``.
1211 Given a string ``table_name`` and an optional string `schema`, return
1212 unique constraint information as a list of
1213 :class:`.ReflectedUniqueConstraint`.
1215 :param table_name: string name of the table. For special quoting,
1216 use :class:`.quoted_name`.
1218 :param schema: string schema name; if omitted, uses the default schema
1219 of the database connection. For special quoting,
1220 use :class:`.quoted_name`.
1222 :param \**kw: Additional keyword argument to pass to the dialect
1223 specific implementation. See the documentation of the dialect
1224 in use for more information.
1226 :return: a list of dictionaries, each representing the
1227 definition of an unique constraint.
1229 .. seealso:: :meth:`Inspector.get_multi_unique_constraints`
1230 """
1232 with self._operation_context() as conn:
1233 return self.dialect.get_unique_constraints(
1234 conn, table_name, schema, info_cache=self.info_cache, **kw
1235 )
1237 def get_multi_unique_constraints(
1238 self,
1239 schema: Optional[str] = None,
1240 filter_names: Optional[Sequence[str]] = None,
1241 kind: ObjectKind = ObjectKind.TABLE,
1242 scope: ObjectScope = ObjectScope.DEFAULT,
1243 **kw: Any,
1244 ) -> Dict[TableKey, List[ReflectedUniqueConstraint]]:
1245 r"""Return information about unique constraints in all tables
1246 in the given schema.
1248 The tables can be filtered by passing the names to use to
1249 ``filter_names``.
1251 For each table the value is a list of
1252 :class:`.ReflectedUniqueConstraint`.
1254 :param schema: string schema name; if omitted, uses the default schema
1255 of the database connection. For special quoting,
1256 use :class:`.quoted_name`.
1258 :param filter_names: optionally return information only for the
1259 objects listed here.
1261 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1262 to reflect. Defaults to ``ObjectKind.TABLE``.
1264 :param scope: a :class:`.ObjectScope` that specifies if constraints of
1265 default, temporary or any tables should be reflected.
1266 Defaults to ``ObjectScope.DEFAULT``.
1268 :param \**kw: Additional keyword argument to pass to the dialect
1269 specific implementation. See the documentation of the dialect
1270 in use for more information.
1272 :return: a dictionary where the keys are two-tuple schema,table-name
1273 and the values are list of dictionaries, each representing the
1274 definition of an unique constraint.
1275 The schema is ``None`` if no schema is provided.
1277 .. versionadded:: 2.0
1279 .. seealso:: :meth:`Inspector.get_unique_constraints`
1280 """
1282 with self._operation_context() as conn:
1283 return dict(
1284 self.dialect.get_multi_unique_constraints(
1285 conn,
1286 schema=schema,
1287 filter_names=filter_names,
1288 kind=kind,
1289 scope=scope,
1290 info_cache=self.info_cache,
1291 **kw,
1292 )
1293 )
1295 def get_table_comment(
1296 self, table_name: str, schema: Optional[str] = None, **kw: Any
1297 ) -> ReflectedTableComment:
1298 r"""Return information about the table comment for ``table_name``.
1300 Given a string ``table_name`` and an optional string ``schema``,
1301 return table comment information as a :class:`.ReflectedTableComment`.
1303 Raises ``NotImplementedError`` for a dialect that does not support
1304 comments.
1306 :param table_name: string name of the table. For special quoting,
1307 use :class:`.quoted_name`.
1309 :param schema: string schema name; if omitted, uses the default schema
1310 of the database connection. For special quoting,
1311 use :class:`.quoted_name`.
1313 :param \**kw: Additional keyword argument to pass to the dialect
1314 specific implementation. See the documentation of the dialect
1315 in use for more information.
1317 :return: a dictionary, with the table comment.
1319 .. seealso:: :meth:`Inspector.get_multi_table_comment`
1320 """
1322 with self._operation_context() as conn:
1323 return self.dialect.get_table_comment(
1324 conn, table_name, schema, info_cache=self.info_cache, **kw
1325 )
1327 def get_multi_table_comment(
1328 self,
1329 schema: Optional[str] = None,
1330 filter_names: Optional[Sequence[str]] = None,
1331 kind: ObjectKind = ObjectKind.TABLE,
1332 scope: ObjectScope = ObjectScope.DEFAULT,
1333 **kw: Any,
1334 ) -> Dict[TableKey, ReflectedTableComment]:
1335 r"""Return information about the table comment in all objects
1336 in the given schema.
1338 The objects can be filtered by passing the names to use to
1339 ``filter_names``.
1341 For each table the value is a :class:`.ReflectedTableComment`.
1343 Raises ``NotImplementedError`` for a dialect that does not support
1344 comments.
1346 :param schema: string schema name; if omitted, uses the default schema
1347 of the database connection. For special quoting,
1348 use :class:`.quoted_name`.
1350 :param filter_names: optionally return information only for the
1351 objects listed here.
1353 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1354 to reflect. Defaults to ``ObjectKind.TABLE``.
1356 :param scope: a :class:`.ObjectScope` that specifies if comments of
1357 default, temporary or any tables should be reflected.
1358 Defaults to ``ObjectScope.DEFAULT``.
1360 :param \**kw: Additional keyword argument to pass to the dialect
1361 specific implementation. See the documentation of the dialect
1362 in use for more information.
1364 :return: a dictionary where the keys are two-tuple schema,table-name
1365 and the values are dictionaries, representing the
1366 table comments.
1367 The schema is ``None`` if no schema is provided.
1369 .. versionadded:: 2.0
1371 .. seealso:: :meth:`Inspector.get_table_comment`
1372 """
1374 with self._operation_context() as conn:
1375 return dict(
1376 self.dialect.get_multi_table_comment(
1377 conn,
1378 schema=schema,
1379 filter_names=filter_names,
1380 kind=kind,
1381 scope=scope,
1382 info_cache=self.info_cache,
1383 **kw,
1384 )
1385 )
1387 def get_check_constraints(
1388 self, table_name: str, schema: Optional[str] = None, **kw: Any
1389 ) -> List[ReflectedCheckConstraint]:
1390 r"""Return information about check constraints in ``table_name``.
1392 Given a string ``table_name`` and an optional string `schema`, return
1393 check constraint information as a list of
1394 :class:`.ReflectedCheckConstraint`.
1396 :param table_name: string name of the table. For special quoting,
1397 use :class:`.quoted_name`.
1399 :param schema: string schema name; if omitted, uses the default schema
1400 of the database connection. For special quoting,
1401 use :class:`.quoted_name`.
1403 :param \**kw: Additional keyword argument to pass to the dialect
1404 specific implementation. See the documentation of the dialect
1405 in use for more information.
1407 :return: a list of dictionaries, each representing the
1408 definition of a check constraints.
1410 .. seealso:: :meth:`Inspector.get_multi_check_constraints`
1411 """
1413 with self._operation_context() as conn:
1414 return self.dialect.get_check_constraints(
1415 conn, table_name, schema, info_cache=self.info_cache, **kw
1416 )
1418 def get_multi_check_constraints(
1419 self,
1420 schema: Optional[str] = None,
1421 filter_names: Optional[Sequence[str]] = None,
1422 kind: ObjectKind = ObjectKind.TABLE,
1423 scope: ObjectScope = ObjectScope.DEFAULT,
1424 **kw: Any,
1425 ) -> Dict[TableKey, List[ReflectedCheckConstraint]]:
1426 r"""Return information about check constraints in all tables
1427 in the given schema.
1429 The tables can be filtered by passing the names to use to
1430 ``filter_names``.
1432 For each table the value is a list of
1433 :class:`.ReflectedCheckConstraint`.
1435 :param schema: string schema name; if omitted, uses the default schema
1436 of the database connection. For special quoting,
1437 use :class:`.quoted_name`.
1439 :param filter_names: optionally return information only for the
1440 objects listed here.
1442 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1443 to reflect. Defaults to ``ObjectKind.TABLE``.
1445 :param scope: a :class:`.ObjectScope` that specifies if constraints of
1446 default, temporary or any tables should be reflected.
1447 Defaults to ``ObjectScope.DEFAULT``.
1449 :param \**kw: Additional keyword argument to pass to the dialect
1450 specific implementation. See the documentation of the dialect
1451 in use for more information.
1453 :return: a dictionary where the keys are two-tuple schema,table-name
1454 and the values are list of dictionaries, each representing the
1455 definition of a check constraints.
1456 The schema is ``None`` if no schema is provided.
1458 .. versionadded:: 2.0
1460 .. seealso:: :meth:`Inspector.get_check_constraints`
1461 """
1463 with self._operation_context() as conn:
1464 return dict(
1465 self.dialect.get_multi_check_constraints(
1466 conn,
1467 schema=schema,
1468 filter_names=filter_names,
1469 kind=kind,
1470 scope=scope,
1471 info_cache=self.info_cache,
1472 **kw,
1473 )
1474 )
1476 def reflect_table(
1477 self,
1478 table: sa_schema.Table,
1479 include_columns: Optional[Collection[str]],
1480 exclude_columns: Collection[str] = (),
1481 resolve_fks: bool = True,
1482 _extend_on: Optional[Set[sa_schema.Table]] = None,
1483 _reflect_info: Optional[_ReflectionInfo] = None,
1484 ) -> None:
1485 """Given a :class:`_schema.Table` object, load its internal
1486 constructs based on introspection.
1488 This is the underlying method used by most dialects to produce
1489 table reflection. Direct usage is like::
1491 from sqlalchemy import create_engine, MetaData, Table
1492 from sqlalchemy import inspect
1494 engine = create_engine("...")
1495 meta = MetaData()
1496 user_table = Table("user", meta)
1497 insp = inspect(engine)
1498 insp.reflect_table(user_table, None)
1500 .. versionchanged:: 1.4 Renamed from ``reflecttable`` to
1501 ``reflect_table``
1503 :param table: a :class:`~sqlalchemy.schema.Table` instance.
1504 :param include_columns: a list of string column names to include
1505 in the reflection process. If ``None``, all columns are reflected.
1507 """
1509 if _extend_on is not None:
1510 if table in _extend_on:
1511 return
1512 else:
1513 _extend_on.add(table)
1515 dialect = self.bind.dialect
1517 with self._operation_context() as conn:
1518 schema = conn.schema_for_object(table)
1520 table_name = table.name
1522 # get table-level arguments that are specifically
1523 # intended for reflection, e.g. oracle_resolve_synonyms.
1524 # these are unconditionally passed to related Table
1525 # objects
1526 reflection_options = {
1527 k: table.dialect_kwargs.get(k)
1528 for k in dialect.reflection_options
1529 if k in table.dialect_kwargs
1530 }
1532 table_key = (schema, table_name)
1533 if _reflect_info is None or table_key not in _reflect_info.columns:
1534 _reflect_info = self._get_reflection_info(
1535 schema,
1536 filter_names=[table_name],
1537 kind=ObjectKind.ANY,
1538 scope=ObjectScope.ANY,
1539 _reflect_info=_reflect_info,
1540 **table.dialect_kwargs,
1541 )
1542 if table_key in _reflect_info.unreflectable:
1543 raise _reflect_info.unreflectable[table_key]
1545 if table_key not in _reflect_info.columns:
1546 raise exc.NoSuchTableError(table_name)
1548 # reflect table options, like mysql_engine
1549 if _reflect_info.table_options:
1550 tbl_opts = _reflect_info.table_options.get(table_key)
1551 if tbl_opts:
1552 # add additional kwargs to the Table if the dialect
1553 # returned them
1554 table._validate_dialect_kwargs(tbl_opts)
1556 found_table = False
1557 cols_by_orig_name: Dict[str, sa_schema.Column[Any]] = {}
1559 for col_d in _reflect_info.columns[table_key]:
1560 found_table = True
1562 self._reflect_column(
1563 table,
1564 col_d,
1565 include_columns,
1566 exclude_columns,
1567 cols_by_orig_name,
1568 )
1570 # NOTE: support tables/views with no columns
1571 if not found_table and not self.has_table(table_name, schema):
1572 raise exc.NoSuchTableError(table_name)
1574 self._reflect_pk(
1575 _reflect_info, table_key, table, cols_by_orig_name, exclude_columns
1576 )
1578 self._reflect_fk(
1579 _reflect_info,
1580 table_key,
1581 table,
1582 cols_by_orig_name,
1583 include_columns,
1584 exclude_columns,
1585 resolve_fks,
1586 _extend_on,
1587 reflection_options,
1588 )
1590 self._reflect_indexes(
1591 _reflect_info,
1592 table_key,
1593 table,
1594 cols_by_orig_name,
1595 include_columns,
1596 exclude_columns,
1597 reflection_options,
1598 )
1600 self._reflect_unique_constraints(
1601 _reflect_info,
1602 table_key,
1603 table,
1604 cols_by_orig_name,
1605 include_columns,
1606 exclude_columns,
1607 reflection_options,
1608 )
1610 self._reflect_check_constraints(
1611 _reflect_info,
1612 table_key,
1613 table,
1614 cols_by_orig_name,
1615 include_columns,
1616 exclude_columns,
1617 reflection_options,
1618 )
1620 self._reflect_table_comment(
1621 _reflect_info,
1622 table_key,
1623 table,
1624 reflection_options,
1625 )
1627 def _reflect_column(
1628 self,
1629 table: sa_schema.Table,
1630 col_d: ReflectedColumn,
1631 include_columns: Optional[Collection[str]],
1632 exclude_columns: Collection[str],
1633 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1634 ) -> None:
1635 orig_name = col_d["name"]
1637 table.metadata.dispatch.column_reflect(self, table, col_d)
1638 table.dispatch.column_reflect(self, table, col_d)
1640 # fetch name again as column_reflect is allowed to
1641 # change it
1642 name = col_d["name"]
1643 if (include_columns and name not in include_columns) or (
1644 exclude_columns and name in exclude_columns
1645 ):
1646 return
1648 coltype = col_d["type"]
1650 col_kw = {
1651 k: col_d[k] # type: ignore[literal-required]
1652 for k in [
1653 "nullable",
1654 "autoincrement",
1655 "quote",
1656 "info",
1657 "key",
1658 "comment",
1659 ]
1660 if k in col_d
1661 }
1663 if "dialect_options" in col_d:
1664 col_kw.update(col_d["dialect_options"])
1666 colargs = []
1667 default: Any
1668 if col_d.get("default") is not None:
1669 default_text = col_d["default"]
1670 assert default_text is not None
1671 if isinstance(default_text, TextClause):
1672 default = sa_schema.DefaultClause(
1673 default_text, _reflected=True
1674 )
1675 elif not isinstance(default_text, sa_schema.FetchedValue):
1676 default = sa_schema.DefaultClause(
1677 sql.text(default_text), _reflected=True
1678 )
1679 else:
1680 default = default_text
1681 colargs.append(default)
1683 if "computed" in col_d:
1684 computed = sa_schema.Computed(**col_d["computed"])
1685 colargs.append(computed)
1687 if "identity" in col_d:
1688 identity = sa_schema.Identity(**col_d["identity"])
1689 colargs.append(identity)
1691 cols_by_orig_name[orig_name] = col = sa_schema.Column(
1692 name, coltype, *colargs, **col_kw
1693 )
1695 if col.key in table.primary_key:
1696 col.primary_key = True
1697 table.append_column(col, replace_existing=True)
1699 def _reflect_pk(
1700 self,
1701 _reflect_info: _ReflectionInfo,
1702 table_key: TableKey,
1703 table: sa_schema.Table,
1704 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1705 exclude_columns: Collection[str],
1706 ) -> None:
1707 pk_cons = _reflect_info.pk_constraint.get(table_key)
1708 if pk_cons:
1709 pk_cols = [
1710 cols_by_orig_name[pk]
1711 for pk in pk_cons["constrained_columns"]
1712 if pk in cols_by_orig_name and pk not in exclude_columns
1713 ]
1715 # update pk constraint name, comment and dialect_kwargs
1716 table.primary_key.name = pk_cons.get("name")
1717 table.primary_key.comment = pk_cons.get("comment", None)
1718 dialect_options = pk_cons.get("dialect_options")
1719 if dialect_options:
1720 table.primary_key.dialect_kwargs.update(dialect_options)
1722 # tell the PKConstraint to re-initialize
1723 # its column collection
1724 table.primary_key._reload(pk_cols)
1726 def _reflect_fk(
1727 self,
1728 _reflect_info: _ReflectionInfo,
1729 table_key: TableKey,
1730 table: sa_schema.Table,
1731 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1732 include_columns: Optional[Collection[str]],
1733 exclude_columns: Collection[str],
1734 resolve_fks: bool,
1735 _extend_on: Optional[Set[sa_schema.Table]],
1736 reflection_options: Dict[str, Any],
1737 ) -> None:
1738 fkeys = _reflect_info.foreign_keys.get(table_key, [])
1739 for fkey_d in fkeys:
1740 conname = fkey_d["name"]
1741 # look for columns by orig name in cols_by_orig_name,
1742 # but support columns that are in-Python only as fallback
1743 constrained_columns = [
1744 cols_by_orig_name[c].key if c in cols_by_orig_name else c
1745 for c in fkey_d["constrained_columns"]
1746 ]
1748 if (
1749 exclude_columns
1750 and set(constrained_columns).intersection(exclude_columns)
1751 or (
1752 include_columns
1753 and set(constrained_columns).difference(include_columns)
1754 )
1755 ):
1756 continue
1758 referred_schema = fkey_d["referred_schema"]
1759 referred_table = fkey_d["referred_table"]
1760 referred_columns = fkey_d["referred_columns"]
1761 refspec = []
1762 if referred_schema is not None:
1763 if resolve_fks:
1764 sa_schema.Table(
1765 referred_table,
1766 table.metadata,
1767 schema=referred_schema,
1768 autoload_with=self.bind,
1769 _extend_on=_extend_on,
1770 _reflect_info=_reflect_info,
1771 **reflection_options,
1772 )
1773 for column in referred_columns:
1774 refspec.append(
1775 ".".join([referred_schema, referred_table, column])
1776 )
1777 else:
1778 if resolve_fks:
1779 sa_schema.Table(
1780 referred_table,
1781 table.metadata,
1782 autoload_with=self.bind,
1783 schema=sa_schema.BLANK_SCHEMA,
1784 _extend_on=_extend_on,
1785 _reflect_info=_reflect_info,
1786 **reflection_options,
1787 )
1788 for column in referred_columns:
1789 refspec.append(".".join([referred_table, column]))
1790 if "options" in fkey_d:
1791 options = fkey_d["options"]
1792 else:
1793 options = {}
1795 try:
1796 table.append_constraint(
1797 sa_schema.ForeignKeyConstraint(
1798 constrained_columns,
1799 refspec,
1800 conname,
1801 link_to_name=True,
1802 comment=fkey_d.get("comment"),
1803 **options,
1804 )
1805 )
1806 except exc.ConstraintColumnNotFoundError:
1807 util.warn(
1808 f"On reflected table {table.name}, skipping reflection of "
1809 "foreign key constraint "
1810 f"{conname}; one or more subject columns within "
1811 f"name(s) {', '.join(constrained_columns)} are not "
1812 "present in the table"
1813 )
1815 _index_sort_exprs = {
1816 "asc": operators.asc_op,
1817 "desc": operators.desc_op,
1818 "nulls_first": operators.nulls_first_op,
1819 "nulls_last": operators.nulls_last_op,
1820 }
1822 def _reflect_indexes(
1823 self,
1824 _reflect_info: _ReflectionInfo,
1825 table_key: TableKey,
1826 table: sa_schema.Table,
1827 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1828 include_columns: Optional[Collection[str]],
1829 exclude_columns: Collection[str],
1830 reflection_options: Dict[str, Any],
1831 ) -> None:
1832 # Indexes
1833 indexes = _reflect_info.indexes.get(table_key, [])
1834 for index_d in indexes:
1835 name = index_d["name"]
1836 columns = index_d["column_names"]
1837 expressions = index_d.get("expressions")
1838 column_sorting = index_d.get("column_sorting", {})
1839 unique = index_d["unique"]
1840 flavor = index_d.get("type", "index")
1841 dialect_options = index_d.get("dialect_options", {})
1843 duplicates = index_d.get("duplicates_constraint")
1844 if include_columns and not set(columns).issubset(include_columns):
1845 continue
1846 if duplicates:
1847 continue
1848 # look for columns by orig name in cols_by_orig_name,
1849 # but support columns that are in-Python only as fallback
1850 idx_element: Any
1851 idx_elements = []
1852 for index, c in enumerate(columns):
1853 if c is None:
1854 if not expressions:
1855 util.warn(
1856 f"Skipping {flavor} {name!r} because key "
1857 f"{index + 1} reflected as None but no "
1858 "'expressions' were returned"
1859 )
1860 break
1861 idx_element = sql.text(expressions[index])
1862 else:
1863 try:
1864 if c in cols_by_orig_name:
1865 idx_element = cols_by_orig_name[c]
1866 else:
1867 idx_element = table.c[c]
1868 except KeyError:
1869 util.warn(
1870 f"{flavor} key {c!r} was not located in "
1871 f"columns for table {table.name!r}"
1872 )
1873 continue
1874 for option in column_sorting.get(c, ()):
1875 if option in self._index_sort_exprs:
1876 op = self._index_sort_exprs[option]
1877 idx_element = op(idx_element)
1878 idx_elements.append(idx_element)
1879 else:
1880 sa_schema.Index(
1881 name,
1882 *idx_elements,
1883 _table=table,
1884 unique=unique,
1885 **dialect_options,
1886 )
1888 def _reflect_unique_constraints(
1889 self,
1890 _reflect_info: _ReflectionInfo,
1891 table_key: TableKey,
1892 table: sa_schema.Table,
1893 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1894 include_columns: Optional[Collection[str]],
1895 exclude_columns: Collection[str],
1896 reflection_options: Dict[str, Any],
1897 ) -> None:
1898 constraints = _reflect_info.unique_constraints.get(table_key, [])
1899 # Unique Constraints
1900 for const_d in constraints:
1901 conname = const_d["name"]
1902 columns = const_d["column_names"]
1903 comment = const_d.get("comment")
1904 duplicates = const_d.get("duplicates_index")
1905 dialect_options = const_d.get("dialect_options", {})
1906 if include_columns and not set(columns).issubset(include_columns):
1907 continue
1908 if duplicates:
1909 continue
1910 # look for columns by orig name in cols_by_orig_name,
1911 # but support columns that are in-Python only as fallback
1912 constrained_cols = []
1913 for c in columns:
1914 try:
1915 constrained_col = (
1916 cols_by_orig_name[c]
1917 if c in cols_by_orig_name
1918 else table.c[c]
1919 )
1920 except KeyError:
1921 util.warn(
1922 "unique constraint key '%s' was not located in "
1923 "columns for table '%s'" % (c, table.name)
1924 )
1925 else:
1926 constrained_cols.append(constrained_col)
1927 table.append_constraint(
1928 sa_schema.UniqueConstraint(
1929 *constrained_cols,
1930 name=conname,
1931 comment=comment,
1932 **dialect_options,
1933 )
1934 )
1936 def _reflect_check_constraints(
1937 self,
1938 _reflect_info: _ReflectionInfo,
1939 table_key: TableKey,
1940 table: sa_schema.Table,
1941 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1942 include_columns: Optional[Collection[str]],
1943 exclude_columns: Collection[str],
1944 reflection_options: Dict[str, Any],
1945 ) -> None:
1946 constraints = _reflect_info.check_constraints.get(table_key, [])
1947 for const_d in constraints:
1948 table.append_constraint(sa_schema.CheckConstraint(**const_d))
1950 def _reflect_table_comment(
1951 self,
1952 _reflect_info: _ReflectionInfo,
1953 table_key: TableKey,
1954 table: sa_schema.Table,
1955 reflection_options: Dict[str, Any],
1956 ) -> None:
1957 comment_dict = _reflect_info.table_comment.get(table_key)
1958 if comment_dict:
1959 table.comment = comment_dict["text"]
1961 def _get_reflection_info(
1962 self,
1963 schema: Optional[str] = None,
1964 filter_names: Optional[Collection[str]] = None,
1965 available: Optional[Collection[str]] = None,
1966 _reflect_info: Optional[_ReflectionInfo] = None,
1967 **kw: Any,
1968 ) -> _ReflectionInfo:
1969 kw["schema"] = schema
1971 if filter_names and available and len(filter_names) > 100:
1972 fraction = len(filter_names) / len(available)
1973 else:
1974 fraction = None
1976 unreflectable: Dict[TableKey, exc.UnreflectableTableError]
1977 kw["unreflectable"] = unreflectable = {}
1979 has_result: bool = True
1981 def run(
1982 meth: Any,
1983 *,
1984 optional: bool = False,
1985 check_filter_names_from_meth: bool = False,
1986 ) -> Any:
1987 nonlocal has_result
1988 # simple heuristic to improve reflection performance if a
1989 # dialect implements multi_reflection:
1990 # if more than 50% of the tables in the db are in filter_names
1991 # load all the tables, since it's most likely faster to avoid
1992 # a filter on that many tables.
1993 if (
1994 fraction is None
1995 or fraction <= 0.5
1996 or not self.dialect._overrides_default(meth.__name__)
1997 ):
1998 _fn = filter_names
1999 else:
2000 _fn = None
2001 try:
2002 if has_result:
2003 res = meth(filter_names=_fn, **kw)
2004 if check_filter_names_from_meth and not res:
2005 # method returned no result data.
2006 # skip any future call methods
2007 has_result = False
2008 else:
2009 res = {}
2010 except NotImplementedError:
2011 if not optional:
2012 raise
2013 res = {}
2014 return res
2016 info = _ReflectionInfo(
2017 columns=run(
2018 self.get_multi_columns, check_filter_names_from_meth=True
2019 ),
2020 pk_constraint=run(self.get_multi_pk_constraint),
2021 foreign_keys=run(self.get_multi_foreign_keys),
2022 indexes=run(self.get_multi_indexes),
2023 unique_constraints=run(
2024 self.get_multi_unique_constraints, optional=True
2025 ),
2026 table_comment=run(self.get_multi_table_comment, optional=True),
2027 check_constraints=run(
2028 self.get_multi_check_constraints, optional=True
2029 ),
2030 table_options=run(self.get_multi_table_options, optional=True),
2031 unreflectable=unreflectable,
2032 )
2033 if _reflect_info:
2034 _reflect_info.update(info)
2035 return _reflect_info
2036 else:
2037 return info
2040@final
2041class ReflectionDefaults:
2042 """provides blank default values for reflection methods."""
2044 @classmethod
2045 def columns(cls) -> List[ReflectedColumn]:
2046 return []
2048 @classmethod
2049 def pk_constraint(cls) -> ReflectedPrimaryKeyConstraint:
2050 return {
2051 "name": None,
2052 "constrained_columns": [],
2053 }
2055 @classmethod
2056 def foreign_keys(cls) -> List[ReflectedForeignKeyConstraint]:
2057 return []
2059 @classmethod
2060 def indexes(cls) -> List[ReflectedIndex]:
2061 return []
2063 @classmethod
2064 def unique_constraints(cls) -> List[ReflectedUniqueConstraint]:
2065 return []
2067 @classmethod
2068 def check_constraints(cls) -> List[ReflectedCheckConstraint]:
2069 return []
2071 @classmethod
2072 def table_options(cls) -> Dict[str, Any]:
2073 return {}
2075 @classmethod
2076 def table_comment(cls) -> ReflectedTableComment:
2077 return {"text": None}
2080@dataclass
2081class _ReflectionInfo:
2082 columns: Dict[TableKey, List[ReflectedColumn]]
2083 pk_constraint: Dict[TableKey, Optional[ReflectedPrimaryKeyConstraint]]
2084 foreign_keys: Dict[TableKey, List[ReflectedForeignKeyConstraint]]
2085 indexes: Dict[TableKey, List[ReflectedIndex]]
2086 # optionals
2087 unique_constraints: Dict[TableKey, List[ReflectedUniqueConstraint]]
2088 table_comment: Dict[TableKey, Optional[ReflectedTableComment]]
2089 check_constraints: Dict[TableKey, List[ReflectedCheckConstraint]]
2090 table_options: Dict[TableKey, Dict[str, Any]]
2091 unreflectable: Dict[TableKey, exc.UnreflectableTableError]
2093 def update(self, other: _ReflectionInfo) -> None:
2094 for k, v in self.__dict__.items():
2095 ov = getattr(other, k)
2096 if ov is not None:
2097 if v is None:
2098 setattr(self, k, ov)
2099 else:
2100 v.update(ov)