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-2024 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 TextClause
60from ..sql.type_api import TypeEngine
61from ..sql.visitors import InternalTraversal
62from ..util import topological
64if TYPE_CHECKING:
65 from .interfaces import Dialect
66 from .interfaces import ReflectedCheckConstraint
67 from .interfaces import ReflectedColumn
68 from .interfaces import ReflectedForeignKeyConstraint
69 from .interfaces import ReflectedIndex
70 from .interfaces import ReflectedPrimaryKeyConstraint
71 from .interfaces import ReflectedTableComment
72 from .interfaces import ReflectedUniqueConstraint
73 from .interfaces import TableKey
75_R = TypeVar("_R")
78@util.decorator
79def cache(
80 fn: Callable[..., _R],
81 self: Dialect,
82 con: Connection,
83 *args: Any,
84 **kw: Any,
85) -> _R:
86 info_cache = kw.get("info_cache", None)
87 if info_cache is None:
88 return fn(self, con, *args, **kw)
89 exclude = {"info_cache", "unreflectable"}
90 key = (
91 fn.__name__,
92 tuple(a for a in args if isinstance(a, str)),
93 tuple((k, v) for k, v in kw.items() if k not in exclude),
94 )
95 ret: _R = info_cache.get(key)
96 if ret is None:
97 ret = fn(self, con, *args, **kw)
98 info_cache[key] = ret
99 return ret
102def flexi_cache(
103 *traverse_args: Tuple[str, InternalTraversal]
104) -> Callable[[Callable[..., _R]], Callable[..., _R]]:
105 @util.decorator
106 def go(
107 fn: Callable[..., _R],
108 self: Dialect,
109 con: Connection,
110 *args: Any,
111 **kw: Any,
112 ) -> _R:
113 info_cache = kw.get("info_cache", None)
114 if info_cache is None:
115 return fn(self, con, *args, **kw)
116 key = _ad_hoc_cache_key_from_args((fn.__name__,), traverse_args, args)
117 ret: _R = info_cache.get(key)
118 if ret is None:
119 ret = fn(self, con, *args, **kw)
120 info_cache[key] = ret
121 return ret
123 return go
126@unique
127class ObjectKind(Flag):
128 """Enumerator that indicates which kind of object to return when calling
129 the ``get_multi`` methods.
131 This is a Flag enum, so custom combinations can be passed. For example,
132 to reflect tables and plain views ``ObjectKind.TABLE | ObjectKind.VIEW``
133 may be used.
135 .. note::
136 Not all dialect may support all kind of object. If a dialect does
137 not support a particular object an empty dict is returned.
138 In case a dialect supports an object, but the requested method
139 is not applicable for the specified kind the default value
140 will be returned for each reflected object. For example reflecting
141 check constraints of view return a dict with all the views with
142 empty lists as values.
143 """
145 TABLE = auto()
146 "Reflect table objects"
147 VIEW = auto()
148 "Reflect plain view objects"
149 MATERIALIZED_VIEW = auto()
150 "Reflect materialized view object"
152 ANY_VIEW = VIEW | MATERIALIZED_VIEW
153 "Reflect any kind of view objects"
154 ANY = TABLE | VIEW | MATERIALIZED_VIEW
155 "Reflect all type of objects"
158@unique
159class ObjectScope(Flag):
160 """Enumerator that indicates which scope to use when calling
161 the ``get_multi`` methods.
162 """
164 DEFAULT = auto()
165 "Include default scope"
166 TEMPORARY = auto()
167 "Include only temp scope"
168 ANY = DEFAULT | TEMPORARY
169 "Include both default and temp scope"
172@inspection._self_inspects
173class Inspector(inspection.Inspectable["Inspector"]):
174 """Performs database schema inspection.
176 The Inspector acts as a proxy to the reflection methods of the
177 :class:`~sqlalchemy.engine.interfaces.Dialect`, providing a
178 consistent interface as well as caching support for previously
179 fetched metadata.
181 A :class:`_reflection.Inspector` object is usually created via the
182 :func:`_sa.inspect` function, which may be passed an
183 :class:`_engine.Engine`
184 or a :class:`_engine.Connection`::
186 from sqlalchemy import inspect, create_engine
187 engine = create_engine('...')
188 insp = inspect(engine)
190 Where above, the :class:`~sqlalchemy.engine.interfaces.Dialect` associated
191 with the engine may opt to return an :class:`_reflection.Inspector`
192 subclass that
193 provides additional methods specific to the dialect's target database.
195 """
197 bind: Union[Engine, Connection]
198 engine: Engine
199 _op_context_requires_connect: bool
200 dialect: Dialect
201 info_cache: Dict[Any, Any]
203 @util.deprecated(
204 "1.4",
205 "The __init__() method on :class:`_reflection.Inspector` "
206 "is deprecated and "
207 "will be removed in a future release. Please use the "
208 ":func:`.sqlalchemy.inspect` "
209 "function on an :class:`_engine.Engine` or "
210 ":class:`_engine.Connection` "
211 "in order to "
212 "acquire an :class:`_reflection.Inspector`.",
213 )
214 def __init__(self, bind: Union[Engine, Connection]):
215 """Initialize a new :class:`_reflection.Inspector`.
217 :param bind: a :class:`~sqlalchemy.engine.Connection`,
218 which is typically an instance of
219 :class:`~sqlalchemy.engine.Engine` or
220 :class:`~sqlalchemy.engine.Connection`.
222 For a dialect-specific instance of :class:`_reflection.Inspector`, see
223 :meth:`_reflection.Inspector.from_engine`
225 """
226 self._init_legacy(bind)
228 @classmethod
229 def _construct(
230 cls, init: Callable[..., Any], bind: Union[Engine, Connection]
231 ) -> Inspector:
232 if hasattr(bind.dialect, "inspector"):
233 cls = bind.dialect.inspector
235 self = cls.__new__(cls)
236 init(self, bind)
237 return self
239 def _init_legacy(self, bind: Union[Engine, Connection]) -> None:
240 if hasattr(bind, "exec_driver_sql"):
241 self._init_connection(bind) # type: ignore[arg-type]
242 else:
243 self._init_engine(bind)
245 def _init_engine(self, engine: Engine) -> None:
246 self.bind = self.engine = engine
247 engine.connect().close()
248 self._op_context_requires_connect = True
249 self.dialect = self.engine.dialect
250 self.info_cache = {}
252 def _init_connection(self, connection: Connection) -> None:
253 self.bind = connection
254 self.engine = connection.engine
255 self._op_context_requires_connect = False
256 self.dialect = self.engine.dialect
257 self.info_cache = {}
259 def clear_cache(self) -> None:
260 """reset the cache for this :class:`.Inspector`.
262 Inspection methods that have data cached will emit SQL queries
263 when next called to get new data.
265 .. versionadded:: 2.0
267 """
268 self.info_cache.clear()
270 @classmethod
271 @util.deprecated(
272 "1.4",
273 "The from_engine() method on :class:`_reflection.Inspector` "
274 "is deprecated and "
275 "will be removed in a future release. Please use the "
276 ":func:`.sqlalchemy.inspect` "
277 "function on an :class:`_engine.Engine` or "
278 ":class:`_engine.Connection` "
279 "in order to "
280 "acquire an :class:`_reflection.Inspector`.",
281 )
282 def from_engine(cls, bind: Engine) -> Inspector:
283 """Construct a new dialect-specific Inspector object from the given
284 engine or connection.
286 :param bind: a :class:`~sqlalchemy.engine.Connection`
287 or :class:`~sqlalchemy.engine.Engine`.
289 This method differs from direct a direct constructor call of
290 :class:`_reflection.Inspector` in that the
291 :class:`~sqlalchemy.engine.interfaces.Dialect` is given a chance to
292 provide a dialect-specific :class:`_reflection.Inspector` instance,
293 which may
294 provide additional methods.
296 See the example at :class:`_reflection.Inspector`.
298 """
299 return cls._construct(cls._init_legacy, bind)
301 @inspection._inspects(Engine)
302 def _engine_insp(bind: Engine) -> Inspector: # type: ignore[misc]
303 return Inspector._construct(Inspector._init_engine, bind)
305 @inspection._inspects(Connection)
306 def _connection_insp(bind: Connection) -> Inspector: # type: ignore[misc]
307 return Inspector._construct(Inspector._init_connection, bind)
309 @contextlib.contextmanager
310 def _operation_context(self) -> Generator[Connection, None, None]:
311 """Return a context that optimizes for multiple operations on a single
312 transaction.
314 This essentially allows connect()/close() to be called if we detected
315 that we're against an :class:`_engine.Engine` and not a
316 :class:`_engine.Connection`.
318 """
319 conn: Connection
320 if self._op_context_requires_connect:
321 conn = self.bind.connect() # type: ignore[union-attr]
322 else:
323 conn = self.bind # type: ignore[assignment]
324 try:
325 yield conn
326 finally:
327 if self._op_context_requires_connect:
328 conn.close()
330 @contextlib.contextmanager
331 def _inspection_context(self) -> Generator[Inspector, None, None]:
332 """Return an :class:`_reflection.Inspector`
333 from this one that will run all
334 operations on a single connection.
336 """
338 with self._operation_context() as conn:
339 sub_insp = self._construct(self.__class__._init_connection, conn)
340 sub_insp.info_cache = self.info_cache
341 yield sub_insp
343 @property
344 def default_schema_name(self) -> Optional[str]:
345 """Return the default schema name presented by the dialect
346 for the current engine's database user.
348 E.g. this is typically ``public`` for PostgreSQL and ``dbo``
349 for SQL Server.
351 """
352 return self.dialect.default_schema_name
354 def get_schema_names(self, **kw: Any) -> List[str]:
355 r"""Return all schema names.
357 :param \**kw: Additional keyword argument to pass to the dialect
358 specific implementation. See the documentation of the dialect
359 in use for more information.
360 """
362 with self._operation_context() as conn:
363 return self.dialect.get_schema_names(
364 conn, info_cache=self.info_cache, **kw
365 )
367 def get_table_names(
368 self, schema: Optional[str] = None, **kw: Any
369 ) -> List[str]:
370 r"""Return all table names within a particular schema.
372 The names are expected to be real tables only, not views.
373 Views are instead returned using the
374 :meth:`_reflection.Inspector.get_view_names` and/or
375 :meth:`_reflection.Inspector.get_materialized_view_names`
376 methods.
378 :param schema: Schema name. If ``schema`` is left at ``None``, the
379 database's default schema is
380 used, else the named schema is searched. If the database does not
381 support named schemas, behavior is undefined if ``schema`` is not
382 passed as ``None``. For special quoting, use :class:`.quoted_name`.
383 :param \**kw: Additional keyword argument to pass to the dialect
384 specific implementation. See the documentation of the dialect
385 in use for more information.
387 .. seealso::
389 :meth:`_reflection.Inspector.get_sorted_table_and_fkc_names`
391 :attr:`_schema.MetaData.sorted_tables`
393 """
395 with self._operation_context() as conn:
396 return self.dialect.get_table_names(
397 conn, schema, info_cache=self.info_cache, **kw
398 )
400 def has_table(
401 self, table_name: str, schema: Optional[str] = None, **kw: Any
402 ) -> bool:
403 r"""Return True if the backend has a table, view, or temporary
404 table of the given name.
406 :param table_name: name of the table to check
407 :param schema: schema name to query, if not the default schema.
408 :param \**kw: Additional keyword argument to pass to the dialect
409 specific implementation. See the documentation of the dialect
410 in use for more information.
412 .. versionadded:: 1.4 - the :meth:`.Inspector.has_table` method
413 replaces the :meth:`_engine.Engine.has_table` method.
415 .. versionchanged:: 2.0:: :meth:`.Inspector.has_table` now formally
416 supports checking for additional table-like objects:
418 * any type of views (plain or materialized)
419 * temporary tables of any kind
421 Previously, these two checks were not formally specified and
422 different dialects would vary in their behavior. The dialect
423 testing suite now includes tests for all of these object types
424 and should be supported by all SQLAlchemy-included dialects.
425 Support among third party dialects may be lagging, however.
427 """
428 with self._operation_context() as conn:
429 return self.dialect.has_table(
430 conn, table_name, schema, info_cache=self.info_cache, **kw
431 )
433 def has_sequence(
434 self, sequence_name: str, schema: Optional[str] = None, **kw: Any
435 ) -> bool:
436 r"""Return True if the backend has a sequence with the given name.
438 :param sequence_name: name of the sequence to check
439 :param schema: schema name to query, if not the default schema.
440 :param \**kw: Additional keyword argument to pass to the dialect
441 specific implementation. See the documentation of the dialect
442 in use for more information.
444 .. versionadded:: 1.4
446 """
447 with self._operation_context() as conn:
448 return self.dialect.has_sequence(
449 conn, sequence_name, schema, info_cache=self.info_cache, **kw
450 )
452 def has_index(
453 self,
454 table_name: str,
455 index_name: str,
456 schema: Optional[str] = None,
457 **kw: Any,
458 ) -> bool:
459 r"""Check the existence of a particular index name in the database.
461 :param table_name: the name of the table the index belongs to
462 :param index_name: the name of the index to check
463 :param schema: schema name to query, if not the default schema.
464 :param \**kw: Additional keyword argument to pass to the dialect
465 specific implementation. See the documentation of the dialect
466 in use for more information.
468 .. versionadded:: 2.0
470 """
471 with self._operation_context() as conn:
472 return self.dialect.has_index(
473 conn,
474 table_name,
475 index_name,
476 schema,
477 info_cache=self.info_cache,
478 **kw,
479 )
481 def has_schema(self, schema_name: str, **kw: Any) -> bool:
482 r"""Return True if the backend has a schema with the given name.
484 :param schema_name: name of the schema to check
485 :param \**kw: Additional keyword argument to pass to the dialect
486 specific implementation. See the documentation of the dialect
487 in use for more information.
489 .. versionadded:: 2.0
491 """
492 with self._operation_context() as conn:
493 return self.dialect.has_schema(
494 conn, schema_name, info_cache=self.info_cache, **kw
495 )
497 def get_sorted_table_and_fkc_names(
498 self,
499 schema: Optional[str] = None,
500 **kw: Any,
501 ) -> List[Tuple[Optional[str], List[Tuple[str, Optional[str]]]]]:
502 r"""Return dependency-sorted table and foreign key constraint names in
503 referred to within a particular schema.
505 This will yield 2-tuples of
506 ``(tablename, [(tname, fkname), (tname, fkname), ...])``
507 consisting of table names in CREATE order grouped with the foreign key
508 constraint names that are not detected as belonging to a cycle.
509 The final element
510 will be ``(None, [(tname, fkname), (tname, fkname), ..])``
511 which will consist of remaining
512 foreign key constraint names that would require a separate CREATE
513 step after-the-fact, based on dependencies between tables.
515 :param schema: schema name to query, if not the default schema.
516 :param \**kw: Additional keyword argument to pass to the dialect
517 specific implementation. See the documentation of the dialect
518 in use for more information.
520 .. seealso::
522 :meth:`_reflection.Inspector.get_table_names`
524 :func:`.sort_tables_and_constraints` - similar method which works
525 with an already-given :class:`_schema.MetaData`.
527 """
529 return [
530 (
531 table_key[1] if table_key else None,
532 [(tname, fks) for (_, tname), fks in fk_collection],
533 )
534 for (
535 table_key,
536 fk_collection,
537 ) in self.sort_tables_on_foreign_key_dependency(
538 consider_schemas=(schema,)
539 )
540 ]
542 def sort_tables_on_foreign_key_dependency(
543 self,
544 consider_schemas: Collection[Optional[str]] = (None,),
545 **kw: Any,
546 ) -> List[
547 Tuple[
548 Optional[Tuple[Optional[str], str]],
549 List[Tuple[Tuple[Optional[str], str], Optional[str]]],
550 ]
551 ]:
552 r"""Return dependency-sorted table and foreign key constraint names
553 referred to within multiple schemas.
555 This method may be compared to
556 :meth:`.Inspector.get_sorted_table_and_fkc_names`, which
557 works on one schema at a time; here, the method is a generalization
558 that will consider multiple schemas at once including that it will
559 resolve for cross-schema foreign keys.
561 .. versionadded:: 2.0
563 """
564 SchemaTab = Tuple[Optional[str], str]
566 tuples: Set[Tuple[SchemaTab, SchemaTab]] = set()
567 remaining_fkcs: Set[Tuple[SchemaTab, Optional[str]]] = set()
568 fknames_for_table: Dict[SchemaTab, Set[Optional[str]]] = {}
569 tnames: List[SchemaTab] = []
571 for schname in consider_schemas:
572 schema_fkeys = self.get_multi_foreign_keys(schname, **kw)
573 tnames.extend(schema_fkeys)
574 for (_, tname), fkeys in schema_fkeys.items():
575 fknames_for_table[(schname, tname)] = {
576 fk["name"] for fk in fkeys
577 }
578 for fkey in fkeys:
579 if (
580 tname != fkey["referred_table"]
581 or schname != fkey["referred_schema"]
582 ):
583 tuples.add(
584 (
585 (
586 fkey["referred_schema"],
587 fkey["referred_table"],
588 ),
589 (schname, tname),
590 )
591 )
592 try:
593 candidate_sort = list(topological.sort(tuples, tnames))
594 except exc.CircularDependencyError as err:
595 edge: Tuple[SchemaTab, SchemaTab]
596 for edge in err.edges:
597 tuples.remove(edge)
598 remaining_fkcs.update(
599 (edge[1], fkc) for fkc in fknames_for_table[edge[1]]
600 )
602 candidate_sort = list(topological.sort(tuples, tnames))
603 ret: List[
604 Tuple[Optional[SchemaTab], List[Tuple[SchemaTab, Optional[str]]]]
605 ]
606 ret = [
607 (
608 (schname, tname),
609 [
610 ((schname, tname), fk)
611 for fk in fknames_for_table[(schname, tname)].difference(
612 name for _, name in remaining_fkcs
613 )
614 ],
615 )
616 for (schname, tname) in candidate_sort
617 ]
618 return ret + [(None, list(remaining_fkcs))]
620 def get_temp_table_names(self, **kw: Any) -> List[str]:
621 r"""Return a list of temporary table names for the current bind.
623 This method is unsupported by most dialects; currently
624 only Oracle, PostgreSQL and SQLite implements it.
626 :param \**kw: Additional keyword argument to pass to the dialect
627 specific implementation. See the documentation of the dialect
628 in use for more information.
630 """
632 with self._operation_context() as conn:
633 return self.dialect.get_temp_table_names(
634 conn, info_cache=self.info_cache, **kw
635 )
637 def get_temp_view_names(self, **kw: Any) -> List[str]:
638 r"""Return a list of temporary view names for the current bind.
640 This method is unsupported by most dialects; currently
641 only PostgreSQL and SQLite implements it.
643 :param \**kw: Additional keyword argument to pass to the dialect
644 specific implementation. See the documentation of the dialect
645 in use for more information.
647 """
648 with self._operation_context() as conn:
649 return self.dialect.get_temp_view_names(
650 conn, info_cache=self.info_cache, **kw
651 )
653 def get_table_options(
654 self, table_name: str, schema: Optional[str] = None, **kw: Any
655 ) -> Dict[str, Any]:
656 r"""Return a dictionary of options specified when the table of the
657 given name was created.
659 This currently includes some options that apply to MySQL and Oracle
660 tables.
662 :param table_name: string name of the table. For special quoting,
663 use :class:`.quoted_name`.
665 :param schema: string schema name; if omitted, uses the default schema
666 of the database connection. For special quoting,
667 use :class:`.quoted_name`.
669 :param \**kw: Additional keyword argument to pass to the dialect
670 specific implementation. See the documentation of the dialect
671 in use for more information.
673 :return: a dict with the table options. The returned keys depend on the
674 dialect in use. Each one is prefixed with the dialect name.
676 .. seealso:: :meth:`Inspector.get_multi_table_options`
678 """
679 with self._operation_context() as conn:
680 return self.dialect.get_table_options(
681 conn, table_name, schema, info_cache=self.info_cache, **kw
682 )
684 def get_multi_table_options(
685 self,
686 schema: Optional[str] = None,
687 filter_names: Optional[Sequence[str]] = None,
688 kind: ObjectKind = ObjectKind.TABLE,
689 scope: ObjectScope = ObjectScope.DEFAULT,
690 **kw: Any,
691 ) -> Dict[TableKey, Dict[str, Any]]:
692 r"""Return a dictionary of options specified when the tables in the
693 given schema were created.
695 The tables can be filtered by passing the names to use to
696 ``filter_names``.
698 This currently includes some options that apply to MySQL and Oracle
699 tables.
701 :param schema: string schema name; if omitted, uses the default schema
702 of the database connection. For special quoting,
703 use :class:`.quoted_name`.
705 :param filter_names: optionally return information only for the
706 objects listed here.
708 :param kind: a :class:`.ObjectKind` that specifies the type of objects
709 to reflect. Defaults to ``ObjectKind.TABLE``.
711 :param scope: a :class:`.ObjectScope` that specifies if options of
712 default, temporary or any tables should be reflected.
713 Defaults to ``ObjectScope.DEFAULT``.
715 :param \**kw: Additional keyword argument to pass to the dialect
716 specific implementation. See the documentation of the dialect
717 in use for more information.
719 :return: a dictionary where the keys are two-tuple schema,table-name
720 and the values are dictionaries with the table options.
721 The returned keys in each dict depend on the
722 dialect in use. Each one is prefixed with the dialect name.
723 The schema is ``None`` if no schema is provided.
725 .. versionadded:: 2.0
727 .. seealso:: :meth:`Inspector.get_table_options`
728 """
729 with self._operation_context() as conn:
730 res = self.dialect.get_multi_table_options(
731 conn,
732 schema=schema,
733 filter_names=filter_names,
734 kind=kind,
735 scope=scope,
736 info_cache=self.info_cache,
737 **kw,
738 )
739 return dict(res)
741 def get_view_names(
742 self, schema: Optional[str] = None, **kw: Any
743 ) -> List[str]:
744 r"""Return all non-materialized view names in `schema`.
746 :param schema: Optional, retrieve names from a non-default schema.
747 For special quoting, use :class:`.quoted_name`.
748 :param \**kw: Additional keyword argument to pass to the dialect
749 specific implementation. See the documentation of the dialect
750 in use for more information.
753 .. versionchanged:: 2.0 For those dialects that previously included
754 the names of materialized views in this list (currently PostgreSQL),
755 this method no longer returns the names of materialized views.
756 the :meth:`.Inspector.get_materialized_view_names` method should
757 be used instead.
759 .. seealso::
761 :meth:`.Inspector.get_materialized_view_names`
763 """
765 with self._operation_context() as conn:
766 return self.dialect.get_view_names(
767 conn, schema, info_cache=self.info_cache, **kw
768 )
770 def get_materialized_view_names(
771 self, schema: Optional[str] = None, **kw: Any
772 ) -> List[str]:
773 r"""Return all materialized view names in `schema`.
775 :param schema: Optional, retrieve names from a non-default schema.
776 For special quoting, use :class:`.quoted_name`.
777 :param \**kw: Additional keyword argument to pass to the dialect
778 specific implementation. See the documentation of the dialect
779 in use for more information.
781 .. versionadded:: 2.0
783 .. seealso::
785 :meth:`.Inspector.get_view_names`
787 """
789 with self._operation_context() as conn:
790 return self.dialect.get_materialized_view_names(
791 conn, schema, info_cache=self.info_cache, **kw
792 )
794 def get_sequence_names(
795 self, schema: Optional[str] = None, **kw: Any
796 ) -> List[str]:
797 r"""Return all sequence names in `schema`.
799 :param schema: Optional, retrieve names from a non-default schema.
800 For special quoting, use :class:`.quoted_name`.
801 :param \**kw: Additional keyword argument to pass to the dialect
802 specific implementation. See the documentation of the dialect
803 in use for more information.
805 """
807 with self._operation_context() as conn:
808 return self.dialect.get_sequence_names(
809 conn, schema, info_cache=self.info_cache, **kw
810 )
812 def get_view_definition(
813 self, view_name: str, schema: Optional[str] = None, **kw: Any
814 ) -> str:
815 r"""Return definition for the plain or materialized view called
816 ``view_name``.
818 :param view_name: Name of the view.
819 :param schema: Optional, retrieve names from a non-default schema.
820 For special quoting, use :class:`.quoted_name`.
821 :param \**kw: Additional keyword argument to pass to the dialect
822 specific implementation. See the documentation of the dialect
823 in use for more information.
825 """
827 with self._operation_context() as conn:
828 return self.dialect.get_view_definition(
829 conn, view_name, schema, info_cache=self.info_cache, **kw
830 )
832 def get_columns(
833 self, table_name: str, schema: Optional[str] = None, **kw: Any
834 ) -> List[ReflectedColumn]:
835 r"""Return information about columns in ``table_name``.
837 Given a string ``table_name`` and an optional string ``schema``,
838 return column information as a list of :class:`.ReflectedColumn`.
840 :param table_name: string name of the table. For special quoting,
841 use :class:`.quoted_name`.
843 :param schema: string schema name; if omitted, uses the default schema
844 of the database connection. For special quoting,
845 use :class:`.quoted_name`.
847 :param \**kw: Additional keyword argument to pass to the dialect
848 specific implementation. See the documentation of the dialect
849 in use for more information.
851 :return: list of dictionaries, each representing the definition of
852 a database column.
854 .. seealso:: :meth:`Inspector.get_multi_columns`.
856 """
858 with self._operation_context() as conn:
859 col_defs = self.dialect.get_columns(
860 conn, table_name, schema, info_cache=self.info_cache, **kw
861 )
862 if col_defs:
863 self._instantiate_types([col_defs])
864 return col_defs
866 def _instantiate_types(
867 self, data: Iterable[List[ReflectedColumn]]
868 ) -> None:
869 # make this easy and only return instances for coltype
870 for col_defs in data:
871 for col_def in col_defs:
872 coltype = col_def["type"]
873 if not isinstance(coltype, TypeEngine):
874 col_def["type"] = coltype()
876 def get_multi_columns(
877 self,
878 schema: Optional[str] = None,
879 filter_names: Optional[Sequence[str]] = None,
880 kind: ObjectKind = ObjectKind.TABLE,
881 scope: ObjectScope = ObjectScope.DEFAULT,
882 **kw: Any,
883 ) -> Dict[TableKey, List[ReflectedColumn]]:
884 r"""Return information about columns in all objects in the given
885 schema.
887 The objects can be filtered by passing the names to use to
888 ``filter_names``.
890 For each table the value is a list of :class:`.ReflectedColumn`.
892 :param schema: string schema name; if omitted, uses the default schema
893 of the database connection. For special quoting,
894 use :class:`.quoted_name`.
896 :param filter_names: optionally return information only for the
897 objects listed here.
899 :param kind: a :class:`.ObjectKind` that specifies the type of objects
900 to reflect. Defaults to ``ObjectKind.TABLE``.
902 :param scope: a :class:`.ObjectScope` that specifies if columns of
903 default, temporary or any tables should be reflected.
904 Defaults to ``ObjectScope.DEFAULT``.
906 :param \**kw: Additional keyword argument to pass to the dialect
907 specific implementation. See the documentation of the dialect
908 in use for more information.
910 :return: a dictionary where the keys are two-tuple schema,table-name
911 and the values are list of dictionaries, each representing the
912 definition of a database column.
913 The schema is ``None`` if no schema is provided.
915 .. versionadded:: 2.0
917 .. seealso:: :meth:`Inspector.get_columns`
918 """
920 with self._operation_context() as conn:
921 table_col_defs = dict(
922 self.dialect.get_multi_columns(
923 conn,
924 schema=schema,
925 filter_names=filter_names,
926 kind=kind,
927 scope=scope,
928 info_cache=self.info_cache,
929 **kw,
930 )
931 )
932 self._instantiate_types(table_col_defs.values())
933 return table_col_defs
935 def get_pk_constraint(
936 self, table_name: str, schema: Optional[str] = None, **kw: Any
937 ) -> ReflectedPrimaryKeyConstraint:
938 r"""Return information about primary key constraint in ``table_name``.
940 Given a string ``table_name``, and an optional string `schema`, return
941 primary key information as a :class:`.ReflectedPrimaryKeyConstraint`.
943 :param table_name: string name of the table. For special quoting,
944 use :class:`.quoted_name`.
946 :param schema: string schema name; if omitted, uses the default schema
947 of the database connection. For special quoting,
948 use :class:`.quoted_name`.
950 :param \**kw: Additional keyword argument to pass to the dialect
951 specific implementation. See the documentation of the dialect
952 in use for more information.
954 :return: a dictionary representing the definition of
955 a primary key constraint.
957 .. seealso:: :meth:`Inspector.get_multi_pk_constraint`
958 """
959 with self._operation_context() as conn:
960 return self.dialect.get_pk_constraint(
961 conn, table_name, schema, info_cache=self.info_cache, **kw
962 )
964 def get_multi_pk_constraint(
965 self,
966 schema: Optional[str] = None,
967 filter_names: Optional[Sequence[str]] = None,
968 kind: ObjectKind = ObjectKind.TABLE,
969 scope: ObjectScope = ObjectScope.DEFAULT,
970 **kw: Any,
971 ) -> Dict[TableKey, ReflectedPrimaryKeyConstraint]:
972 r"""Return information about primary key constraints in
973 all tables in the given schema.
975 The tables can be filtered by passing the names to use to
976 ``filter_names``.
978 For each table the value is a :class:`.ReflectedPrimaryKeyConstraint`.
980 :param schema: string schema name; if omitted, uses the default schema
981 of the database connection. For special quoting,
982 use :class:`.quoted_name`.
984 :param filter_names: optionally return information only for the
985 objects listed here.
987 :param kind: a :class:`.ObjectKind` that specifies the type of objects
988 to reflect. Defaults to ``ObjectKind.TABLE``.
990 :param scope: a :class:`.ObjectScope` that specifies if primary keys of
991 default, temporary or any tables should be reflected.
992 Defaults to ``ObjectScope.DEFAULT``.
994 :param \**kw: Additional keyword argument to pass to the dialect
995 specific implementation. See the documentation of the dialect
996 in use for more information.
998 :return: a dictionary where the keys are two-tuple schema,table-name
999 and the values are dictionaries, each representing the
1000 definition of a primary key constraint.
1001 The schema is ``None`` if no schema is provided.
1003 .. versionadded:: 2.0
1005 .. seealso:: :meth:`Inspector.get_pk_constraint`
1006 """
1007 with self._operation_context() as conn:
1008 return dict(
1009 self.dialect.get_multi_pk_constraint(
1010 conn,
1011 schema=schema,
1012 filter_names=filter_names,
1013 kind=kind,
1014 scope=scope,
1015 info_cache=self.info_cache,
1016 **kw,
1017 )
1018 )
1020 def get_foreign_keys(
1021 self, table_name: str, schema: Optional[str] = None, **kw: Any
1022 ) -> List[ReflectedForeignKeyConstraint]:
1023 r"""Return information about foreign_keys in ``table_name``.
1025 Given a string ``table_name``, and an optional string `schema`, return
1026 foreign key information as a list of
1027 :class:`.ReflectedForeignKeyConstraint`.
1029 :param table_name: string name of the table. For special quoting,
1030 use :class:`.quoted_name`.
1032 :param schema: string schema name; if omitted, uses the default schema
1033 of the database connection. For special quoting,
1034 use :class:`.quoted_name`.
1036 :param \**kw: Additional keyword argument to pass to the dialect
1037 specific implementation. See the documentation of the dialect
1038 in use for more information.
1040 :return: a list of dictionaries, each representing the
1041 a foreign key definition.
1043 .. seealso:: :meth:`Inspector.get_multi_foreign_keys`
1044 """
1046 with self._operation_context() as conn:
1047 return self.dialect.get_foreign_keys(
1048 conn, table_name, schema, info_cache=self.info_cache, **kw
1049 )
1051 def get_multi_foreign_keys(
1052 self,
1053 schema: Optional[str] = None,
1054 filter_names: Optional[Sequence[str]] = None,
1055 kind: ObjectKind = ObjectKind.TABLE,
1056 scope: ObjectScope = ObjectScope.DEFAULT,
1057 **kw: Any,
1058 ) -> Dict[TableKey, List[ReflectedForeignKeyConstraint]]:
1059 r"""Return information about foreign_keys in all tables
1060 in the given schema.
1062 The tables can be filtered by passing the names to use to
1063 ``filter_names``.
1065 For each table the value is a list of
1066 :class:`.ReflectedForeignKeyConstraint`.
1068 :param schema: string schema name; if omitted, uses the default schema
1069 of the database connection. For special quoting,
1070 use :class:`.quoted_name`.
1072 :param filter_names: optionally return information only for the
1073 objects listed here.
1075 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1076 to reflect. Defaults to ``ObjectKind.TABLE``.
1078 :param scope: a :class:`.ObjectScope` that specifies if foreign keys of
1079 default, temporary or any tables should be reflected.
1080 Defaults to ``ObjectScope.DEFAULT``.
1082 :param \**kw: Additional keyword argument to pass to the dialect
1083 specific implementation. See the documentation of the dialect
1084 in use for more information.
1086 :return: a dictionary where the keys are two-tuple schema,table-name
1087 and the values are list of dictionaries, each representing
1088 a foreign key definition.
1089 The schema is ``None`` if no schema is provided.
1091 .. versionadded:: 2.0
1093 .. seealso:: :meth:`Inspector.get_foreign_keys`
1094 """
1096 with self._operation_context() as conn:
1097 return dict(
1098 self.dialect.get_multi_foreign_keys(
1099 conn,
1100 schema=schema,
1101 filter_names=filter_names,
1102 kind=kind,
1103 scope=scope,
1104 info_cache=self.info_cache,
1105 **kw,
1106 )
1107 )
1109 def get_indexes(
1110 self, table_name: str, schema: Optional[str] = None, **kw: Any
1111 ) -> List[ReflectedIndex]:
1112 r"""Return information about indexes in ``table_name``.
1114 Given a string ``table_name`` and an optional string `schema`, return
1115 index information as a list of :class:`.ReflectedIndex`.
1117 :param table_name: string name of the table. For special quoting,
1118 use :class:`.quoted_name`.
1120 :param schema: string schema name; if omitted, uses the default schema
1121 of the database connection. For special quoting,
1122 use :class:`.quoted_name`.
1124 :param \**kw: Additional keyword argument to pass to the dialect
1125 specific implementation. See the documentation of the dialect
1126 in use for more information.
1128 :return: a list of dictionaries, each representing the
1129 definition of an index.
1131 .. seealso:: :meth:`Inspector.get_multi_indexes`
1132 """
1134 with self._operation_context() as conn:
1135 return self.dialect.get_indexes(
1136 conn, table_name, schema, info_cache=self.info_cache, **kw
1137 )
1139 def get_multi_indexes(
1140 self,
1141 schema: Optional[str] = None,
1142 filter_names: Optional[Sequence[str]] = None,
1143 kind: ObjectKind = ObjectKind.TABLE,
1144 scope: ObjectScope = ObjectScope.DEFAULT,
1145 **kw: Any,
1146 ) -> Dict[TableKey, List[ReflectedIndex]]:
1147 r"""Return information about indexes in in all objects
1148 in the given schema.
1150 The objects can be filtered by passing the names to use to
1151 ``filter_names``.
1153 For each table the value is a list of :class:`.ReflectedIndex`.
1155 :param schema: string schema name; if omitted, uses the default schema
1156 of the database connection. For special quoting,
1157 use :class:`.quoted_name`.
1159 :param filter_names: optionally return information only for the
1160 objects listed here.
1162 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1163 to reflect. Defaults to ``ObjectKind.TABLE``.
1165 :param scope: a :class:`.ObjectScope` that specifies if indexes of
1166 default, temporary or any tables should be reflected.
1167 Defaults to ``ObjectScope.DEFAULT``.
1169 :param \**kw: Additional keyword argument to pass to the dialect
1170 specific implementation. See the documentation of the dialect
1171 in use for more information.
1173 :return: a dictionary where the keys are two-tuple schema,table-name
1174 and the values are list of dictionaries, each representing the
1175 definition of an index.
1176 The schema is ``None`` if no schema is provided.
1178 .. versionadded:: 2.0
1180 .. seealso:: :meth:`Inspector.get_indexes`
1181 """
1183 with self._operation_context() as conn:
1184 return dict(
1185 self.dialect.get_multi_indexes(
1186 conn,
1187 schema=schema,
1188 filter_names=filter_names,
1189 kind=kind,
1190 scope=scope,
1191 info_cache=self.info_cache,
1192 **kw,
1193 )
1194 )
1196 def get_unique_constraints(
1197 self, table_name: str, schema: Optional[str] = None, **kw: Any
1198 ) -> List[ReflectedUniqueConstraint]:
1199 r"""Return information about unique constraints in ``table_name``.
1201 Given a string ``table_name`` and an optional string `schema`, return
1202 unique constraint information as a list of
1203 :class:`.ReflectedUniqueConstraint`.
1205 :param table_name: string name of the table. For special quoting,
1206 use :class:`.quoted_name`.
1208 :param schema: string schema name; if omitted, uses the default schema
1209 of the database connection. For special quoting,
1210 use :class:`.quoted_name`.
1212 :param \**kw: Additional keyword argument to pass to the dialect
1213 specific implementation. See the documentation of the dialect
1214 in use for more information.
1216 :return: a list of dictionaries, each representing the
1217 definition of an unique constraint.
1219 .. seealso:: :meth:`Inspector.get_multi_unique_constraints`
1220 """
1222 with self._operation_context() as conn:
1223 return self.dialect.get_unique_constraints(
1224 conn, table_name, schema, info_cache=self.info_cache, **kw
1225 )
1227 def get_multi_unique_constraints(
1228 self,
1229 schema: Optional[str] = None,
1230 filter_names: Optional[Sequence[str]] = None,
1231 kind: ObjectKind = ObjectKind.TABLE,
1232 scope: ObjectScope = ObjectScope.DEFAULT,
1233 **kw: Any,
1234 ) -> Dict[TableKey, List[ReflectedUniqueConstraint]]:
1235 r"""Return information about unique constraints in all tables
1236 in the given schema.
1238 The tables can be filtered by passing the names to use to
1239 ``filter_names``.
1241 For each table the value is a list of
1242 :class:`.ReflectedUniqueConstraint`.
1244 :param schema: string schema name; if omitted, uses the default schema
1245 of the database connection. For special quoting,
1246 use :class:`.quoted_name`.
1248 :param filter_names: optionally return information only for the
1249 objects listed here.
1251 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1252 to reflect. Defaults to ``ObjectKind.TABLE``.
1254 :param scope: a :class:`.ObjectScope` that specifies if constraints of
1255 default, temporary or any tables should be reflected.
1256 Defaults to ``ObjectScope.DEFAULT``.
1258 :param \**kw: Additional keyword argument to pass to the dialect
1259 specific implementation. See the documentation of the dialect
1260 in use for more information.
1262 :return: a dictionary where the keys are two-tuple schema,table-name
1263 and the values are list of dictionaries, each representing the
1264 definition of an unique constraint.
1265 The schema is ``None`` if no schema is provided.
1267 .. versionadded:: 2.0
1269 .. seealso:: :meth:`Inspector.get_unique_constraints`
1270 """
1272 with self._operation_context() as conn:
1273 return dict(
1274 self.dialect.get_multi_unique_constraints(
1275 conn,
1276 schema=schema,
1277 filter_names=filter_names,
1278 kind=kind,
1279 scope=scope,
1280 info_cache=self.info_cache,
1281 **kw,
1282 )
1283 )
1285 def get_table_comment(
1286 self, table_name: str, schema: Optional[str] = None, **kw: Any
1287 ) -> ReflectedTableComment:
1288 r"""Return information about the table comment for ``table_name``.
1290 Given a string ``table_name`` and an optional string ``schema``,
1291 return table comment information as a :class:`.ReflectedTableComment`.
1293 Raises ``NotImplementedError`` for a dialect that does not support
1294 comments.
1296 :param table_name: string name of the table. For special quoting,
1297 use :class:`.quoted_name`.
1299 :param schema: string schema name; if omitted, uses the default schema
1300 of the database connection. For special quoting,
1301 use :class:`.quoted_name`.
1303 :param \**kw: Additional keyword argument to pass to the dialect
1304 specific implementation. See the documentation of the dialect
1305 in use for more information.
1307 :return: a dictionary, with the table comment.
1309 .. versionadded:: 1.2
1311 .. seealso:: :meth:`Inspector.get_multi_table_comment`
1312 """
1314 with self._operation_context() as conn:
1315 return self.dialect.get_table_comment(
1316 conn, table_name, schema, info_cache=self.info_cache, **kw
1317 )
1319 def get_multi_table_comment(
1320 self,
1321 schema: Optional[str] = None,
1322 filter_names: Optional[Sequence[str]] = None,
1323 kind: ObjectKind = ObjectKind.TABLE,
1324 scope: ObjectScope = ObjectScope.DEFAULT,
1325 **kw: Any,
1326 ) -> Dict[TableKey, ReflectedTableComment]:
1327 r"""Return information about the table comment in all objects
1328 in the given schema.
1330 The objects can be filtered by passing the names to use to
1331 ``filter_names``.
1333 For each table the value is a :class:`.ReflectedTableComment`.
1335 Raises ``NotImplementedError`` for a dialect that does not support
1336 comments.
1338 :param schema: string schema name; if omitted, uses the default schema
1339 of the database connection. For special quoting,
1340 use :class:`.quoted_name`.
1342 :param filter_names: optionally return information only for the
1343 objects listed here.
1345 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1346 to reflect. Defaults to ``ObjectKind.TABLE``.
1348 :param scope: a :class:`.ObjectScope` that specifies if comments of
1349 default, temporary or any tables should be reflected.
1350 Defaults to ``ObjectScope.DEFAULT``.
1352 :param \**kw: Additional keyword argument to pass to the dialect
1353 specific implementation. See the documentation of the dialect
1354 in use for more information.
1356 :return: a dictionary where the keys are two-tuple schema,table-name
1357 and the values are dictionaries, representing the
1358 table comments.
1359 The schema is ``None`` if no schema is provided.
1361 .. versionadded:: 2.0
1363 .. seealso:: :meth:`Inspector.get_table_comment`
1364 """
1366 with self._operation_context() as conn:
1367 return dict(
1368 self.dialect.get_multi_table_comment(
1369 conn,
1370 schema=schema,
1371 filter_names=filter_names,
1372 kind=kind,
1373 scope=scope,
1374 info_cache=self.info_cache,
1375 **kw,
1376 )
1377 )
1379 def get_check_constraints(
1380 self, table_name: str, schema: Optional[str] = None, **kw: Any
1381 ) -> List[ReflectedCheckConstraint]:
1382 r"""Return information about check constraints in ``table_name``.
1384 Given a string ``table_name`` and an optional string `schema`, return
1385 check constraint information as a list of
1386 :class:`.ReflectedCheckConstraint`.
1388 :param table_name: string name of the table. For special quoting,
1389 use :class:`.quoted_name`.
1391 :param schema: string schema name; if omitted, uses the default schema
1392 of the database connection. For special quoting,
1393 use :class:`.quoted_name`.
1395 :param \**kw: Additional keyword argument to pass to the dialect
1396 specific implementation. See the documentation of the dialect
1397 in use for more information.
1399 :return: a list of dictionaries, each representing the
1400 definition of a check constraints.
1402 .. seealso:: :meth:`Inspector.get_multi_check_constraints`
1403 """
1405 with self._operation_context() as conn:
1406 return self.dialect.get_check_constraints(
1407 conn, table_name, schema, info_cache=self.info_cache, **kw
1408 )
1410 def get_multi_check_constraints(
1411 self,
1412 schema: Optional[str] = None,
1413 filter_names: Optional[Sequence[str]] = None,
1414 kind: ObjectKind = ObjectKind.TABLE,
1415 scope: ObjectScope = ObjectScope.DEFAULT,
1416 **kw: Any,
1417 ) -> Dict[TableKey, List[ReflectedCheckConstraint]]:
1418 r"""Return information about check constraints in all tables
1419 in the given schema.
1421 The tables can be filtered by passing the names to use to
1422 ``filter_names``.
1424 For each table the value is a list of
1425 :class:`.ReflectedCheckConstraint`.
1427 :param schema: string schema name; if omitted, uses the default schema
1428 of the database connection. For special quoting,
1429 use :class:`.quoted_name`.
1431 :param filter_names: optionally return information only for the
1432 objects listed here.
1434 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1435 to reflect. Defaults to ``ObjectKind.TABLE``.
1437 :param scope: a :class:`.ObjectScope` that specifies if constraints of
1438 default, temporary or any tables should be reflected.
1439 Defaults to ``ObjectScope.DEFAULT``.
1441 :param \**kw: Additional keyword argument to pass to the dialect
1442 specific implementation. See the documentation of the dialect
1443 in use for more information.
1445 :return: a dictionary where the keys are two-tuple schema,table-name
1446 and the values are list of dictionaries, each representing the
1447 definition of a check constraints.
1448 The schema is ``None`` if no schema is provided.
1450 .. versionadded:: 2.0
1452 .. seealso:: :meth:`Inspector.get_check_constraints`
1453 """
1455 with self._operation_context() as conn:
1456 return dict(
1457 self.dialect.get_multi_check_constraints(
1458 conn,
1459 schema=schema,
1460 filter_names=filter_names,
1461 kind=kind,
1462 scope=scope,
1463 info_cache=self.info_cache,
1464 **kw,
1465 )
1466 )
1468 def reflect_table(
1469 self,
1470 table: sa_schema.Table,
1471 include_columns: Optional[Collection[str]],
1472 exclude_columns: Collection[str] = (),
1473 resolve_fks: bool = True,
1474 _extend_on: Optional[Set[sa_schema.Table]] = None,
1475 _reflect_info: Optional[_ReflectionInfo] = None,
1476 ) -> None:
1477 """Given a :class:`_schema.Table` object, load its internal
1478 constructs based on introspection.
1480 This is the underlying method used by most dialects to produce
1481 table reflection. Direct usage is like::
1483 from sqlalchemy import create_engine, MetaData, Table
1484 from sqlalchemy import inspect
1486 engine = create_engine('...')
1487 meta = MetaData()
1488 user_table = Table('user', meta)
1489 insp = inspect(engine)
1490 insp.reflect_table(user_table, None)
1492 .. versionchanged:: 1.4 Renamed from ``reflecttable`` to
1493 ``reflect_table``
1495 :param table: a :class:`~sqlalchemy.schema.Table` instance.
1496 :param include_columns: a list of string column names to include
1497 in the reflection process. If ``None``, all columns are reflected.
1499 """
1501 if _extend_on is not None:
1502 if table in _extend_on:
1503 return
1504 else:
1505 _extend_on.add(table)
1507 dialect = self.bind.dialect
1509 with self._operation_context() as conn:
1510 schema = conn.schema_for_object(table)
1512 table_name = table.name
1514 # get table-level arguments that are specifically
1515 # intended for reflection, e.g. oracle_resolve_synonyms.
1516 # these are unconditionally passed to related Table
1517 # objects
1518 reflection_options = {
1519 k: table.dialect_kwargs.get(k)
1520 for k in dialect.reflection_options
1521 if k in table.dialect_kwargs
1522 }
1524 table_key = (schema, table_name)
1525 if _reflect_info is None or table_key not in _reflect_info.columns:
1526 _reflect_info = self._get_reflection_info(
1527 schema,
1528 filter_names=[table_name],
1529 kind=ObjectKind.ANY,
1530 scope=ObjectScope.ANY,
1531 _reflect_info=_reflect_info,
1532 **table.dialect_kwargs,
1533 )
1534 if table_key in _reflect_info.unreflectable:
1535 raise _reflect_info.unreflectable[table_key]
1537 if table_key not in _reflect_info.columns:
1538 raise exc.NoSuchTableError(table_name)
1540 # reflect table options, like mysql_engine
1541 if _reflect_info.table_options:
1542 tbl_opts = _reflect_info.table_options.get(table_key)
1543 if tbl_opts:
1544 # add additional kwargs to the Table if the dialect
1545 # returned them
1546 table._validate_dialect_kwargs(tbl_opts)
1548 found_table = False
1549 cols_by_orig_name: Dict[str, sa_schema.Column[Any]] = {}
1551 for col_d in _reflect_info.columns[table_key]:
1552 found_table = True
1554 self._reflect_column(
1555 table,
1556 col_d,
1557 include_columns,
1558 exclude_columns,
1559 cols_by_orig_name,
1560 )
1562 # NOTE: support tables/views with no columns
1563 if not found_table and not self.has_table(table_name, schema):
1564 raise exc.NoSuchTableError(table_name)
1566 self._reflect_pk(
1567 _reflect_info, table_key, table, cols_by_orig_name, exclude_columns
1568 )
1570 self._reflect_fk(
1571 _reflect_info,
1572 table_key,
1573 table,
1574 cols_by_orig_name,
1575 include_columns,
1576 exclude_columns,
1577 resolve_fks,
1578 _extend_on,
1579 reflection_options,
1580 )
1582 self._reflect_indexes(
1583 _reflect_info,
1584 table_key,
1585 table,
1586 cols_by_orig_name,
1587 include_columns,
1588 exclude_columns,
1589 reflection_options,
1590 )
1592 self._reflect_unique_constraints(
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_check_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_table_comment(
1613 _reflect_info,
1614 table_key,
1615 table,
1616 reflection_options,
1617 )
1619 def _reflect_column(
1620 self,
1621 table: sa_schema.Table,
1622 col_d: ReflectedColumn,
1623 include_columns: Optional[Collection[str]],
1624 exclude_columns: Collection[str],
1625 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1626 ) -> None:
1627 orig_name = col_d["name"]
1629 table.metadata.dispatch.column_reflect(self, table, col_d)
1630 table.dispatch.column_reflect(self, table, col_d)
1632 # fetch name again as column_reflect is allowed to
1633 # change it
1634 name = col_d["name"]
1635 if (include_columns and name not in include_columns) or (
1636 exclude_columns and name in exclude_columns
1637 ):
1638 return
1640 coltype = col_d["type"]
1642 col_kw = {
1643 k: col_d[k] # type: ignore[literal-required]
1644 for k in [
1645 "nullable",
1646 "autoincrement",
1647 "quote",
1648 "info",
1649 "key",
1650 "comment",
1651 ]
1652 if k in col_d
1653 }
1655 if "dialect_options" in col_d:
1656 col_kw.update(col_d["dialect_options"])
1658 colargs = []
1659 default: Any
1660 if col_d.get("default") is not None:
1661 default_text = col_d["default"]
1662 assert default_text is not None
1663 if isinstance(default_text, TextClause):
1664 default = sa_schema.DefaultClause(
1665 default_text, _reflected=True
1666 )
1667 elif not isinstance(default_text, sa_schema.FetchedValue):
1668 default = sa_schema.DefaultClause(
1669 sql.text(default_text), _reflected=True
1670 )
1671 else:
1672 default = default_text
1673 colargs.append(default)
1675 if "computed" in col_d:
1676 computed = sa_schema.Computed(**col_d["computed"])
1677 colargs.append(computed)
1679 if "identity" in col_d:
1680 identity = sa_schema.Identity(**col_d["identity"])
1681 colargs.append(identity)
1683 cols_by_orig_name[orig_name] = col = sa_schema.Column(
1684 name, coltype, *colargs, **col_kw
1685 )
1687 if col.key in table.primary_key:
1688 col.primary_key = True
1689 table.append_column(col, replace_existing=True)
1691 def _reflect_pk(
1692 self,
1693 _reflect_info: _ReflectionInfo,
1694 table_key: TableKey,
1695 table: sa_schema.Table,
1696 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1697 exclude_columns: Collection[str],
1698 ) -> None:
1699 pk_cons = _reflect_info.pk_constraint.get(table_key)
1700 if pk_cons:
1701 pk_cols = [
1702 cols_by_orig_name[pk]
1703 for pk in pk_cons["constrained_columns"]
1704 if pk in cols_by_orig_name and pk not in exclude_columns
1705 ]
1707 # update pk constraint name and comment
1708 table.primary_key.name = pk_cons.get("name")
1709 table.primary_key.comment = pk_cons.get("comment", None)
1711 # tell the PKConstraint to re-initialize
1712 # its column collection
1713 table.primary_key._reload(pk_cols)
1715 def _reflect_fk(
1716 self,
1717 _reflect_info: _ReflectionInfo,
1718 table_key: TableKey,
1719 table: sa_schema.Table,
1720 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1721 include_columns: Optional[Collection[str]],
1722 exclude_columns: Collection[str],
1723 resolve_fks: bool,
1724 _extend_on: Optional[Set[sa_schema.Table]],
1725 reflection_options: Dict[str, Any],
1726 ) -> None:
1727 fkeys = _reflect_info.foreign_keys.get(table_key, [])
1728 for fkey_d in fkeys:
1729 conname = fkey_d["name"]
1730 # look for columns by orig name in cols_by_orig_name,
1731 # but support columns that are in-Python only as fallback
1732 constrained_columns = [
1733 cols_by_orig_name[c].key if c in cols_by_orig_name else c
1734 for c in fkey_d["constrained_columns"]
1735 ]
1737 if (
1738 exclude_columns
1739 and set(constrained_columns).intersection(exclude_columns)
1740 or (
1741 include_columns
1742 and set(constrained_columns).difference(include_columns)
1743 )
1744 ):
1745 continue
1747 referred_schema = fkey_d["referred_schema"]
1748 referred_table = fkey_d["referred_table"]
1749 referred_columns = fkey_d["referred_columns"]
1750 refspec = []
1751 if referred_schema is not None:
1752 if resolve_fks:
1753 sa_schema.Table(
1754 referred_table,
1755 table.metadata,
1756 schema=referred_schema,
1757 autoload_with=self.bind,
1758 _extend_on=_extend_on,
1759 _reflect_info=_reflect_info,
1760 **reflection_options,
1761 )
1762 for column in referred_columns:
1763 refspec.append(
1764 ".".join([referred_schema, referred_table, column])
1765 )
1766 else:
1767 if resolve_fks:
1768 sa_schema.Table(
1769 referred_table,
1770 table.metadata,
1771 autoload_with=self.bind,
1772 schema=sa_schema.BLANK_SCHEMA,
1773 _extend_on=_extend_on,
1774 _reflect_info=_reflect_info,
1775 **reflection_options,
1776 )
1777 for column in referred_columns:
1778 refspec.append(".".join([referred_table, column]))
1779 if "options" in fkey_d:
1780 options = fkey_d["options"]
1781 else:
1782 options = {}
1784 try:
1785 table.append_constraint(
1786 sa_schema.ForeignKeyConstraint(
1787 constrained_columns,
1788 refspec,
1789 conname,
1790 link_to_name=True,
1791 comment=fkey_d.get("comment"),
1792 **options,
1793 )
1794 )
1795 except exc.ConstraintColumnNotFoundError:
1796 util.warn(
1797 f"On reflected table {table.name}, skipping reflection of "
1798 "foreign key constraint "
1799 f"{conname}; one or more subject columns within "
1800 f"name(s) {', '.join(constrained_columns)} are not "
1801 "present in the table"
1802 )
1804 _index_sort_exprs = {
1805 "asc": operators.asc_op,
1806 "desc": operators.desc_op,
1807 "nulls_first": operators.nulls_first_op,
1808 "nulls_last": operators.nulls_last_op,
1809 }
1811 def _reflect_indexes(
1812 self,
1813 _reflect_info: _ReflectionInfo,
1814 table_key: TableKey,
1815 table: sa_schema.Table,
1816 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1817 include_columns: Optional[Collection[str]],
1818 exclude_columns: Collection[str],
1819 reflection_options: Dict[str, Any],
1820 ) -> None:
1821 # Indexes
1822 indexes = _reflect_info.indexes.get(table_key, [])
1823 for index_d in indexes:
1824 name = index_d["name"]
1825 columns = index_d["column_names"]
1826 expressions = index_d.get("expressions")
1827 column_sorting = index_d.get("column_sorting", {})
1828 unique = index_d["unique"]
1829 flavor = index_d.get("type", "index")
1830 dialect_options = index_d.get("dialect_options", {})
1832 duplicates = index_d.get("duplicates_constraint")
1833 if include_columns and not set(columns).issubset(include_columns):
1834 continue
1835 if duplicates:
1836 continue
1837 # look for columns by orig name in cols_by_orig_name,
1838 # but support columns that are in-Python only as fallback
1839 idx_element: Any
1840 idx_elements = []
1841 for index, c in enumerate(columns):
1842 if c is None:
1843 if not expressions:
1844 util.warn(
1845 f"Skipping {flavor} {name!r} because key "
1846 f"{index + 1} reflected as None but no "
1847 "'expressions' were returned"
1848 )
1849 break
1850 idx_element = sql.text(expressions[index])
1851 else:
1852 try:
1853 if c in cols_by_orig_name:
1854 idx_element = cols_by_orig_name[c]
1855 else:
1856 idx_element = table.c[c]
1857 except KeyError:
1858 util.warn(
1859 f"{flavor} key {c!r} was not located in "
1860 f"columns for table {table.name!r}"
1861 )
1862 continue
1863 for option in column_sorting.get(c, ()):
1864 if option in self._index_sort_exprs:
1865 op = self._index_sort_exprs[option]
1866 idx_element = op(idx_element)
1867 idx_elements.append(idx_element)
1868 else:
1869 sa_schema.Index(
1870 name,
1871 *idx_elements,
1872 _table=table,
1873 unique=unique,
1874 **dialect_options,
1875 )
1877 def _reflect_unique_constraints(
1878 self,
1879 _reflect_info: _ReflectionInfo,
1880 table_key: TableKey,
1881 table: sa_schema.Table,
1882 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1883 include_columns: Optional[Collection[str]],
1884 exclude_columns: Collection[str],
1885 reflection_options: Dict[str, Any],
1886 ) -> None:
1887 constraints = _reflect_info.unique_constraints.get(table_key, [])
1888 # Unique Constraints
1889 for const_d in constraints:
1890 conname = const_d["name"]
1891 columns = const_d["column_names"]
1892 comment = const_d.get("comment")
1893 duplicates = const_d.get("duplicates_index")
1894 dialect_options = const_d.get("dialect_options", {})
1895 if include_columns and not set(columns).issubset(include_columns):
1896 continue
1897 if duplicates:
1898 continue
1899 # look for columns by orig name in cols_by_orig_name,
1900 # but support columns that are in-Python only as fallback
1901 constrained_cols = []
1902 for c in columns:
1903 try:
1904 constrained_col = (
1905 cols_by_orig_name[c]
1906 if c in cols_by_orig_name
1907 else table.c[c]
1908 )
1909 except KeyError:
1910 util.warn(
1911 "unique constraint key '%s' was not located in "
1912 "columns for table '%s'" % (c, table.name)
1913 )
1914 else:
1915 constrained_cols.append(constrained_col)
1916 table.append_constraint(
1917 sa_schema.UniqueConstraint(
1918 *constrained_cols,
1919 name=conname,
1920 comment=comment,
1921 **dialect_options,
1922 )
1923 )
1925 def _reflect_check_constraints(
1926 self,
1927 _reflect_info: _ReflectionInfo,
1928 table_key: TableKey,
1929 table: sa_schema.Table,
1930 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1931 include_columns: Optional[Collection[str]],
1932 exclude_columns: Collection[str],
1933 reflection_options: Dict[str, Any],
1934 ) -> None:
1935 constraints = _reflect_info.check_constraints.get(table_key, [])
1936 for const_d in constraints:
1937 table.append_constraint(sa_schema.CheckConstraint(**const_d))
1939 def _reflect_table_comment(
1940 self,
1941 _reflect_info: _ReflectionInfo,
1942 table_key: TableKey,
1943 table: sa_schema.Table,
1944 reflection_options: Dict[str, Any],
1945 ) -> None:
1946 comment_dict = _reflect_info.table_comment.get(table_key)
1947 if comment_dict:
1948 table.comment = comment_dict["text"]
1950 def _get_reflection_info(
1951 self,
1952 schema: Optional[str] = None,
1953 filter_names: Optional[Collection[str]] = None,
1954 available: Optional[Collection[str]] = None,
1955 _reflect_info: Optional[_ReflectionInfo] = None,
1956 **kw: Any,
1957 ) -> _ReflectionInfo:
1958 kw["schema"] = schema
1960 if filter_names and available and len(filter_names) > 100:
1961 fraction = len(filter_names) / len(available)
1962 else:
1963 fraction = None
1965 unreflectable: Dict[TableKey, exc.UnreflectableTableError]
1966 kw["unreflectable"] = unreflectable = {}
1968 has_result: bool = True
1970 def run(
1971 meth: Any,
1972 *,
1973 optional: bool = False,
1974 check_filter_names_from_meth: bool = False,
1975 ) -> Any:
1976 nonlocal has_result
1977 # simple heuristic to improve reflection performance if a
1978 # dialect implements multi_reflection:
1979 # if more than 50% of the tables in the db are in filter_names
1980 # load all the tables, since it's most likely faster to avoid
1981 # a filter on that many tables.
1982 if (
1983 fraction is None
1984 or fraction <= 0.5
1985 or not self.dialect._overrides_default(meth.__name__)
1986 ):
1987 _fn = filter_names
1988 else:
1989 _fn = None
1990 try:
1991 if has_result:
1992 res = meth(filter_names=_fn, **kw)
1993 if check_filter_names_from_meth and not res:
1994 # method returned no result data.
1995 # skip any future call methods
1996 has_result = False
1997 else:
1998 res = {}
1999 except NotImplementedError:
2000 if not optional:
2001 raise
2002 res = {}
2003 return res
2005 info = _ReflectionInfo(
2006 columns=run(
2007 self.get_multi_columns, check_filter_names_from_meth=True
2008 ),
2009 pk_constraint=run(self.get_multi_pk_constraint),
2010 foreign_keys=run(self.get_multi_foreign_keys),
2011 indexes=run(self.get_multi_indexes),
2012 unique_constraints=run(
2013 self.get_multi_unique_constraints, optional=True
2014 ),
2015 table_comment=run(self.get_multi_table_comment, optional=True),
2016 check_constraints=run(
2017 self.get_multi_check_constraints, optional=True
2018 ),
2019 table_options=run(self.get_multi_table_options, optional=True),
2020 unreflectable=unreflectable,
2021 )
2022 if _reflect_info:
2023 _reflect_info.update(info)
2024 return _reflect_info
2025 else:
2026 return info
2029@final
2030class ReflectionDefaults:
2031 """provides blank default values for reflection methods."""
2033 @classmethod
2034 def columns(cls) -> List[ReflectedColumn]:
2035 return []
2037 @classmethod
2038 def pk_constraint(cls) -> ReflectedPrimaryKeyConstraint:
2039 return {
2040 "name": None,
2041 "constrained_columns": [],
2042 }
2044 @classmethod
2045 def foreign_keys(cls) -> List[ReflectedForeignKeyConstraint]:
2046 return []
2048 @classmethod
2049 def indexes(cls) -> List[ReflectedIndex]:
2050 return []
2052 @classmethod
2053 def unique_constraints(cls) -> List[ReflectedUniqueConstraint]:
2054 return []
2056 @classmethod
2057 def check_constraints(cls) -> List[ReflectedCheckConstraint]:
2058 return []
2060 @classmethod
2061 def table_options(cls) -> Dict[str, Any]:
2062 return {}
2064 @classmethod
2065 def table_comment(cls) -> ReflectedTableComment:
2066 return {"text": None}
2069@dataclass
2070class _ReflectionInfo:
2071 columns: Dict[TableKey, List[ReflectedColumn]]
2072 pk_constraint: Dict[TableKey, Optional[ReflectedPrimaryKeyConstraint]]
2073 foreign_keys: Dict[TableKey, List[ReflectedForeignKeyConstraint]]
2074 indexes: Dict[TableKey, List[ReflectedIndex]]
2075 # optionals
2076 unique_constraints: Dict[TableKey, List[ReflectedUniqueConstraint]]
2077 table_comment: Dict[TableKey, Optional[ReflectedTableComment]]
2078 check_constraints: Dict[TableKey, List[ReflectedCheckConstraint]]
2079 table_options: Dict[TableKey, Dict[str, Any]]
2080 unreflectable: Dict[TableKey, exc.UnreflectableTableError]
2082 def update(self, other: _ReflectionInfo) -> None:
2083 for k, v in self.__dict__.items():
2084 ov = getattr(other, k)
2085 if ov is not None:
2086 if v is None:
2087 setattr(self, k, ov)
2088 else:
2089 v.update(ov)