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