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 quoted_name
60from ..sql.elements import TextClause
61from ..sql.type_api import TypeEngine
62from ..sql.visitors import InternalTraversal
63from ..util import topological
65if TYPE_CHECKING:
66 from .interfaces import Dialect
67 from .interfaces import ReflectedCheckConstraint
68 from .interfaces import ReflectedColumn
69 from .interfaces import ReflectedForeignKeyConstraint
70 from .interfaces import ReflectedIndex
71 from .interfaces import ReflectedPrimaryKeyConstraint
72 from .interfaces import ReflectedTableComment
73 from .interfaces import ReflectedUniqueConstraint
74 from .interfaces import TableKey
76_R = TypeVar("_R")
79@util.decorator
80def cache(
81 fn: Callable[..., _R],
82 self: Dialect,
83 con: Connection,
84 *args: Any,
85 **kw: Any,
86) -> _R:
87 info_cache = kw.get("info_cache", None)
88 if info_cache is None:
89 return fn(self, con, *args, **kw)
90 exclude = {"info_cache", "unreflectable"}
91 key = (
92 fn.__name__,
93 tuple(
94 (str(a), a.quote) if isinstance(a, quoted_name) else a
95 for a in args
96 if isinstance(a, str)
97 ),
98 tuple(
99 (k, (str(v), v.quote) if isinstance(v, quoted_name) else v)
100 for k, v in kw.items()
101 if k not in exclude
102 ),
103 )
104 ret: _R = info_cache.get(key)
105 if ret is None:
106 ret = fn(self, con, *args, **kw)
107 info_cache[key] = ret
108 return ret
111def flexi_cache(
112 *traverse_args: Tuple[str, InternalTraversal]
113) -> Callable[[Callable[..., _R]], Callable[..., _R]]:
114 @util.decorator
115 def go(
116 fn: Callable[..., _R],
117 self: Dialect,
118 con: Connection,
119 *args: Any,
120 **kw: Any,
121 ) -> _R:
122 info_cache = kw.get("info_cache", None)
123 if info_cache is None:
124 return fn(self, con, *args, **kw)
125 key = _ad_hoc_cache_key_from_args((fn.__name__,), traverse_args, args)
126 ret: _R = info_cache.get(key)
127 if ret is None:
128 ret = fn(self, con, *args, **kw)
129 info_cache[key] = ret
130 return ret
132 return go
135@unique
136class ObjectKind(Flag):
137 """Enumerator that indicates which kind of object to return when calling
138 the ``get_multi`` methods.
140 This is a Flag enum, so custom combinations can be passed. For example,
141 to reflect tables and plain views ``ObjectKind.TABLE | ObjectKind.VIEW``
142 may be used.
144 .. note::
145 Not all dialect may support all kind of object. If a dialect does
146 not support a particular object an empty dict is returned.
147 In case a dialect supports an object, but the requested method
148 is not applicable for the specified kind the default value
149 will be returned for each reflected object. For example reflecting
150 check constraints of view return a dict with all the views with
151 empty lists as values.
152 """
154 TABLE = auto()
155 "Reflect table objects"
156 VIEW = auto()
157 "Reflect plain view objects"
158 MATERIALIZED_VIEW = auto()
159 "Reflect materialized view object"
161 ANY_VIEW = VIEW | MATERIALIZED_VIEW
162 "Reflect any kind of view objects"
163 ANY = TABLE | VIEW | MATERIALIZED_VIEW
164 "Reflect all type of objects"
167@unique
168class ObjectScope(Flag):
169 """Enumerator that indicates which scope to use when calling
170 the ``get_multi`` methods.
171 """
173 DEFAULT = auto()
174 "Include default scope"
175 TEMPORARY = auto()
176 "Include only temp scope"
177 ANY = DEFAULT | TEMPORARY
178 "Include both default and temp scope"
181@inspection._self_inspects
182class Inspector(inspection.Inspectable["Inspector"]):
183 """Performs database schema inspection.
185 The Inspector acts as a proxy to the reflection methods of the
186 :class:`~sqlalchemy.engine.interfaces.Dialect`, providing a
187 consistent interface as well as caching support for previously
188 fetched metadata.
190 A :class:`_reflection.Inspector` object is usually created via the
191 :func:`_sa.inspect` function, which may be passed an
192 :class:`_engine.Engine`
193 or a :class:`_engine.Connection`::
195 from sqlalchemy import inspect, create_engine
196 engine = create_engine('...')
197 insp = inspect(engine)
199 Where above, the :class:`~sqlalchemy.engine.interfaces.Dialect` associated
200 with the engine may opt to return an :class:`_reflection.Inspector`
201 subclass that
202 provides additional methods specific to the dialect's target database.
204 """
206 bind: Union[Engine, Connection]
207 engine: Engine
208 _op_context_requires_connect: bool
209 dialect: Dialect
210 info_cache: Dict[Any, Any]
212 @util.deprecated(
213 "1.4",
214 "The __init__() method on :class:`_reflection.Inspector` "
215 "is deprecated and "
216 "will be removed in a future release. Please use the "
217 ":func:`.sqlalchemy.inspect` "
218 "function on an :class:`_engine.Engine` or "
219 ":class:`_engine.Connection` "
220 "in order to "
221 "acquire an :class:`_reflection.Inspector`.",
222 )
223 def __init__(self, bind: Union[Engine, Connection]):
224 """Initialize a new :class:`_reflection.Inspector`.
226 :param bind: a :class:`~sqlalchemy.engine.Connection`,
227 which is typically an instance of
228 :class:`~sqlalchemy.engine.Engine` or
229 :class:`~sqlalchemy.engine.Connection`.
231 For a dialect-specific instance of :class:`_reflection.Inspector`, see
232 :meth:`_reflection.Inspector.from_engine`
234 """
235 self._init_legacy(bind)
237 @classmethod
238 def _construct(
239 cls, init: Callable[..., Any], bind: Union[Engine, Connection]
240 ) -> Inspector:
241 if hasattr(bind.dialect, "inspector"):
242 cls = bind.dialect.inspector
244 self = cls.__new__(cls)
245 init(self, bind)
246 return self
248 def _init_legacy(self, bind: Union[Engine, Connection]) -> None:
249 if hasattr(bind, "exec_driver_sql"):
250 self._init_connection(bind) # type: ignore[arg-type]
251 else:
252 self._init_engine(bind)
254 def _init_engine(self, engine: Engine) -> None:
255 self.bind = self.engine = engine
256 engine.connect().close()
257 self._op_context_requires_connect = True
258 self.dialect = self.engine.dialect
259 self.info_cache = {}
261 def _init_connection(self, connection: Connection) -> None:
262 self.bind = connection
263 self.engine = connection.engine
264 self._op_context_requires_connect = False
265 self.dialect = self.engine.dialect
266 self.info_cache = {}
268 def clear_cache(self) -> None:
269 """reset the cache for this :class:`.Inspector`.
271 Inspection methods that have data cached will emit SQL queries
272 when next called to get new data.
274 .. versionadded:: 2.0
276 """
277 self.info_cache.clear()
279 @classmethod
280 @util.deprecated(
281 "1.4",
282 "The from_engine() method on :class:`_reflection.Inspector` "
283 "is deprecated and "
284 "will be removed in a future release. Please use the "
285 ":func:`.sqlalchemy.inspect` "
286 "function on an :class:`_engine.Engine` or "
287 ":class:`_engine.Connection` "
288 "in order to "
289 "acquire an :class:`_reflection.Inspector`.",
290 )
291 def from_engine(cls, bind: Engine) -> Inspector:
292 """Construct a new dialect-specific Inspector object from the given
293 engine or connection.
295 :param bind: a :class:`~sqlalchemy.engine.Connection`
296 or :class:`~sqlalchemy.engine.Engine`.
298 This method differs from direct a direct constructor call of
299 :class:`_reflection.Inspector` in that the
300 :class:`~sqlalchemy.engine.interfaces.Dialect` is given a chance to
301 provide a dialect-specific :class:`_reflection.Inspector` instance,
302 which may
303 provide additional methods.
305 See the example at :class:`_reflection.Inspector`.
307 """
308 return cls._construct(cls._init_legacy, bind)
310 @inspection._inspects(Engine)
311 def _engine_insp(bind: Engine) -> Inspector: # type: ignore[misc]
312 return Inspector._construct(Inspector._init_engine, bind)
314 @inspection._inspects(Connection)
315 def _connection_insp(bind: Connection) -> Inspector: # type: ignore[misc]
316 return Inspector._construct(Inspector._init_connection, bind)
318 @contextlib.contextmanager
319 def _operation_context(self) -> Generator[Connection, None, None]:
320 """Return a context that optimizes for multiple operations on a single
321 transaction.
323 This essentially allows connect()/close() to be called if we detected
324 that we're against an :class:`_engine.Engine` and not a
325 :class:`_engine.Connection`.
327 """
328 conn: Connection
329 if self._op_context_requires_connect:
330 conn = self.bind.connect() # type: ignore[union-attr]
331 else:
332 conn = self.bind # type: ignore[assignment]
333 try:
334 yield conn
335 finally:
336 if self._op_context_requires_connect:
337 conn.close()
339 @contextlib.contextmanager
340 def _inspection_context(self) -> Generator[Inspector, None, None]:
341 """Return an :class:`_reflection.Inspector`
342 from this one that will run all
343 operations on a single connection.
345 """
347 with self._operation_context() as conn:
348 sub_insp = self._construct(self.__class__._init_connection, conn)
349 sub_insp.info_cache = self.info_cache
350 yield sub_insp
352 @property
353 def default_schema_name(self) -> Optional[str]:
354 """Return the default schema name presented by the dialect
355 for the current engine's database user.
357 E.g. this is typically ``public`` for PostgreSQL and ``dbo``
358 for SQL Server.
360 """
361 return self.dialect.default_schema_name
363 def get_schema_names(self, **kw: Any) -> List[str]:
364 r"""Return all schema names.
366 :param \**kw: Additional keyword argument to pass to the dialect
367 specific implementation. See the documentation of the dialect
368 in use for more information.
369 """
371 with self._operation_context() as conn:
372 return self.dialect.get_schema_names(
373 conn, info_cache=self.info_cache, **kw
374 )
376 def get_table_names(
377 self, schema: Optional[str] = None, **kw: Any
378 ) -> List[str]:
379 r"""Return all table names within a particular schema.
381 The names are expected to be real tables only, not views.
382 Views are instead returned using the
383 :meth:`_reflection.Inspector.get_view_names` and/or
384 :meth:`_reflection.Inspector.get_materialized_view_names`
385 methods.
387 :param schema: Schema name. If ``schema`` is left at ``None``, the
388 database's default schema is
389 used, else the named schema is searched. If the database does not
390 support named schemas, behavior is undefined if ``schema`` is not
391 passed as ``None``. For special quoting, use :class:`.quoted_name`.
392 :param \**kw: Additional keyword argument to pass to the dialect
393 specific implementation. See the documentation of the dialect
394 in use for more information.
396 .. seealso::
398 :meth:`_reflection.Inspector.get_sorted_table_and_fkc_names`
400 :attr:`_schema.MetaData.sorted_tables`
402 """
404 with self._operation_context() as conn:
405 return self.dialect.get_table_names(
406 conn, schema, info_cache=self.info_cache, **kw
407 )
409 def has_table(
410 self, table_name: str, schema: Optional[str] = None, **kw: Any
411 ) -> bool:
412 r"""Return True if the backend has a table, view, or temporary
413 table of the given name.
415 :param table_name: name of the table to check
416 :param schema: schema name to query, if not the default schema.
417 :param \**kw: Additional keyword argument to pass to the dialect
418 specific implementation. See the documentation of the dialect
419 in use for more information.
421 .. versionadded:: 1.4 - the :meth:`.Inspector.has_table` method
422 replaces the :meth:`_engine.Engine.has_table` method.
424 .. versionchanged:: 2.0:: :meth:`.Inspector.has_table` now formally
425 supports checking for additional table-like objects:
427 * any type of views (plain or materialized)
428 * temporary tables of any kind
430 Previously, these two checks were not formally specified and
431 different dialects would vary in their behavior. The dialect
432 testing suite now includes tests for all of these object types
433 and should be supported by all SQLAlchemy-included dialects.
434 Support among third party dialects may be lagging, however.
436 """
437 with self._operation_context() as conn:
438 return self.dialect.has_table(
439 conn, table_name, schema, info_cache=self.info_cache, **kw
440 )
442 def has_sequence(
443 self, sequence_name: str, schema: Optional[str] = None, **kw: Any
444 ) -> bool:
445 r"""Return True if the backend has a sequence with the given name.
447 :param sequence_name: name of the sequence to check
448 :param schema: schema name to query, if not the default schema.
449 :param \**kw: Additional keyword argument to pass to the dialect
450 specific implementation. See the documentation of the dialect
451 in use for more information.
453 .. versionadded:: 1.4
455 """
456 with self._operation_context() as conn:
457 return self.dialect.has_sequence(
458 conn, sequence_name, schema, info_cache=self.info_cache, **kw
459 )
461 def has_index(
462 self,
463 table_name: str,
464 index_name: str,
465 schema: Optional[str] = None,
466 **kw: Any,
467 ) -> bool:
468 r"""Check the existence of a particular index name in the database.
470 :param table_name: the name of the table the index belongs to
471 :param index_name: the name of the index to check
472 :param schema: schema name to query, if not the default schema.
473 :param \**kw: Additional keyword argument to pass to the dialect
474 specific implementation. See the documentation of the dialect
475 in use for more information.
477 .. versionadded:: 2.0
479 """
480 with self._operation_context() as conn:
481 return self.dialect.has_index(
482 conn,
483 table_name,
484 index_name,
485 schema,
486 info_cache=self.info_cache,
487 **kw,
488 )
490 def has_schema(self, schema_name: str, **kw: Any) -> bool:
491 r"""Return True if the backend has a schema with the given name.
493 :param schema_name: name of the schema to check
494 :param \**kw: Additional keyword argument to pass to the dialect
495 specific implementation. See the documentation of the dialect
496 in use for more information.
498 .. versionadded:: 2.0
500 """
501 with self._operation_context() as conn:
502 return self.dialect.has_schema(
503 conn, schema_name, info_cache=self.info_cache, **kw
504 )
506 def get_sorted_table_and_fkc_names(
507 self,
508 schema: Optional[str] = None,
509 **kw: Any,
510 ) -> List[Tuple[Optional[str], List[Tuple[str, Optional[str]]]]]:
511 r"""Return dependency-sorted table and foreign key constraint names in
512 referred to within a particular schema.
514 This will yield 2-tuples of
515 ``(tablename, [(tname, fkname), (tname, fkname), ...])``
516 consisting of table names in CREATE order grouped with the foreign key
517 constraint names that are not detected as belonging to a cycle.
518 The final element
519 will be ``(None, [(tname, fkname), (tname, fkname), ..])``
520 which will consist of remaining
521 foreign key constraint names that would require a separate CREATE
522 step after-the-fact, based on dependencies between tables.
524 :param schema: schema name to query, if not the default schema.
525 :param \**kw: Additional keyword argument to pass to the dialect
526 specific implementation. See the documentation of the dialect
527 in use for more information.
529 .. seealso::
531 :meth:`_reflection.Inspector.get_table_names`
533 :func:`.sort_tables_and_constraints` - similar method which works
534 with an already-given :class:`_schema.MetaData`.
536 """
538 return [
539 (
540 table_key[1] if table_key else None,
541 [(tname, fks) for (_, tname), fks in fk_collection],
542 )
543 for (
544 table_key,
545 fk_collection,
546 ) in self.sort_tables_on_foreign_key_dependency(
547 consider_schemas=(schema,)
548 )
549 ]
551 def sort_tables_on_foreign_key_dependency(
552 self,
553 consider_schemas: Collection[Optional[str]] = (None,),
554 **kw: Any,
555 ) -> List[
556 Tuple[
557 Optional[Tuple[Optional[str], str]],
558 List[Tuple[Tuple[Optional[str], str], Optional[str]]],
559 ]
560 ]:
561 r"""Return dependency-sorted table and foreign key constraint names
562 referred to within multiple schemas.
564 This method may be compared to
565 :meth:`.Inspector.get_sorted_table_and_fkc_names`, which
566 works on one schema at a time; here, the method is a generalization
567 that will consider multiple schemas at once including that it will
568 resolve for cross-schema foreign keys.
570 .. versionadded:: 2.0
572 """
573 SchemaTab = Tuple[Optional[str], str]
575 tuples: Set[Tuple[SchemaTab, SchemaTab]] = set()
576 remaining_fkcs: Set[Tuple[SchemaTab, Optional[str]]] = set()
577 fknames_for_table: Dict[SchemaTab, Set[Optional[str]]] = {}
578 tnames: List[SchemaTab] = []
580 for schname in consider_schemas:
581 schema_fkeys = self.get_multi_foreign_keys(schname, **kw)
582 tnames.extend(schema_fkeys)
583 for (_, tname), fkeys in schema_fkeys.items():
584 fknames_for_table[(schname, tname)] = {
585 fk["name"] for fk in fkeys
586 }
587 for fkey in fkeys:
588 if (
589 tname != fkey["referred_table"]
590 or schname != fkey["referred_schema"]
591 ):
592 tuples.add(
593 (
594 (
595 fkey["referred_schema"],
596 fkey["referred_table"],
597 ),
598 (schname, tname),
599 )
600 )
601 try:
602 candidate_sort = list(topological.sort(tuples, tnames))
603 except exc.CircularDependencyError as err:
604 edge: Tuple[SchemaTab, SchemaTab]
605 for edge in err.edges:
606 tuples.remove(edge)
607 remaining_fkcs.update(
608 (edge[1], fkc) for fkc in fknames_for_table[edge[1]]
609 )
611 candidate_sort = list(topological.sort(tuples, tnames))
612 ret: List[
613 Tuple[Optional[SchemaTab], List[Tuple[SchemaTab, Optional[str]]]]
614 ]
615 ret = [
616 (
617 (schname, tname),
618 [
619 ((schname, tname), fk)
620 for fk in fknames_for_table[(schname, tname)].difference(
621 name for _, name in remaining_fkcs
622 )
623 ],
624 )
625 for (schname, tname) in candidate_sort
626 ]
627 return ret + [(None, list(remaining_fkcs))]
629 def get_temp_table_names(self, **kw: Any) -> List[str]:
630 r"""Return a list of temporary table names for the current bind.
632 This method is unsupported by most dialects; currently
633 only Oracle, PostgreSQL and SQLite implements it.
635 :param \**kw: Additional keyword argument to pass to the dialect
636 specific implementation. See the documentation of the dialect
637 in use for more information.
639 """
641 with self._operation_context() as conn:
642 return self.dialect.get_temp_table_names(
643 conn, info_cache=self.info_cache, **kw
644 )
646 def get_temp_view_names(self, **kw: Any) -> List[str]:
647 r"""Return a list of temporary view names for the current bind.
649 This method is unsupported by most dialects; currently
650 only PostgreSQL and SQLite implements it.
652 :param \**kw: Additional keyword argument to pass to the dialect
653 specific implementation. See the documentation of the dialect
654 in use for more information.
656 """
657 with self._operation_context() as conn:
658 return self.dialect.get_temp_view_names(
659 conn, info_cache=self.info_cache, **kw
660 )
662 def get_table_options(
663 self, table_name: str, schema: Optional[str] = None, **kw: Any
664 ) -> Dict[str, Any]:
665 r"""Return a dictionary of options specified when the table of the
666 given name was created.
668 This currently includes some options that apply to MySQL and Oracle
669 tables.
671 :param table_name: string name of the table. For special quoting,
672 use :class:`.quoted_name`.
674 :param schema: string schema name; if omitted, uses the default schema
675 of the database connection. For special quoting,
676 use :class:`.quoted_name`.
678 :param \**kw: Additional keyword argument to pass to the dialect
679 specific implementation. See the documentation of the dialect
680 in use for more information.
682 :return: a dict with the table options. The returned keys depend on the
683 dialect in use. Each one is prefixed with the dialect name.
685 .. seealso:: :meth:`Inspector.get_multi_table_options`
687 """
688 with self._operation_context() as conn:
689 return self.dialect.get_table_options(
690 conn, table_name, schema, info_cache=self.info_cache, **kw
691 )
693 def get_multi_table_options(
694 self,
695 schema: Optional[str] = None,
696 filter_names: Optional[Sequence[str]] = None,
697 kind: ObjectKind = ObjectKind.TABLE,
698 scope: ObjectScope = ObjectScope.DEFAULT,
699 **kw: Any,
700 ) -> Dict[TableKey, Dict[str, Any]]:
701 r"""Return a dictionary of options specified when the tables in the
702 given schema were created.
704 The tables can be filtered by passing the names to use to
705 ``filter_names``.
707 This currently includes some options that apply to MySQL and Oracle
708 tables.
710 :param schema: string schema name; if omitted, uses the default schema
711 of the database connection. For special quoting,
712 use :class:`.quoted_name`.
714 :param filter_names: optionally return information only for the
715 objects listed here.
717 :param kind: a :class:`.ObjectKind` that specifies the type of objects
718 to reflect. Defaults to ``ObjectKind.TABLE``.
720 :param scope: a :class:`.ObjectScope` that specifies if options of
721 default, temporary or any tables should be reflected.
722 Defaults to ``ObjectScope.DEFAULT``.
724 :param \**kw: Additional keyword argument to pass to the dialect
725 specific implementation. See the documentation of the dialect
726 in use for more information.
728 :return: a dictionary where the keys are two-tuple schema,table-name
729 and the values are dictionaries with the table options.
730 The returned keys in each dict depend on the
731 dialect in use. Each one is prefixed with the dialect name.
732 The schema is ``None`` if no schema is provided.
734 .. versionadded:: 2.0
736 .. seealso:: :meth:`Inspector.get_table_options`
737 """
738 with self._operation_context() as conn:
739 res = self.dialect.get_multi_table_options(
740 conn,
741 schema=schema,
742 filter_names=filter_names,
743 kind=kind,
744 scope=scope,
745 info_cache=self.info_cache,
746 **kw,
747 )
748 return dict(res)
750 def get_view_names(
751 self, schema: Optional[str] = None, **kw: Any
752 ) -> List[str]:
753 r"""Return all non-materialized view names in `schema`.
755 :param schema: Optional, retrieve names from a non-default schema.
756 For special quoting, use :class:`.quoted_name`.
757 :param \**kw: Additional keyword argument to pass to the dialect
758 specific implementation. See the documentation of the dialect
759 in use for more information.
762 .. versionchanged:: 2.0 For those dialects that previously included
763 the names of materialized views in this list (currently PostgreSQL),
764 this method no longer returns the names of materialized views.
765 the :meth:`.Inspector.get_materialized_view_names` method should
766 be used instead.
768 .. seealso::
770 :meth:`.Inspector.get_materialized_view_names`
772 """
774 with self._operation_context() as conn:
775 return self.dialect.get_view_names(
776 conn, schema, info_cache=self.info_cache, **kw
777 )
779 def get_materialized_view_names(
780 self, schema: Optional[str] = None, **kw: Any
781 ) -> List[str]:
782 r"""Return all materialized view names in `schema`.
784 :param schema: Optional, retrieve names from a non-default schema.
785 For special quoting, use :class:`.quoted_name`.
786 :param \**kw: Additional keyword argument to pass to the dialect
787 specific implementation. See the documentation of the dialect
788 in use for more information.
790 .. versionadded:: 2.0
792 .. seealso::
794 :meth:`.Inspector.get_view_names`
796 """
798 with self._operation_context() as conn:
799 return self.dialect.get_materialized_view_names(
800 conn, schema, info_cache=self.info_cache, **kw
801 )
803 def get_sequence_names(
804 self, schema: Optional[str] = None, **kw: Any
805 ) -> List[str]:
806 r"""Return all sequence names in `schema`.
808 :param schema: Optional, retrieve names from a non-default schema.
809 For special quoting, use :class:`.quoted_name`.
810 :param \**kw: Additional keyword argument to pass to the dialect
811 specific implementation. See the documentation of the dialect
812 in use for more information.
814 """
816 with self._operation_context() as conn:
817 return self.dialect.get_sequence_names(
818 conn, schema, info_cache=self.info_cache, **kw
819 )
821 def get_view_definition(
822 self, view_name: str, schema: Optional[str] = None, **kw: Any
823 ) -> str:
824 r"""Return definition for the plain or materialized view called
825 ``view_name``.
827 :param view_name: Name of the view.
828 :param schema: Optional, retrieve names from a non-default schema.
829 For special quoting, use :class:`.quoted_name`.
830 :param \**kw: Additional keyword argument to pass to the dialect
831 specific implementation. See the documentation of the dialect
832 in use for more information.
834 """
836 with self._operation_context() as conn:
837 return self.dialect.get_view_definition(
838 conn, view_name, schema, info_cache=self.info_cache, **kw
839 )
841 def get_columns(
842 self, table_name: str, schema: Optional[str] = None, **kw: Any
843 ) -> List[ReflectedColumn]:
844 r"""Return information about columns in ``table_name``.
846 Given a string ``table_name`` and an optional string ``schema``,
847 return column information as a list of :class:`.ReflectedColumn`.
849 :param table_name: string name of the table. For special quoting,
850 use :class:`.quoted_name`.
852 :param schema: string schema name; if omitted, uses the default schema
853 of the database connection. For special quoting,
854 use :class:`.quoted_name`.
856 :param \**kw: Additional keyword argument to pass to the dialect
857 specific implementation. See the documentation of the dialect
858 in use for more information.
860 :return: list of dictionaries, each representing the definition of
861 a database column.
863 .. seealso:: :meth:`Inspector.get_multi_columns`.
865 """
867 with self._operation_context() as conn:
868 col_defs = self.dialect.get_columns(
869 conn, table_name, schema, info_cache=self.info_cache, **kw
870 )
871 if col_defs:
872 self._instantiate_types([col_defs])
873 return col_defs
875 def _instantiate_types(
876 self, data: Iterable[List[ReflectedColumn]]
877 ) -> None:
878 # make this easy and only return instances for coltype
879 for col_defs in data:
880 for col_def in col_defs:
881 coltype = col_def["type"]
882 if not isinstance(coltype, TypeEngine):
883 col_def["type"] = coltype()
885 def get_multi_columns(
886 self,
887 schema: Optional[str] = None,
888 filter_names: Optional[Sequence[str]] = None,
889 kind: ObjectKind = ObjectKind.TABLE,
890 scope: ObjectScope = ObjectScope.DEFAULT,
891 **kw: Any,
892 ) -> Dict[TableKey, List[ReflectedColumn]]:
893 r"""Return information about columns in all objects in the given
894 schema.
896 The objects can be filtered by passing the names to use to
897 ``filter_names``.
899 For each table the value is a list of :class:`.ReflectedColumn`.
901 :param schema: string schema name; if omitted, uses the default schema
902 of the database connection. For special quoting,
903 use :class:`.quoted_name`.
905 :param filter_names: optionally return information only for the
906 objects listed here.
908 :param kind: a :class:`.ObjectKind` that specifies the type of objects
909 to reflect. Defaults to ``ObjectKind.TABLE``.
911 :param scope: a :class:`.ObjectScope` that specifies if columns of
912 default, temporary or any tables should be reflected.
913 Defaults to ``ObjectScope.DEFAULT``.
915 :param \**kw: Additional keyword argument to pass to the dialect
916 specific implementation. See the documentation of the dialect
917 in use for more information.
919 :return: a dictionary where the keys are two-tuple schema,table-name
920 and the values are list of dictionaries, each representing the
921 definition of a database column.
922 The schema is ``None`` if no schema is provided.
924 .. versionadded:: 2.0
926 .. seealso:: :meth:`Inspector.get_columns`
927 """
929 with self._operation_context() as conn:
930 table_col_defs = dict(
931 self.dialect.get_multi_columns(
932 conn,
933 schema=schema,
934 filter_names=filter_names,
935 kind=kind,
936 scope=scope,
937 info_cache=self.info_cache,
938 **kw,
939 )
940 )
941 self._instantiate_types(table_col_defs.values())
942 return table_col_defs
944 def get_pk_constraint(
945 self, table_name: str, schema: Optional[str] = None, **kw: Any
946 ) -> ReflectedPrimaryKeyConstraint:
947 r"""Return information about primary key constraint in ``table_name``.
949 Given a string ``table_name``, and an optional string `schema`, return
950 primary key information as a :class:`.ReflectedPrimaryKeyConstraint`.
952 :param table_name: string name of the table. For special quoting,
953 use :class:`.quoted_name`.
955 :param schema: string schema name; if omitted, uses the default schema
956 of the database connection. For special quoting,
957 use :class:`.quoted_name`.
959 :param \**kw: Additional keyword argument to pass to the dialect
960 specific implementation. See the documentation of the dialect
961 in use for more information.
963 :return: a dictionary representing the definition of
964 a primary key constraint.
966 .. seealso:: :meth:`Inspector.get_multi_pk_constraint`
967 """
968 with self._operation_context() as conn:
969 return self.dialect.get_pk_constraint(
970 conn, table_name, schema, info_cache=self.info_cache, **kw
971 )
973 def get_multi_pk_constraint(
974 self,
975 schema: Optional[str] = None,
976 filter_names: Optional[Sequence[str]] = None,
977 kind: ObjectKind = ObjectKind.TABLE,
978 scope: ObjectScope = ObjectScope.DEFAULT,
979 **kw: Any,
980 ) -> Dict[TableKey, ReflectedPrimaryKeyConstraint]:
981 r"""Return information about primary key constraints in
982 all tables in the given schema.
984 The tables can be filtered by passing the names to use to
985 ``filter_names``.
987 For each table the value is a :class:`.ReflectedPrimaryKeyConstraint`.
989 :param schema: string schema name; if omitted, uses the default schema
990 of the database connection. For special quoting,
991 use :class:`.quoted_name`.
993 :param filter_names: optionally return information only for the
994 objects listed here.
996 :param kind: a :class:`.ObjectKind` that specifies the type of objects
997 to reflect. Defaults to ``ObjectKind.TABLE``.
999 :param scope: a :class:`.ObjectScope` that specifies if primary keys of
1000 default, temporary or any tables should be reflected.
1001 Defaults to ``ObjectScope.DEFAULT``.
1003 :param \**kw: Additional keyword argument to pass to the dialect
1004 specific implementation. See the documentation of the dialect
1005 in use for more information.
1007 :return: a dictionary where the keys are two-tuple schema,table-name
1008 and the values are dictionaries, each representing the
1009 definition of a primary key constraint.
1010 The schema is ``None`` if no schema is provided.
1012 .. versionadded:: 2.0
1014 .. seealso:: :meth:`Inspector.get_pk_constraint`
1015 """
1016 with self._operation_context() as conn:
1017 return dict(
1018 self.dialect.get_multi_pk_constraint(
1019 conn,
1020 schema=schema,
1021 filter_names=filter_names,
1022 kind=kind,
1023 scope=scope,
1024 info_cache=self.info_cache,
1025 **kw,
1026 )
1027 )
1029 def get_foreign_keys(
1030 self, table_name: str, schema: Optional[str] = None, **kw: Any
1031 ) -> List[ReflectedForeignKeyConstraint]:
1032 r"""Return information about foreign_keys in ``table_name``.
1034 Given a string ``table_name``, and an optional string `schema`, return
1035 foreign key information as a list of
1036 :class:`.ReflectedForeignKeyConstraint`.
1038 :param table_name: string name of the table. For special quoting,
1039 use :class:`.quoted_name`.
1041 :param schema: string schema name; if omitted, uses the default schema
1042 of the database connection. For special quoting,
1043 use :class:`.quoted_name`.
1045 :param \**kw: Additional keyword argument to pass to the dialect
1046 specific implementation. See the documentation of the dialect
1047 in use for more information.
1049 :return: a list of dictionaries, each representing the
1050 a foreign key definition.
1052 .. seealso:: :meth:`Inspector.get_multi_foreign_keys`
1053 """
1055 with self._operation_context() as conn:
1056 return self.dialect.get_foreign_keys(
1057 conn, table_name, schema, info_cache=self.info_cache, **kw
1058 )
1060 def get_multi_foreign_keys(
1061 self,
1062 schema: Optional[str] = None,
1063 filter_names: Optional[Sequence[str]] = None,
1064 kind: ObjectKind = ObjectKind.TABLE,
1065 scope: ObjectScope = ObjectScope.DEFAULT,
1066 **kw: Any,
1067 ) -> Dict[TableKey, List[ReflectedForeignKeyConstraint]]:
1068 r"""Return information about foreign_keys in all tables
1069 in the given schema.
1071 The tables can be filtered by passing the names to use to
1072 ``filter_names``.
1074 For each table the value is a list of
1075 :class:`.ReflectedForeignKeyConstraint`.
1077 :param schema: string schema name; if omitted, uses the default schema
1078 of the database connection. For special quoting,
1079 use :class:`.quoted_name`.
1081 :param filter_names: optionally return information only for the
1082 objects listed here.
1084 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1085 to reflect. Defaults to ``ObjectKind.TABLE``.
1087 :param scope: a :class:`.ObjectScope` that specifies if foreign keys of
1088 default, temporary or any tables should be reflected.
1089 Defaults to ``ObjectScope.DEFAULT``.
1091 :param \**kw: Additional keyword argument to pass to the dialect
1092 specific implementation. See the documentation of the dialect
1093 in use for more information.
1095 :return: a dictionary where the keys are two-tuple schema,table-name
1096 and the values are list of dictionaries, each representing
1097 a foreign key definition.
1098 The schema is ``None`` if no schema is provided.
1100 .. versionadded:: 2.0
1102 .. seealso:: :meth:`Inspector.get_foreign_keys`
1103 """
1105 with self._operation_context() as conn:
1106 return dict(
1107 self.dialect.get_multi_foreign_keys(
1108 conn,
1109 schema=schema,
1110 filter_names=filter_names,
1111 kind=kind,
1112 scope=scope,
1113 info_cache=self.info_cache,
1114 **kw,
1115 )
1116 )
1118 def get_indexes(
1119 self, table_name: str, schema: Optional[str] = None, **kw: Any
1120 ) -> List[ReflectedIndex]:
1121 r"""Return information about indexes in ``table_name``.
1123 Given a string ``table_name`` and an optional string `schema`, return
1124 index information as a list of :class:`.ReflectedIndex`.
1126 :param table_name: string name of the table. For special quoting,
1127 use :class:`.quoted_name`.
1129 :param schema: string schema name; if omitted, uses the default schema
1130 of the database connection. For special quoting,
1131 use :class:`.quoted_name`.
1133 :param \**kw: Additional keyword argument to pass to the dialect
1134 specific implementation. See the documentation of the dialect
1135 in use for more information.
1137 :return: a list of dictionaries, each representing the
1138 definition of an index.
1140 .. seealso:: :meth:`Inspector.get_multi_indexes`
1141 """
1143 with self._operation_context() as conn:
1144 return self.dialect.get_indexes(
1145 conn, table_name, schema, info_cache=self.info_cache, **kw
1146 )
1148 def get_multi_indexes(
1149 self,
1150 schema: Optional[str] = None,
1151 filter_names: Optional[Sequence[str]] = None,
1152 kind: ObjectKind = ObjectKind.TABLE,
1153 scope: ObjectScope = ObjectScope.DEFAULT,
1154 **kw: Any,
1155 ) -> Dict[TableKey, List[ReflectedIndex]]:
1156 r"""Return information about indexes in in all objects
1157 in the given schema.
1159 The objects can be filtered by passing the names to use to
1160 ``filter_names``.
1162 For each table the value is a list of :class:`.ReflectedIndex`.
1164 :param schema: string schema name; if omitted, uses the default schema
1165 of the database connection. For special quoting,
1166 use :class:`.quoted_name`.
1168 :param filter_names: optionally return information only for the
1169 objects listed here.
1171 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1172 to reflect. Defaults to ``ObjectKind.TABLE``.
1174 :param scope: a :class:`.ObjectScope` that specifies if indexes of
1175 default, temporary or any tables should be reflected.
1176 Defaults to ``ObjectScope.DEFAULT``.
1178 :param \**kw: Additional keyword argument to pass to the dialect
1179 specific implementation. See the documentation of the dialect
1180 in use for more information.
1182 :return: a dictionary where the keys are two-tuple schema,table-name
1183 and the values are list of dictionaries, each representing the
1184 definition of an index.
1185 The schema is ``None`` if no schema is provided.
1187 .. versionadded:: 2.0
1189 .. seealso:: :meth:`Inspector.get_indexes`
1190 """
1192 with self._operation_context() as conn:
1193 return dict(
1194 self.dialect.get_multi_indexes(
1195 conn,
1196 schema=schema,
1197 filter_names=filter_names,
1198 kind=kind,
1199 scope=scope,
1200 info_cache=self.info_cache,
1201 **kw,
1202 )
1203 )
1205 def get_unique_constraints(
1206 self, table_name: str, schema: Optional[str] = None, **kw: Any
1207 ) -> List[ReflectedUniqueConstraint]:
1208 r"""Return information about unique constraints in ``table_name``.
1210 Given a string ``table_name`` and an optional string `schema`, return
1211 unique constraint information as a list of
1212 :class:`.ReflectedUniqueConstraint`.
1214 :param table_name: string name of the table. For special quoting,
1215 use :class:`.quoted_name`.
1217 :param schema: string schema name; if omitted, uses the default schema
1218 of the database connection. For special quoting,
1219 use :class:`.quoted_name`.
1221 :param \**kw: Additional keyword argument to pass to the dialect
1222 specific implementation. See the documentation of the dialect
1223 in use for more information.
1225 :return: a list of dictionaries, each representing the
1226 definition of an unique constraint.
1228 .. seealso:: :meth:`Inspector.get_multi_unique_constraints`
1229 """
1231 with self._operation_context() as conn:
1232 return self.dialect.get_unique_constraints(
1233 conn, table_name, schema, info_cache=self.info_cache, **kw
1234 )
1236 def get_multi_unique_constraints(
1237 self,
1238 schema: Optional[str] = None,
1239 filter_names: Optional[Sequence[str]] = None,
1240 kind: ObjectKind = ObjectKind.TABLE,
1241 scope: ObjectScope = ObjectScope.DEFAULT,
1242 **kw: Any,
1243 ) -> Dict[TableKey, List[ReflectedUniqueConstraint]]:
1244 r"""Return information about unique constraints in all tables
1245 in the given schema.
1247 The tables can be filtered by passing the names to use to
1248 ``filter_names``.
1250 For each table the value is a list of
1251 :class:`.ReflectedUniqueConstraint`.
1253 :param schema: string schema name; if omitted, uses the default schema
1254 of the database connection. For special quoting,
1255 use :class:`.quoted_name`.
1257 :param filter_names: optionally return information only for the
1258 objects listed here.
1260 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1261 to reflect. Defaults to ``ObjectKind.TABLE``.
1263 :param scope: a :class:`.ObjectScope` that specifies if constraints of
1264 default, temporary or any tables should be reflected.
1265 Defaults to ``ObjectScope.DEFAULT``.
1267 :param \**kw: Additional keyword argument to pass to the dialect
1268 specific implementation. See the documentation of the dialect
1269 in use for more information.
1271 :return: a dictionary where the keys are two-tuple schema,table-name
1272 and the values are list of dictionaries, each representing the
1273 definition of an unique constraint.
1274 The schema is ``None`` if no schema is provided.
1276 .. versionadded:: 2.0
1278 .. seealso:: :meth:`Inspector.get_unique_constraints`
1279 """
1281 with self._operation_context() as conn:
1282 return dict(
1283 self.dialect.get_multi_unique_constraints(
1284 conn,
1285 schema=schema,
1286 filter_names=filter_names,
1287 kind=kind,
1288 scope=scope,
1289 info_cache=self.info_cache,
1290 **kw,
1291 )
1292 )
1294 def get_table_comment(
1295 self, table_name: str, schema: Optional[str] = None, **kw: Any
1296 ) -> ReflectedTableComment:
1297 r"""Return information about the table comment for ``table_name``.
1299 Given a string ``table_name`` and an optional string ``schema``,
1300 return table comment information as a :class:`.ReflectedTableComment`.
1302 Raises ``NotImplementedError`` for a dialect that does not support
1303 comments.
1305 :param table_name: string name of the table. For special quoting,
1306 use :class:`.quoted_name`.
1308 :param schema: string schema name; if omitted, uses the default schema
1309 of the database connection. For special quoting,
1310 use :class:`.quoted_name`.
1312 :param \**kw: Additional keyword argument to pass to the dialect
1313 specific implementation. See the documentation of the dialect
1314 in use for more information.
1316 :return: a dictionary, with the table comment.
1318 .. versionadded:: 1.2
1320 .. seealso:: :meth:`Inspector.get_multi_table_comment`
1321 """
1323 with self._operation_context() as conn:
1324 return self.dialect.get_table_comment(
1325 conn, table_name, schema, info_cache=self.info_cache, **kw
1326 )
1328 def get_multi_table_comment(
1329 self,
1330 schema: Optional[str] = None,
1331 filter_names: Optional[Sequence[str]] = None,
1332 kind: ObjectKind = ObjectKind.TABLE,
1333 scope: ObjectScope = ObjectScope.DEFAULT,
1334 **kw: Any,
1335 ) -> Dict[TableKey, ReflectedTableComment]:
1336 r"""Return information about the table comment in all objects
1337 in the given schema.
1339 The objects can be filtered by passing the names to use to
1340 ``filter_names``.
1342 For each table the value is a :class:`.ReflectedTableComment`.
1344 Raises ``NotImplementedError`` for a dialect that does not support
1345 comments.
1347 :param schema: string schema name; if omitted, uses the default schema
1348 of the database connection. For special quoting,
1349 use :class:`.quoted_name`.
1351 :param filter_names: optionally return information only for the
1352 objects listed here.
1354 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1355 to reflect. Defaults to ``ObjectKind.TABLE``.
1357 :param scope: a :class:`.ObjectScope` that specifies if comments of
1358 default, temporary or any tables should be reflected.
1359 Defaults to ``ObjectScope.DEFAULT``.
1361 :param \**kw: Additional keyword argument to pass to the dialect
1362 specific implementation. See the documentation of the dialect
1363 in use for more information.
1365 :return: a dictionary where the keys are two-tuple schema,table-name
1366 and the values are dictionaries, representing the
1367 table comments.
1368 The schema is ``None`` if no schema is provided.
1370 .. versionadded:: 2.0
1372 .. seealso:: :meth:`Inspector.get_table_comment`
1373 """
1375 with self._operation_context() as conn:
1376 return dict(
1377 self.dialect.get_multi_table_comment(
1378 conn,
1379 schema=schema,
1380 filter_names=filter_names,
1381 kind=kind,
1382 scope=scope,
1383 info_cache=self.info_cache,
1384 **kw,
1385 )
1386 )
1388 def get_check_constraints(
1389 self, table_name: str, schema: Optional[str] = None, **kw: Any
1390 ) -> List[ReflectedCheckConstraint]:
1391 r"""Return information about check constraints in ``table_name``.
1393 Given a string ``table_name`` and an optional string `schema`, return
1394 check constraint information as a list of
1395 :class:`.ReflectedCheckConstraint`.
1397 :param table_name: string name of the table. For special quoting,
1398 use :class:`.quoted_name`.
1400 :param schema: string schema name; if omitted, uses the default schema
1401 of the database connection. For special quoting,
1402 use :class:`.quoted_name`.
1404 :param \**kw: Additional keyword argument to pass to the dialect
1405 specific implementation. See the documentation of the dialect
1406 in use for more information.
1408 :return: a list of dictionaries, each representing the
1409 definition of a check constraints.
1411 .. seealso:: :meth:`Inspector.get_multi_check_constraints`
1412 """
1414 with self._operation_context() as conn:
1415 return self.dialect.get_check_constraints(
1416 conn, table_name, schema, info_cache=self.info_cache, **kw
1417 )
1419 def get_multi_check_constraints(
1420 self,
1421 schema: Optional[str] = None,
1422 filter_names: Optional[Sequence[str]] = None,
1423 kind: ObjectKind = ObjectKind.TABLE,
1424 scope: ObjectScope = ObjectScope.DEFAULT,
1425 **kw: Any,
1426 ) -> Dict[TableKey, List[ReflectedCheckConstraint]]:
1427 r"""Return information about check constraints in all tables
1428 in the given schema.
1430 The tables can be filtered by passing the names to use to
1431 ``filter_names``.
1433 For each table the value is a list of
1434 :class:`.ReflectedCheckConstraint`.
1436 :param schema: string schema name; if omitted, uses the default schema
1437 of the database connection. For special quoting,
1438 use :class:`.quoted_name`.
1440 :param filter_names: optionally return information only for the
1441 objects listed here.
1443 :param kind: a :class:`.ObjectKind` that specifies the type of objects
1444 to reflect. Defaults to ``ObjectKind.TABLE``.
1446 :param scope: a :class:`.ObjectScope` that specifies if constraints of
1447 default, temporary or any tables should be reflected.
1448 Defaults to ``ObjectScope.DEFAULT``.
1450 :param \**kw: Additional keyword argument to pass to the dialect
1451 specific implementation. See the documentation of the dialect
1452 in use for more information.
1454 :return: a dictionary where the keys are two-tuple schema,table-name
1455 and the values are list of dictionaries, each representing the
1456 definition of a check constraints.
1457 The schema is ``None`` if no schema is provided.
1459 .. versionadded:: 2.0
1461 .. seealso:: :meth:`Inspector.get_check_constraints`
1462 """
1464 with self._operation_context() as conn:
1465 return dict(
1466 self.dialect.get_multi_check_constraints(
1467 conn,
1468 schema=schema,
1469 filter_names=filter_names,
1470 kind=kind,
1471 scope=scope,
1472 info_cache=self.info_cache,
1473 **kw,
1474 )
1475 )
1477 def reflect_table(
1478 self,
1479 table: sa_schema.Table,
1480 include_columns: Optional[Collection[str]],
1481 exclude_columns: Collection[str] = (),
1482 resolve_fks: bool = True,
1483 _extend_on: Optional[Set[sa_schema.Table]] = None,
1484 _reflect_info: Optional[_ReflectionInfo] = None,
1485 ) -> None:
1486 """Given a :class:`_schema.Table` object, load its internal
1487 constructs based on introspection.
1489 This is the underlying method used by most dialects to produce
1490 table reflection. Direct usage is like::
1492 from sqlalchemy import create_engine, MetaData, Table
1493 from sqlalchemy import inspect
1495 engine = create_engine('...')
1496 meta = MetaData()
1497 user_table = Table('user', meta)
1498 insp = inspect(engine)
1499 insp.reflect_table(user_table, None)
1501 .. versionchanged:: 1.4 Renamed from ``reflecttable`` to
1502 ``reflect_table``
1504 :param table: a :class:`~sqlalchemy.schema.Table` instance.
1505 :param include_columns: a list of string column names to include
1506 in the reflection process. If ``None``, all columns are reflected.
1508 """
1510 if _extend_on is not None:
1511 if table in _extend_on:
1512 return
1513 else:
1514 _extend_on.add(table)
1516 dialect = self.bind.dialect
1518 with self._operation_context() as conn:
1519 schema = conn.schema_for_object(table)
1521 table_name = table.name
1523 # get table-level arguments that are specifically
1524 # intended for reflection, e.g. oracle_resolve_synonyms.
1525 # these are unconditionally passed to related Table
1526 # objects
1527 reflection_options = {
1528 k: table.dialect_kwargs.get(k)
1529 for k in dialect.reflection_options
1530 if k in table.dialect_kwargs
1531 }
1533 table_key = (schema, table_name)
1534 if _reflect_info is None or table_key not in _reflect_info.columns:
1535 _reflect_info = self._get_reflection_info(
1536 schema,
1537 filter_names=[table_name],
1538 kind=ObjectKind.ANY,
1539 scope=ObjectScope.ANY,
1540 _reflect_info=_reflect_info,
1541 **table.dialect_kwargs,
1542 )
1543 if table_key in _reflect_info.unreflectable:
1544 raise _reflect_info.unreflectable[table_key]
1546 if table_key not in _reflect_info.columns:
1547 raise exc.NoSuchTableError(table_name)
1549 # reflect table options, like mysql_engine
1550 if _reflect_info.table_options:
1551 tbl_opts = _reflect_info.table_options.get(table_key)
1552 if tbl_opts:
1553 # add additional kwargs to the Table if the dialect
1554 # returned them
1555 table._validate_dialect_kwargs(tbl_opts)
1557 found_table = False
1558 cols_by_orig_name: Dict[str, sa_schema.Column[Any]] = {}
1560 for col_d in _reflect_info.columns[table_key]:
1561 found_table = True
1563 self._reflect_column(
1564 table,
1565 col_d,
1566 include_columns,
1567 exclude_columns,
1568 cols_by_orig_name,
1569 )
1571 # NOTE: support tables/views with no columns
1572 if not found_table and not self.has_table(table_name, schema):
1573 raise exc.NoSuchTableError(table_name)
1575 self._reflect_pk(
1576 _reflect_info, table_key, table, cols_by_orig_name, exclude_columns
1577 )
1579 self._reflect_fk(
1580 _reflect_info,
1581 table_key,
1582 table,
1583 cols_by_orig_name,
1584 include_columns,
1585 exclude_columns,
1586 resolve_fks,
1587 _extend_on,
1588 reflection_options,
1589 )
1591 self._reflect_indexes(
1592 _reflect_info,
1593 table_key,
1594 table,
1595 cols_by_orig_name,
1596 include_columns,
1597 exclude_columns,
1598 reflection_options,
1599 )
1601 self._reflect_unique_constraints(
1602 _reflect_info,
1603 table_key,
1604 table,
1605 cols_by_orig_name,
1606 include_columns,
1607 exclude_columns,
1608 reflection_options,
1609 )
1611 self._reflect_check_constraints(
1612 _reflect_info,
1613 table_key,
1614 table,
1615 cols_by_orig_name,
1616 include_columns,
1617 exclude_columns,
1618 reflection_options,
1619 )
1621 self._reflect_table_comment(
1622 _reflect_info,
1623 table_key,
1624 table,
1625 reflection_options,
1626 )
1628 def _reflect_column(
1629 self,
1630 table: sa_schema.Table,
1631 col_d: ReflectedColumn,
1632 include_columns: Optional[Collection[str]],
1633 exclude_columns: Collection[str],
1634 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1635 ) -> None:
1636 orig_name = col_d["name"]
1638 table.metadata.dispatch.column_reflect(self, table, col_d)
1639 table.dispatch.column_reflect(self, table, col_d)
1641 # fetch name again as column_reflect is allowed to
1642 # change it
1643 name = col_d["name"]
1644 if (include_columns and name not in include_columns) or (
1645 exclude_columns and name in exclude_columns
1646 ):
1647 return
1649 coltype = col_d["type"]
1651 col_kw = {
1652 k: col_d[k] # type: ignore[literal-required]
1653 for k in [
1654 "nullable",
1655 "autoincrement",
1656 "quote",
1657 "info",
1658 "key",
1659 "comment",
1660 ]
1661 if k in col_d
1662 }
1664 if "dialect_options" in col_d:
1665 col_kw.update(col_d["dialect_options"])
1667 colargs = []
1668 default: Any
1669 if col_d.get("default") is not None:
1670 default_text = col_d["default"]
1671 assert default_text is not None
1672 if isinstance(default_text, TextClause):
1673 default = sa_schema.DefaultClause(
1674 default_text, _reflected=True
1675 )
1676 elif not isinstance(default_text, sa_schema.FetchedValue):
1677 default = sa_schema.DefaultClause(
1678 sql.text(default_text), _reflected=True
1679 )
1680 else:
1681 default = default_text
1682 colargs.append(default)
1684 if "computed" in col_d:
1685 computed = sa_schema.Computed(**col_d["computed"])
1686 colargs.append(computed)
1688 if "identity" in col_d:
1689 identity = sa_schema.Identity(**col_d["identity"])
1690 colargs.append(identity)
1692 cols_by_orig_name[orig_name] = col = sa_schema.Column(
1693 name, coltype, *colargs, **col_kw
1694 )
1696 if col.key in table.primary_key:
1697 col.primary_key = True
1698 table.append_column(col, replace_existing=True)
1700 def _reflect_pk(
1701 self,
1702 _reflect_info: _ReflectionInfo,
1703 table_key: TableKey,
1704 table: sa_schema.Table,
1705 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1706 exclude_columns: Collection[str],
1707 ) -> None:
1708 pk_cons = _reflect_info.pk_constraint.get(table_key)
1709 if pk_cons:
1710 pk_cols = [
1711 cols_by_orig_name[pk]
1712 for pk in pk_cons["constrained_columns"]
1713 if pk in cols_by_orig_name and pk not in exclude_columns
1714 ]
1716 # update pk constraint name and comment
1717 table.primary_key.name = pk_cons.get("name")
1718 table.primary_key.comment = pk_cons.get("comment", None)
1720 # tell the PKConstraint to re-initialize
1721 # its column collection
1722 table.primary_key._reload(pk_cols)
1724 def _reflect_fk(
1725 self,
1726 _reflect_info: _ReflectionInfo,
1727 table_key: TableKey,
1728 table: sa_schema.Table,
1729 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1730 include_columns: Optional[Collection[str]],
1731 exclude_columns: Collection[str],
1732 resolve_fks: bool,
1733 _extend_on: Optional[Set[sa_schema.Table]],
1734 reflection_options: Dict[str, Any],
1735 ) -> None:
1736 fkeys = _reflect_info.foreign_keys.get(table_key, [])
1737 for fkey_d in fkeys:
1738 conname = fkey_d["name"]
1739 # look for columns by orig name in cols_by_orig_name,
1740 # but support columns that are in-Python only as fallback
1741 constrained_columns = [
1742 cols_by_orig_name[c].key if c in cols_by_orig_name else c
1743 for c in fkey_d["constrained_columns"]
1744 ]
1746 if (
1747 exclude_columns
1748 and set(constrained_columns).intersection(exclude_columns)
1749 or (
1750 include_columns
1751 and set(constrained_columns).difference(include_columns)
1752 )
1753 ):
1754 continue
1756 referred_schema = fkey_d["referred_schema"]
1757 referred_table = fkey_d["referred_table"]
1758 referred_columns = fkey_d["referred_columns"]
1759 refspec = []
1760 if referred_schema is not None:
1761 if resolve_fks:
1762 sa_schema.Table(
1763 referred_table,
1764 table.metadata,
1765 schema=referred_schema,
1766 autoload_with=self.bind,
1767 _extend_on=_extend_on,
1768 _reflect_info=_reflect_info,
1769 **reflection_options,
1770 )
1771 for column in referred_columns:
1772 refspec.append(
1773 ".".join([referred_schema, referred_table, column])
1774 )
1775 else:
1776 if resolve_fks:
1777 sa_schema.Table(
1778 referred_table,
1779 table.metadata,
1780 autoload_with=self.bind,
1781 schema=sa_schema.BLANK_SCHEMA,
1782 _extend_on=_extend_on,
1783 _reflect_info=_reflect_info,
1784 **reflection_options,
1785 )
1786 for column in referred_columns:
1787 refspec.append(".".join([referred_table, column]))
1788 if "options" in fkey_d:
1789 options = fkey_d["options"]
1790 else:
1791 options = {}
1793 try:
1794 table.append_constraint(
1795 sa_schema.ForeignKeyConstraint(
1796 constrained_columns,
1797 refspec,
1798 conname,
1799 link_to_name=True,
1800 comment=fkey_d.get("comment"),
1801 **options,
1802 )
1803 )
1804 except exc.ConstraintColumnNotFoundError:
1805 util.warn(
1806 f"On reflected table {table.name}, skipping reflection of "
1807 "foreign key constraint "
1808 f"{conname}; one or more subject columns within "
1809 f"name(s) {', '.join(constrained_columns)} are not "
1810 "present in the table"
1811 )
1813 _index_sort_exprs = {
1814 "asc": operators.asc_op,
1815 "desc": operators.desc_op,
1816 "nulls_first": operators.nulls_first_op,
1817 "nulls_last": operators.nulls_last_op,
1818 }
1820 def _reflect_indexes(
1821 self,
1822 _reflect_info: _ReflectionInfo,
1823 table_key: TableKey,
1824 table: sa_schema.Table,
1825 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1826 include_columns: Optional[Collection[str]],
1827 exclude_columns: Collection[str],
1828 reflection_options: Dict[str, Any],
1829 ) -> None:
1830 # Indexes
1831 indexes = _reflect_info.indexes.get(table_key, [])
1832 for index_d in indexes:
1833 name = index_d["name"]
1834 columns = index_d["column_names"]
1835 expressions = index_d.get("expressions")
1836 column_sorting = index_d.get("column_sorting", {})
1837 unique = index_d["unique"]
1838 flavor = index_d.get("type", "index")
1839 dialect_options = index_d.get("dialect_options", {})
1841 duplicates = index_d.get("duplicates_constraint")
1842 if include_columns and not set(columns).issubset(include_columns):
1843 continue
1844 if duplicates:
1845 continue
1846 # look for columns by orig name in cols_by_orig_name,
1847 # but support columns that are in-Python only as fallback
1848 idx_element: Any
1849 idx_elements = []
1850 for index, c in enumerate(columns):
1851 if c is None:
1852 if not expressions:
1853 util.warn(
1854 f"Skipping {flavor} {name!r} because key "
1855 f"{index + 1} reflected as None but no "
1856 "'expressions' were returned"
1857 )
1858 break
1859 idx_element = sql.text(expressions[index])
1860 else:
1861 try:
1862 if c in cols_by_orig_name:
1863 idx_element = cols_by_orig_name[c]
1864 else:
1865 idx_element = table.c[c]
1866 except KeyError:
1867 util.warn(
1868 f"{flavor} key {c!r} was not located in "
1869 f"columns for table {table.name!r}"
1870 )
1871 continue
1872 for option in column_sorting.get(c, ()):
1873 if option in self._index_sort_exprs:
1874 op = self._index_sort_exprs[option]
1875 idx_element = op(idx_element)
1876 idx_elements.append(idx_element)
1877 else:
1878 sa_schema.Index(
1879 name,
1880 *idx_elements,
1881 _table=table,
1882 unique=unique,
1883 **dialect_options,
1884 )
1886 def _reflect_unique_constraints(
1887 self,
1888 _reflect_info: _ReflectionInfo,
1889 table_key: TableKey,
1890 table: sa_schema.Table,
1891 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1892 include_columns: Optional[Collection[str]],
1893 exclude_columns: Collection[str],
1894 reflection_options: Dict[str, Any],
1895 ) -> None:
1896 constraints = _reflect_info.unique_constraints.get(table_key, [])
1897 # Unique Constraints
1898 for const_d in constraints:
1899 conname = const_d["name"]
1900 columns = const_d["column_names"]
1901 comment = const_d.get("comment")
1902 duplicates = const_d.get("duplicates_index")
1903 dialect_options = const_d.get("dialect_options", {})
1904 if include_columns and not set(columns).issubset(include_columns):
1905 continue
1906 if duplicates:
1907 continue
1908 # look for columns by orig name in cols_by_orig_name,
1909 # but support columns that are in-Python only as fallback
1910 constrained_cols = []
1911 for c in columns:
1912 try:
1913 constrained_col = (
1914 cols_by_orig_name[c]
1915 if c in cols_by_orig_name
1916 else table.c[c]
1917 )
1918 except KeyError:
1919 util.warn(
1920 "unique constraint key '%s' was not located in "
1921 "columns for table '%s'" % (c, table.name)
1922 )
1923 else:
1924 constrained_cols.append(constrained_col)
1925 table.append_constraint(
1926 sa_schema.UniqueConstraint(
1927 *constrained_cols,
1928 name=conname,
1929 comment=comment,
1930 **dialect_options,
1931 )
1932 )
1934 def _reflect_check_constraints(
1935 self,
1936 _reflect_info: _ReflectionInfo,
1937 table_key: TableKey,
1938 table: sa_schema.Table,
1939 cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
1940 include_columns: Optional[Collection[str]],
1941 exclude_columns: Collection[str],
1942 reflection_options: Dict[str, Any],
1943 ) -> None:
1944 constraints = _reflect_info.check_constraints.get(table_key, [])
1945 for const_d in constraints:
1946 table.append_constraint(sa_schema.CheckConstraint(**const_d))
1948 def _reflect_table_comment(
1949 self,
1950 _reflect_info: _ReflectionInfo,
1951 table_key: TableKey,
1952 table: sa_schema.Table,
1953 reflection_options: Dict[str, Any],
1954 ) -> None:
1955 comment_dict = _reflect_info.table_comment.get(table_key)
1956 if comment_dict:
1957 table.comment = comment_dict["text"]
1959 def _get_reflection_info(
1960 self,
1961 schema: Optional[str] = None,
1962 filter_names: Optional[Collection[str]] = None,
1963 available: Optional[Collection[str]] = None,
1964 _reflect_info: Optional[_ReflectionInfo] = None,
1965 **kw: Any,
1966 ) -> _ReflectionInfo:
1967 kw["schema"] = schema
1969 if filter_names and available and len(filter_names) > 100:
1970 fraction = len(filter_names) / len(available)
1971 else:
1972 fraction = None
1974 unreflectable: Dict[TableKey, exc.UnreflectableTableError]
1975 kw["unreflectable"] = unreflectable = {}
1977 has_result: bool = True
1979 def run(
1980 meth: Any,
1981 *,
1982 optional: bool = False,
1983 check_filter_names_from_meth: bool = False,
1984 ) -> Any:
1985 nonlocal has_result
1986 # simple heuristic to improve reflection performance if a
1987 # dialect implements multi_reflection:
1988 # if more than 50% of the tables in the db are in filter_names
1989 # load all the tables, since it's most likely faster to avoid
1990 # a filter on that many tables.
1991 if (
1992 fraction is None
1993 or fraction <= 0.5
1994 or not self.dialect._overrides_default(meth.__name__)
1995 ):
1996 _fn = filter_names
1997 else:
1998 _fn = None
1999 try:
2000 if has_result:
2001 res = meth(filter_names=_fn, **kw)
2002 if check_filter_names_from_meth and not res:
2003 # method returned no result data.
2004 # skip any future call methods
2005 has_result = False
2006 else:
2007 res = {}
2008 except NotImplementedError:
2009 if not optional:
2010 raise
2011 res = {}
2012 return res
2014 info = _ReflectionInfo(
2015 columns=run(
2016 self.get_multi_columns, check_filter_names_from_meth=True
2017 ),
2018 pk_constraint=run(self.get_multi_pk_constraint),
2019 foreign_keys=run(self.get_multi_foreign_keys),
2020 indexes=run(self.get_multi_indexes),
2021 unique_constraints=run(
2022 self.get_multi_unique_constraints, optional=True
2023 ),
2024 table_comment=run(self.get_multi_table_comment, optional=True),
2025 check_constraints=run(
2026 self.get_multi_check_constraints, optional=True
2027 ),
2028 table_options=run(self.get_multi_table_options, optional=True),
2029 unreflectable=unreflectable,
2030 )
2031 if _reflect_info:
2032 _reflect_info.update(info)
2033 return _reflect_info
2034 else:
2035 return info
2038@final
2039class ReflectionDefaults:
2040 """provides blank default values for reflection methods."""
2042 @classmethod
2043 def columns(cls) -> List[ReflectedColumn]:
2044 return []
2046 @classmethod
2047 def pk_constraint(cls) -> ReflectedPrimaryKeyConstraint:
2048 return {
2049 "name": None,
2050 "constrained_columns": [],
2051 }
2053 @classmethod
2054 def foreign_keys(cls) -> List[ReflectedForeignKeyConstraint]:
2055 return []
2057 @classmethod
2058 def indexes(cls) -> List[ReflectedIndex]:
2059 return []
2061 @classmethod
2062 def unique_constraints(cls) -> List[ReflectedUniqueConstraint]:
2063 return []
2065 @classmethod
2066 def check_constraints(cls) -> List[ReflectedCheckConstraint]:
2067 return []
2069 @classmethod
2070 def table_options(cls) -> Dict[str, Any]:
2071 return {}
2073 @classmethod
2074 def table_comment(cls) -> ReflectedTableComment:
2075 return {"text": None}
2078@dataclass
2079class _ReflectionInfo:
2080 columns: Dict[TableKey, List[ReflectedColumn]]
2081 pk_constraint: Dict[TableKey, Optional[ReflectedPrimaryKeyConstraint]]
2082 foreign_keys: Dict[TableKey, List[ReflectedForeignKeyConstraint]]
2083 indexes: Dict[TableKey, List[ReflectedIndex]]
2084 # optionals
2085 unique_constraints: Dict[TableKey, List[ReflectedUniqueConstraint]]
2086 table_comment: Dict[TableKey, Optional[ReflectedTableComment]]
2087 check_constraints: Dict[TableKey, List[ReflectedCheckConstraint]]
2088 table_options: Dict[TableKey, Dict[str, Any]]
2089 unreflectable: Dict[TableKey, exc.UnreflectableTableError]
2091 def update(self, other: _ReflectionInfo) -> None:
2092 for k, v in self.__dict__.items():
2093 ov = getattr(other, k)
2094 if ov is not None:
2095 if v is None:
2096 setattr(self, k, ov)
2097 else:
2098 v.update(ov)