Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/sql/ddl.py: 52%
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# sql/ddl.py
2# Copyright (C) 2009-2025 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: allow-untyped-defs, allow-untyped-calls
9"""
10Provides the hierarchy of DDL-defining schema items as well as routines
11to invoke them for a create/drop call.
13"""
14from __future__ import annotations
16import contextlib
17from enum import auto
18from enum import Flag
19import typing
20from typing import Any
21from typing import Callable
22from typing import Generic
23from typing import Iterable
24from typing import List
25from typing import Optional
26from typing import Protocol
27from typing import Sequence as typing_Sequence
28from typing import Tuple
29from typing import TypeVar
30from typing import Union
32from . import coercions
33from . import roles
34from . import util as sql_util
35from .base import _generative
36from .base import DialectKWArgs
37from .base import Executable
38from .base import SchemaVisitor
39from .elements import ClauseElement
40from .selectable import SelectBase
41from .selectable import TableClause
42from .. import exc
43from .. import util
44from ..util import topological
45from ..util.typing import Self
48if typing.TYPE_CHECKING:
49 from .compiler import Compiled
50 from .compiler import DDLCompiler
51 from .elements import BindParameter
52 from .schema import Column
53 from .schema import Constraint
54 from .schema import ForeignKeyConstraint
55 from .schema import Index
56 from .schema import MetaData
57 from .schema import SchemaItem
58 from .schema import Sequence as Sequence # noqa: F401
59 from .schema import Table
60 from ..engine.base import Connection
61 from ..engine.interfaces import _CoreSingleExecuteParams
62 from ..engine.interfaces import CacheStats
63 from ..engine.interfaces import CompiledCacheType
64 from ..engine.interfaces import Dialect
65 from ..engine.interfaces import SchemaTranslateMapType
67_SI = TypeVar("_SI", bound=Union["SchemaItem", str])
70class BaseDDLElement(ClauseElement):
71 """The root of DDL constructs, including those that are sub-elements
72 within the "create table" and other processes.
74 .. versionadded:: 2.0
76 """
78 _hierarchy_supports_caching = False
79 """disable cache warnings for all _DDLCompiles subclasses. """
81 def _compiler(self, dialect, **kw):
82 """Return a compiler appropriate for this ClauseElement, given a
83 Dialect."""
85 return dialect.ddl_compiler(dialect, self, **kw)
87 def _compile_w_cache(
88 self,
89 dialect: Dialect,
90 *,
91 compiled_cache: Optional[CompiledCacheType],
92 column_keys: List[str],
93 for_executemany: bool = False,
94 schema_translate_map: Optional[SchemaTranslateMapType] = None,
95 **kw: Any,
96 ) -> tuple[
97 Compiled,
98 typing_Sequence[BindParameter[Any]] | None,
99 _CoreSingleExecuteParams | None,
100 CacheStats,
101 ]:
102 raise NotImplementedError()
105class DDLIfCallable(Protocol):
106 def __call__(
107 self,
108 ddl: BaseDDLElement,
109 target: Union[SchemaItem, str],
110 bind: Optional[Connection],
111 tables: Optional[List[Table]] = None,
112 state: Optional[Any] = None,
113 *,
114 dialect: Dialect,
115 compiler: Optional[DDLCompiler] = ...,
116 checkfirst: bool,
117 ) -> bool: ...
120class DDLIf(typing.NamedTuple):
121 dialect: Optional[str]
122 callable_: Optional[DDLIfCallable]
123 state: Optional[Any]
125 def _should_execute(
126 self,
127 ddl: BaseDDLElement,
128 target: Union[SchemaItem, str],
129 bind: Optional[Connection],
130 compiler: Optional[DDLCompiler] = None,
131 **kw: Any,
132 ) -> bool:
133 if bind is not None:
134 dialect = bind.dialect
135 elif compiler is not None:
136 dialect = compiler.dialect
137 else:
138 assert False, "compiler or dialect is required"
140 if isinstance(self.dialect, str):
141 if self.dialect != dialect.name:
142 return False
143 elif isinstance(self.dialect, (tuple, list, set)):
144 if dialect.name not in self.dialect:
145 return False
146 if self.callable_ is not None and not self.callable_(
147 ddl,
148 target,
149 bind,
150 state=self.state,
151 dialect=dialect,
152 compiler=compiler,
153 **kw,
154 ):
155 return False
157 return True
160class ExecutableDDLElement(roles.DDLRole, Executable, BaseDDLElement):
161 """Base class for standalone executable DDL expression constructs.
163 This class is the base for the general purpose :class:`.DDL` class,
164 as well as the various create/drop clause constructs such as
165 :class:`.CreateTable`, :class:`.DropTable`, :class:`.AddConstraint`,
166 etc.
168 .. versionchanged:: 2.0 :class:`.ExecutableDDLElement` is renamed from
169 :class:`.DDLElement`, which still exists for backwards compatibility.
171 :class:`.ExecutableDDLElement` integrates closely with SQLAlchemy events,
172 introduced in :ref:`event_toplevel`. An instance of one is
173 itself an event receiving callable::
175 event.listen(
176 users,
177 "after_create",
178 AddConstraint(constraint).execute_if(dialect="postgresql"),
179 )
181 .. seealso::
183 :class:`.DDL`
185 :class:`.DDLEvents`
187 :ref:`event_toplevel`
189 :ref:`schema_ddl_sequences`
191 """
193 _ddl_if: Optional[DDLIf] = None
194 target: Union[SchemaItem, str, None] = None
196 def _execute_on_connection(
197 self, connection, distilled_params, execution_options
198 ):
199 return connection._execute_ddl(
200 self, distilled_params, execution_options
201 )
203 @_generative
204 def against(self, target: SchemaItem) -> Self:
205 """Return a copy of this :class:`_schema.ExecutableDDLElement` which
206 will include the given target.
208 This essentially applies the given item to the ``.target`` attribute of
209 the returned :class:`_schema.ExecutableDDLElement` object. This target
210 is then usable by event handlers and compilation routines in order to
211 provide services such as tokenization of a DDL string in terms of a
212 particular :class:`_schema.Table`.
214 When a :class:`_schema.ExecutableDDLElement` object is established as
215 an event handler for the :meth:`_events.DDLEvents.before_create` or
216 :meth:`_events.DDLEvents.after_create` events, and the event then
217 occurs for a given target such as a :class:`_schema.Constraint` or
218 :class:`_schema.Table`, that target is established with a copy of the
219 :class:`_schema.ExecutableDDLElement` object using this method, which
220 then proceeds to the :meth:`_schema.ExecutableDDLElement.execute`
221 method in order to invoke the actual DDL instruction.
223 :param target: a :class:`_schema.SchemaItem` that will be the subject
224 of a DDL operation.
226 :return: a copy of this :class:`_schema.ExecutableDDLElement` with the
227 ``.target`` attribute assigned to the given
228 :class:`_schema.SchemaItem`.
230 .. seealso::
232 :class:`_schema.DDL` - uses tokenization against the "target" when
233 processing the DDL string.
235 """
236 self.target = target
237 return self
239 @_generative
240 def execute_if(
241 self,
242 dialect: Optional[str] = None,
243 callable_: Optional[DDLIfCallable] = None,
244 state: Optional[Any] = None,
245 ) -> Self:
246 r"""Return a callable that will execute this
247 :class:`_ddl.ExecutableDDLElement` conditionally within an event
248 handler.
250 Used to provide a wrapper for event listening::
252 event.listen(
253 metadata,
254 "before_create",
255 DDL("my_ddl").execute_if(dialect="postgresql"),
256 )
258 :param dialect: May be a string or tuple of strings.
259 If a string, it will be compared to the name of the
260 executing database dialect::
262 DDL("something").execute_if(dialect="postgresql")
264 If a tuple, specifies multiple dialect names::
266 DDL("something").execute_if(dialect=("postgresql", "mysql"))
268 :param callable\_: A callable, which will be invoked with
269 three positional arguments as well as optional keyword
270 arguments:
272 :ddl:
273 This DDL element.
275 :target:
276 The :class:`_schema.Table` or :class:`_schema.MetaData`
277 object which is the
278 target of this event. May be None if the DDL is executed
279 explicitly.
281 :bind:
282 The :class:`_engine.Connection` being used for DDL execution.
283 May be None if this construct is being created inline within
284 a table, in which case ``compiler`` will be present.
286 :tables:
287 Optional keyword argument - a list of Table objects which are to
288 be created/ dropped within a MetaData.create_all() or drop_all()
289 method call.
291 :dialect: keyword argument, but always present - the
292 :class:`.Dialect` involved in the operation.
294 :compiler: keyword argument. Will be ``None`` for an engine
295 level DDL invocation, but will refer to a :class:`.DDLCompiler`
296 if this DDL element is being created inline within a table.
298 :state:
299 Optional keyword argument - will be the ``state`` argument
300 passed to this function.
302 :checkfirst:
303 Keyword argument, will be True if the 'checkfirst' flag was
304 set during the call to ``create()``, ``create_all()``,
305 ``drop()``, ``drop_all()``.
307 If the callable returns a True value, the DDL statement will be
308 executed.
310 :param state: any value which will be passed to the callable\_
311 as the ``state`` keyword argument.
313 .. seealso::
315 :meth:`.SchemaItem.ddl_if`
317 :class:`.DDLEvents`
319 :ref:`event_toplevel`
321 """
322 self._ddl_if = DDLIf(dialect, callable_, state)
323 return self
325 def _should_execute(self, target, bind, **kw):
326 if self._ddl_if is None:
327 return True
328 else:
329 return self._ddl_if._should_execute(self, target, bind, **kw)
331 def _invoke_with(self, bind):
332 if self._should_execute(self.target, bind):
333 return bind.execute(self)
335 def __call__(self, target, bind, **kw):
336 """Execute the DDL as a ddl_listener."""
338 self.against(target)._invoke_with(bind)
340 def _generate(self):
341 s = self.__class__.__new__(self.__class__)
342 s.__dict__ = self.__dict__.copy()
343 return s
346DDLElement = ExecutableDDLElement
347""":class:`.DDLElement` is renamed to :class:`.ExecutableDDLElement`."""
350class DDL(ExecutableDDLElement):
351 """A literal DDL statement.
353 Specifies literal SQL DDL to be executed by the database. DDL objects
354 function as DDL event listeners, and can be subscribed to those events
355 listed in :class:`.DDLEvents`, using either :class:`_schema.Table` or
356 :class:`_schema.MetaData` objects as targets.
357 Basic templating support allows
358 a single DDL instance to handle repetitive tasks for multiple tables.
360 Examples::
362 from sqlalchemy import event, DDL
364 tbl = Table("users", metadata, Column("uid", Integer))
365 event.listen(tbl, "before_create", DDL("DROP TRIGGER users_trigger"))
367 spow = DDL("ALTER TABLE %(table)s SET secretpowers TRUE")
368 event.listen(tbl, "after_create", spow.execute_if(dialect="somedb"))
370 drop_spow = DDL("ALTER TABLE users SET secretpowers FALSE")
371 connection.execute(drop_spow)
373 When operating on Table events, the following ``statement``
374 string substitutions are available:
376 .. sourcecode:: text
378 %(table)s - the Table name, with any required quoting applied
379 %(schema)s - the schema name, with any required quoting applied
380 %(fullname)s - the Table name including schema, quoted if needed
382 The DDL's "context", if any, will be combined with the standard
383 substitutions noted above. Keys present in the context will override
384 the standard substitutions.
386 """
388 __visit_name__ = "ddl"
390 def __init__(self, statement, context=None):
391 """Create a DDL statement.
393 :param statement:
394 A string or unicode string to be executed. Statements will be
395 processed with Python's string formatting operator using
396 a fixed set of string substitutions, as well as additional
397 substitutions provided by the optional :paramref:`.DDL.context`
398 parameter.
400 A literal '%' in a statement must be escaped as '%%'.
402 SQL bind parameters are not available in DDL statements.
404 :param context:
405 Optional dictionary, defaults to None. These values will be
406 available for use in string substitutions on the DDL statement.
408 .. seealso::
410 :class:`.DDLEvents`
412 :ref:`event_toplevel`
414 """
416 if not isinstance(statement, str):
417 raise exc.ArgumentError(
418 "Expected a string or unicode SQL statement, got '%r'"
419 % statement
420 )
422 self.statement = statement
423 self.context = context or {}
425 def __repr__(self):
426 parts = [repr(self.statement)]
427 if self.context:
428 parts.append(f"context={self.context}")
430 return "<%s@%s; %s>" % (
431 type(self).__name__,
432 id(self),
433 ", ".join(parts),
434 )
437class _CreateDropBase(ExecutableDDLElement, Generic[_SI]):
438 """Base class for DDL constructs that represent CREATE and DROP or
439 equivalents.
441 The common theme of _CreateDropBase is a single
442 ``element`` attribute which refers to the element
443 to be created or dropped.
445 """
447 element: _SI
449 def __init__(self, element: _SI) -> None:
450 self.element = self.target = element
451 self._ddl_if = getattr(element, "_ddl_if", None)
453 @property
454 def stringify_dialect(self): # type: ignore[override]
455 assert not isinstance(self.element, str)
456 return self.element.create_drop_stringify_dialect
458 def _create_rule_disable(self, compiler):
459 """Allow disable of _create_rule using a callable.
461 Pass to _create_rule using
462 util.portable_instancemethod(self._create_rule_disable)
463 to retain serializability.
465 """
466 return False
469class _CreateBase(_CreateDropBase[_SI]):
470 def __init__(self, element: _SI, if_not_exists: bool = False) -> None:
471 super().__init__(element)
472 self.if_not_exists = if_not_exists
475class TableCreateDDL(_CreateBase["Table"]):
477 def to_metadata(self, metadata: MetaData, table: Table) -> Self:
478 raise NotImplementedError()
481class _DropBase(_CreateDropBase[_SI]):
482 def __init__(self, element: _SI, if_exists: bool = False) -> None:
483 super().__init__(element)
484 self.if_exists = if_exists
487class TableDropDDL(_DropBase["Table"]):
489 def to_metadata(self, metadata: MetaData, table: Table) -> Self:
490 raise NotImplementedError()
493class CreateSchema(_CreateBase[str]):
494 """Represent a CREATE SCHEMA statement.
496 The argument here is the string name of the schema.
498 """
500 __visit_name__ = "create_schema"
502 stringify_dialect = "default"
504 def __init__(
505 self,
506 name: str,
507 if_not_exists: bool = False,
508 ) -> None:
509 """Create a new :class:`.CreateSchema` construct."""
511 super().__init__(element=name, if_not_exists=if_not_exists)
514class DropSchema(_DropBase[str]):
515 """Represent a DROP SCHEMA statement.
517 The argument here is the string name of the schema.
519 """
521 __visit_name__ = "drop_schema"
523 stringify_dialect = "default"
525 def __init__(
526 self,
527 name: str,
528 cascade: bool = False,
529 if_exists: bool = False,
530 ) -> None:
531 """Create a new :class:`.DropSchema` construct."""
533 super().__init__(element=name, if_exists=if_exists)
534 self.cascade = cascade
537class CreateTable(TableCreateDDL):
538 """Represent a CREATE TABLE statement."""
540 __visit_name__ = "create_table"
542 def __init__(
543 self,
544 element: Table,
545 include_foreign_key_constraints: Optional[
546 typing_Sequence[ForeignKeyConstraint]
547 ] = None,
548 if_not_exists: bool = False,
549 ) -> None:
550 """Create a :class:`.CreateTable` construct.
552 :param element: a :class:`_schema.Table` that's the subject
553 of the CREATE
554 :param on: See the description for 'on' in :class:`.DDL`.
555 :param include_foreign_key_constraints: optional sequence of
556 :class:`_schema.ForeignKeyConstraint` objects that will be included
557 inline within the CREATE construct; if omitted, all foreign key
558 constraints that do not specify use_alter=True are included.
560 :param if_not_exists: if True, an IF NOT EXISTS operator will be
561 applied to the construct.
563 .. versionadded:: 1.4.0b2
565 """
566 super().__init__(element, if_not_exists=if_not_exists)
567 self.columns = [CreateColumn(column) for column in element.columns]
568 self.include_foreign_key_constraints = include_foreign_key_constraints
570 def to_metadata(self, metadata: MetaData, table: Table) -> Self:
571 return self.__class__(table, if_not_exists=self.if_not_exists)
574class _TableViaSelect(TableCreateDDL, ExecutableDDLElement):
575 """Common base class for DDL constructs that generate and render for a
576 :class:`.Table` given a :class:`.Select`
578 .. versionadded:: 2.1
580 """
582 table: Table
583 """:class:`.Table` object representing the table that this
584 :class:`.CreateTableAs` would generate when executed."""
586 def __init__(
587 self,
588 selectable: SelectBase,
589 name: str,
590 *,
591 metadata: Optional["MetaData"] = None,
592 schema: Optional[str] = None,
593 temporary: bool = False,
594 if_not_exists: bool = False,
595 ):
596 # Coerce selectable to a Select statement
597 selectable = coercions.expect(roles.DMLSelectRole, selectable)
599 self.schema = schema
600 self.selectable = selectable
601 self.temporary = bool(temporary)
602 self.if_not_exists = bool(if_not_exists)
603 self.metadata = metadata
604 self.table_name = name
605 self._gen_table()
607 @property
608 def element(self): # type: ignore
609 return self.table
611 def to_metadata(self, metadata: MetaData, table: Table) -> Self:
612 new = self.__class__.__new__(self.__class__)
613 new.__dict__.update(self.__dict__)
614 new.metadata = metadata
615 new.table = table
616 return new
618 @util.preload_module("sqlalchemy.sql.schema")
619 def _gen_table(self) -> None:
620 MetaData = util.preloaded.sql_schema.MetaData
621 Column = util.preloaded.sql_schema.Column
622 Table = util.preloaded.sql_schema.Table
623 MetaData = util.preloaded.sql_schema.MetaData
625 column_name_type_pairs = (
626 (name, col_element.type)
627 for _, name, _, col_element, _ in (
628 self.selectable._generate_columns_plus_names(
629 anon_for_dupe_key=False
630 )
631 )
632 )
634 if self.metadata is None:
635 self.metadata = metadata = MetaData()
636 else:
637 metadata = self.metadata
639 self.table = Table(
640 self.table_name,
641 metadata,
642 *(Column(name, typ) for name, typ in column_name_type_pairs),
643 schema=self.schema,
644 _creator_ddl=self,
645 )
648class CreateTableAs(DialectKWArgs, _TableViaSelect):
649 """Represent a CREATE TABLE ... AS statement.
651 This creates a new table directly from the output of a SELECT, including
652 its schema and its initial set of data. Unlike a view, the
653 new table is fixed and does not synchronize further with the originating
654 SELECT statement.
656 The example below illustrates basic use of :class:`.CreateTableAs`; given a
657 :class:`.Select` and optional :class:`.MetaData`, the
658 :class:`.CreateTableAs` may be invoked directly via
659 :meth:`.Connection.execute` or indirectly via :meth:`.MetaData.create_all`;
660 the :attr:`.CreateTableAs.table` attribute provides a :class:`.Table`
661 object with which to generate new queries::
663 from sqlalchemy import CreateTableAs
664 from sqlalchemy import select
666 # instantiate CreateTableAs given a select() and optional MetaData
667 cas = CreateTableAs(
668 select(users.c.id, users.c.name).where(users.c.status == "active"),
669 "active_users",
670 metadata=some_metadata,
671 )
673 # a Table object is available immediately via the .table attribute
674 new_statement = select(cas.table)
676 # to emit CREATE TABLE AS, either invoke CreateTableAs directly...
677 with engine.begin() as conn:
678 conn.execute(cas)
680 # or alternatively, invoke metadata.create_all()
681 some_metdata.create_all(engine)
683 # drop is performed in the usual way, via drop_all
684 # or table.drop()
685 some_metdata.drop_all(engine)
687 For detailed background on :class:`.CreateTableAs` see
688 :ref:`metadata_create_table_as`.
690 .. versionadded:: 2.1
692 :param selectable: :class:`_sql.Select`
693 The SELECT statement providing the columns and rows.
695 :param table_name: table name as a string. Combine with the optional
696 :paramref:`.CreateTableAs.schema` parameter to indicate a
697 schema-qualified table name.
699 :param metadata: :class:`_schema.MetaData`, optional
700 If provided, the :class:`_schema.Table` object available via the
701 :attr:`.table` attribute will be associated with this
702 :class:`.MetaData`. Otherwise, a new, empty :class:`.MetaData`
703 is created.
705 :param schema: str, optional schema or owner name.
707 :param temporary: bool, default False.
708 If True, render ``TEMPORARY``
710 :param if_not_exists: bool, default False.
711 If True, render ``IF NOT EXISTS``
713 .. seealso::
715 :ref:`metadata_create_table_as` - in :ref:`metadata_toplevel`
717 :meth:`_sql.SelectBase.into` - convenience method to create a
718 :class:`_schema.CreateTableAs` from a SELECT statement
720 :class:`.CreateView`
723 """
725 __visit_name__ = "create_table_as"
726 inherit_cache = False
728 table: Table
729 """:class:`.Table` object representing the table that this
730 :class:`.CreateTableAs` would generate when executed."""
732 def __init__(
733 self,
734 selectable: SelectBase,
735 table_name: str,
736 *,
737 metadata: Optional["MetaData"] = None,
738 schema: Optional[str] = None,
739 temporary: bool = False,
740 if_not_exists: bool = False,
741 **dialect_kwargs: Any,
742 ):
743 self._validate_dialect_kwargs(dialect_kwargs)
744 super().__init__(
745 selectable=selectable,
746 name=table_name,
747 metadata=metadata,
748 schema=schema,
749 temporary=temporary,
750 if_not_exists=if_not_exists,
751 )
754class CreateView(DialectKWArgs, _TableViaSelect):
755 """Represent a CREATE VIEW statement.
757 This creates a new view based on a particular SELECT statement. The schema
758 of the view is based on the columns of the SELECT statement, and the data
759 present in the view is derived from the rows represented by the
760 SELECT. A non-materialized view will evaluate the SELECT statement
761 dynamicaly as it is queried, whereas a materialized view represents a
762 snapshot of the SELECT statement at a particular point in time and
763 typically needs to be refreshed manually using database-specific commands.
765 The example below illustrates basic use of :class:`.CreateView`; given a
766 :class:`.Select` and optional :class:`.MetaData`, the
767 :class:`.CreateView` may be invoked directly via
768 :meth:`.Connection.execute` or indirectly via :meth:`.MetaData.create_all`;
769 the :attr:`.CreateView.table` attribute provides a :class:`.Table`
770 object with which to generate new queries::
773 from sqlalchemy import select
774 from sqlalchemy.sql.ddl import CreateView
776 # instantiate CreateView given a select() and optional MetaData
777 create_view = CreateView(
778 select(users.c.id, users.c.name).where(users.c.status == "active"),
779 "active_users_view",
780 metadata=some_metadata,
781 )
783 # a Table object is available immediately via the .table attribute
784 new_statement = select(create_view.table)
786 # to emit CREATE VIEW, either invoke CreateView directly...
787 with engine.begin() as conn:
788 conn.execute(create_view)
790 # or alternatively, invoke metadata.create_all()
791 some_metdata.create_all(engine)
793 # drop is performed in the usual way, via drop_all
794 # or table.drop() (will emit DROP VIEW)
795 some_metdata.drop_all(engine)
797 For detailed background on :class:`.CreateView` see
798 :ref:`metadata_create_view`.
800 .. versionadded:: 2.1
802 :param selectable: :class:`_sql.Select`
803 The SELECT statement defining the view.
805 :param view_name: table name as a string. Combine with the optional
806 :paramref:`.CreateView.schema` parameter to indicate a
807 schema-qualified table name.
809 :param metadata: :class:`_schema.MetaData`, optional
810 If provided, the :class:`_schema.Table` object available via the
811 :attr:`.table` attribute will be associated with this
812 :class:`.MetaData`. Otherwise, a new, empty :class:`.MetaData`
813 is created.
815 :param schema: str, optional schema or owner name.
817 :param temporary: bool, default False.
818 If True, render ``TEMPORARY``
820 :param or_replace: bool, default False.
821 If True, render ``OR REPLACE`` to replace an existing view if it
822 exists. Supported by PostgreSQL, MySQL, MariaDB, and Oracle.
823 Not supported by SQLite or SQL Server.
825 .. versionadded:: 2.1
827 :param materialized: bool, default False.
828 If True, render ``MATERIALIZED`` to create a materialized view.
829 Materialized views store the query results physically and can be
830 refreshed periodically. Not supported by all database backends.
832 .. versionadded:: 2.1
834 :param dialect_kw: Additional keyword arguments are dialect-specific and
835 are passed as keyword arguments to the dialect's compiler.
837 .. note::
839 For SQLite, the ``sqlite_if_not_exists`` boolean parameter
840 is supported to render ``CREATE VIEW IF NOT EXISTS``.
842 .. versionadded:: 2.1
844 .. seealso::
846 :ref:`metadata_create_view` - in :ref:`metadata_toplevel`
848 :class:`.CreateTableAs` - for creating a table from a SELECT statement
850 """
852 __visit_name__ = "create_view"
854 inherit_cache = False
856 table: Table
857 """:class:`.Table` object representing the view that this
858 :class:`.CreateView` would generate when executed."""
860 materialized: bool
861 """Boolean flag indicating if this is a materialized view."""
863 or_replace: bool
864 """Boolean flag indicating if OR REPLACE should be used."""
866 def __init__(
867 self,
868 selectable: SelectBase,
869 view_name: str,
870 *,
871 metadata: Optional["MetaData"] = None,
872 schema: Optional[str] = None,
873 temporary: bool = False,
874 or_replace: bool = False,
875 materialized: bool = False,
876 **dialect_kwargs: Any,
877 ):
878 self._validate_dialect_kwargs(dialect_kwargs)
879 super().__init__(
880 selectable=selectable,
881 name=view_name,
882 metadata=metadata,
883 schema=schema,
884 temporary=temporary,
885 if_not_exists=False,
886 )
887 self.materialized = materialized
888 self.or_replace = or_replace
889 self.table._dropper_ddl = DropView(
890 self.table, materialized=materialized
891 )
894class DropView(TableDropDDL):
895 """'DROP VIEW' construct.
897 .. versionadded:: 2.1 the :class:`.DropView` construct became public
898 and was renamed from ``_DropView``.
900 """
902 __visit_name__ = "drop_view"
904 materialized: bool
905 """Boolean flag indicating if this is a materialized view."""
907 def __init__(
908 self,
909 element: Table,
910 *,
911 if_exists: bool = False,
912 materialized: bool = False,
913 ) -> None:
914 super().__init__(element, if_exists=if_exists)
915 self.materialized = materialized
917 def to_metadata(self, metadata: MetaData, table: Table) -> Self:
918 new = self.__class__.__new__(self.__class__)
919 new.__dict__.update(self.__dict__)
920 new.element = table
921 return new
924class CreateConstraint(BaseDDLElement):
925 element: Constraint
927 def __init__(self, element: Constraint) -> None:
928 self.element = element
931class CreateColumn(BaseDDLElement):
932 """Represent a :class:`_schema.Column`
933 as rendered in a CREATE TABLE statement,
934 via the :class:`.CreateTable` construct.
936 This is provided to support custom column DDL within the generation
937 of CREATE TABLE statements, by using the
938 compiler extension documented in :ref:`sqlalchemy.ext.compiler_toplevel`
939 to extend :class:`.CreateColumn`.
941 Typical integration is to examine the incoming :class:`_schema.Column`
942 object, and to redirect compilation if a particular flag or condition
943 is found::
945 from sqlalchemy import schema
946 from sqlalchemy.ext.compiler import compiles
949 @compiles(schema.CreateColumn)
950 def compile(element, compiler, **kw):
951 column = element.element
953 if "special" not in column.info:
954 return compiler.visit_create_column(element, **kw)
956 text = "%s SPECIAL DIRECTIVE %s" % (
957 column.name,
958 compiler.type_compiler.process(column.type),
959 )
960 default = compiler.get_column_default_string(column)
961 if default is not None:
962 text += " DEFAULT " + default
964 if not column.nullable:
965 text += " NOT NULL"
967 if column.constraints:
968 text += " ".join(
969 compiler.process(const) for const in column.constraints
970 )
971 return text
973 The above construct can be applied to a :class:`_schema.Table`
974 as follows::
976 from sqlalchemy import Table, Metadata, Column, Integer, String
977 from sqlalchemy import schema
979 metadata = MetaData()
981 table = Table(
982 "mytable",
983 MetaData(),
984 Column("x", Integer, info={"special": True}, primary_key=True),
985 Column("y", String(50)),
986 Column("z", String(20), info={"special": True}),
987 )
989 metadata.create_all(conn)
991 Above, the directives we've added to the :attr:`_schema.Column.info`
992 collection
993 will be detected by our custom compilation scheme:
995 .. sourcecode:: sql
997 CREATE TABLE mytable (
998 x SPECIAL DIRECTIVE INTEGER NOT NULL,
999 y VARCHAR(50),
1000 z SPECIAL DIRECTIVE VARCHAR(20),
1001 PRIMARY KEY (x)
1002 )
1004 The :class:`.CreateColumn` construct can also be used to skip certain
1005 columns when producing a ``CREATE TABLE``. This is accomplished by
1006 creating a compilation rule that conditionally returns ``None``.
1007 This is essentially how to produce the same effect as using the
1008 ``system=True`` argument on :class:`_schema.Column`, which marks a column
1009 as an implicitly-present "system" column.
1011 For example, suppose we wish to produce a :class:`_schema.Table`
1012 which skips
1013 rendering of the PostgreSQL ``xmin`` column against the PostgreSQL
1014 backend, but on other backends does render it, in anticipation of a
1015 triggered rule. A conditional compilation rule could skip this name only
1016 on PostgreSQL::
1018 from sqlalchemy.schema import CreateColumn
1021 @compiles(CreateColumn, "postgresql")
1022 def skip_xmin(element, compiler, **kw):
1023 if element.element.name == "xmin":
1024 return None
1025 else:
1026 return compiler.visit_create_column(element, **kw)
1029 my_table = Table(
1030 "mytable",
1031 metadata,
1032 Column("id", Integer, primary_key=True),
1033 Column("xmin", Integer),
1034 )
1036 Above, a :class:`.CreateTable` construct will generate a ``CREATE TABLE``
1037 which only includes the ``id`` column in the string; the ``xmin`` column
1038 will be omitted, but only against the PostgreSQL backend.
1040 """
1042 __visit_name__ = "create_column"
1044 element: Column[Any]
1046 def __init__(self, element: Column[Any]) -> None:
1047 self.element = element
1050class DropTable(TableDropDDL):
1051 """Represent a DROP TABLE statement."""
1053 __visit_name__ = "drop_table"
1055 def __init__(self, element: Table, if_exists: bool = False) -> None:
1056 """Create a :class:`.DropTable` construct.
1058 :param element: a :class:`_schema.Table` that's the subject
1059 of the DROP.
1060 :param on: See the description for 'on' in :class:`.DDL`.
1061 :param if_exists: if True, an IF EXISTS operator will be applied to the
1062 construct.
1064 .. versionadded:: 1.4.0b2
1066 """
1067 super().__init__(element, if_exists=if_exists)
1069 def to_metadata(self, metadata: MetaData, table: Table) -> Self:
1070 return self.__class__(table, if_exists=self.if_exists)
1073class CreateSequence(_CreateBase["Sequence"]):
1074 """Represent a CREATE SEQUENCE statement."""
1076 __visit_name__ = "create_sequence"
1079class DropSequence(_DropBase["Sequence"]):
1080 """Represent a DROP SEQUENCE statement."""
1082 __visit_name__ = "drop_sequence"
1085class CreateIndex(_CreateBase["Index"]):
1086 """Represent a CREATE INDEX statement."""
1088 __visit_name__ = "create_index"
1090 def __init__(self, element: Index, if_not_exists: bool = False) -> None:
1091 """Create a :class:`.Createindex` construct.
1093 :param element: a :class:`_schema.Index` that's the subject
1094 of the CREATE.
1095 :param if_not_exists: if True, an IF NOT EXISTS operator will be
1096 applied to the construct.
1098 .. versionadded:: 1.4.0b2
1100 """
1101 super().__init__(element, if_not_exists=if_not_exists)
1104class DropIndex(_DropBase["Index"]):
1105 """Represent a DROP INDEX statement."""
1107 __visit_name__ = "drop_index"
1109 def __init__(self, element: Index, if_exists: bool = False) -> None:
1110 """Create a :class:`.DropIndex` construct.
1112 :param element: a :class:`_schema.Index` that's the subject
1113 of the DROP.
1114 :param if_exists: if True, an IF EXISTS operator will be applied to the
1115 construct.
1117 .. versionadded:: 1.4.0b2
1119 """
1120 super().__init__(element, if_exists=if_exists)
1123class AddConstraint(_CreateBase["Constraint"]):
1124 """Represent an ALTER TABLE ADD CONSTRAINT statement."""
1126 __visit_name__ = "add_constraint"
1128 def __init__(
1129 self,
1130 element: Constraint,
1131 *,
1132 isolate_from_table: bool = True,
1133 ) -> None:
1134 """Construct a new :class:`.AddConstraint` construct.
1136 :param element: a :class:`.Constraint` object
1138 :param isolate_from_table: optional boolean, defaults to True. Has
1139 the effect of the incoming constraint being isolated from being
1140 included in a CREATE TABLE sequence when associated with a
1141 :class:`.Table`.
1143 .. versionadded:: 2.0.39 - added
1144 :paramref:`.AddConstraint.isolate_from_table`, defaulting
1145 to True. Previously, the behavior of this parameter was implicitly
1146 turned on in all cases.
1148 """
1149 super().__init__(element)
1151 if isolate_from_table:
1152 element._create_rule = self._create_rule_disable
1155class DropConstraint(_DropBase["Constraint"]):
1156 """Represent an ALTER TABLE DROP CONSTRAINT statement."""
1158 __visit_name__ = "drop_constraint"
1160 def __init__(
1161 self,
1162 element: Constraint,
1163 *,
1164 cascade: bool = False,
1165 if_exists: bool = False,
1166 isolate_from_table: bool = True,
1167 **kw: Any,
1168 ) -> None:
1169 """Construct a new :class:`.DropConstraint` construct.
1171 :param element: a :class:`.Constraint` object
1172 :param cascade: optional boolean, indicates backend-specific
1173 "CASCADE CONSTRAINT" directive should be rendered if available
1174 :param if_exists: optional boolean, indicates backend-specific
1175 "IF EXISTS" directive should be rendered if available
1176 :param isolate_from_table: optional boolean, defaults to True. Has
1177 the effect of the incoming constraint being isolated from being
1178 included in a CREATE TABLE sequence when associated with a
1179 :class:`.Table`.
1181 .. versionadded:: 2.0.39 - added
1182 :paramref:`.DropConstraint.isolate_from_table`, defaulting
1183 to True. Previously, the behavior of this parameter was implicitly
1184 turned on in all cases.
1186 """
1187 self.cascade = cascade
1188 super().__init__(element, if_exists=if_exists, **kw)
1190 if isolate_from_table:
1191 element._create_rule = self._create_rule_disable
1194class SetTableComment(_CreateDropBase["Table"]):
1195 """Represent a COMMENT ON TABLE IS statement."""
1197 __visit_name__ = "set_table_comment"
1200class DropTableComment(_CreateDropBase["Table"]):
1201 """Represent a COMMENT ON TABLE '' statement.
1203 Note this varies a lot across database backends.
1205 """
1207 __visit_name__ = "drop_table_comment"
1210class SetColumnComment(_CreateDropBase["Column[Any]"]):
1211 """Represent a COMMENT ON COLUMN IS statement."""
1213 __visit_name__ = "set_column_comment"
1216class DropColumnComment(_CreateDropBase["Column[Any]"]):
1217 """Represent a COMMENT ON COLUMN IS NULL statement."""
1219 __visit_name__ = "drop_column_comment"
1222class SetConstraintComment(_CreateDropBase["Constraint"]):
1223 """Represent a COMMENT ON CONSTRAINT IS statement."""
1225 __visit_name__ = "set_constraint_comment"
1228class DropConstraintComment(_CreateDropBase["Constraint"]):
1229 """Represent a COMMENT ON CONSTRAINT IS NULL statement."""
1231 __visit_name__ = "drop_constraint_comment"
1234class InvokeDDLBase(SchemaVisitor):
1235 def __init__(self, connection, **kw):
1236 self.connection = connection
1237 assert not kw, f"Unexpected keywords: {kw.keys()}"
1239 @contextlib.contextmanager
1240 def with_ddl_events(self, target, **kw):
1241 """helper context manager that will apply appropriate DDL events
1242 to a CREATE or DROP operation."""
1244 raise NotImplementedError()
1247class InvokeCreateDDLBase(InvokeDDLBase):
1248 @contextlib.contextmanager
1249 def with_ddl_events(self, target, **kw):
1250 """helper context manager that will apply appropriate DDL events
1251 to a CREATE or DROP operation."""
1253 target.dispatch.before_create(
1254 target, self.connection, _ddl_runner=self, **kw
1255 )
1256 yield
1257 target.dispatch.after_create(
1258 target, self.connection, _ddl_runner=self, **kw
1259 )
1262class InvokeDropDDLBase(InvokeDDLBase):
1263 @contextlib.contextmanager
1264 def with_ddl_events(self, target, **kw):
1265 """helper context manager that will apply appropriate DDL events
1266 to a CREATE or DROP operation."""
1268 target.dispatch.before_drop(
1269 target, self.connection, _ddl_runner=self, **kw
1270 )
1271 yield
1272 target.dispatch.after_drop(
1273 target, self.connection, _ddl_runner=self, **kw
1274 )
1277class CheckFirst(Flag):
1278 """Enumeration for the :paramref:`.MetaData.create_all.checkfirst`
1279 parameter passed to methods like :meth:`.MetaData.create_all`,
1280 :meth:`.MetaData.drop_all`, :meth:`.Table.create`, :meth:`.Table.drop` and
1281 others.
1283 This enumeration indicates what kinds of objects should be "checked"
1284 with a separate query before emitting CREATE or DROP for that object.
1286 Can use ``CheckFirst(bool_value)`` to convert from a boolean value.
1288 .. versionadded:: 2.1
1290 """
1292 NONE = 0 # equivalent to False
1293 """No items should be checked"""
1295 # avoid 1 so that bool True doesn't match by value
1296 TABLES = 2
1297 """Check for tables"""
1299 VIEWS = auto()
1300 """Check for views"""
1302 INDEXES = auto()
1303 """Check for indexes"""
1305 SEQUENCES = auto()
1306 """Check for sequences"""
1308 TYPES = auto()
1309 """Check for custom datatypes that are created server-side
1311 This is currently used by PostgreSQL.
1313 """
1315 ALL = TABLES | VIEWS | INDEXES | SEQUENCES | TYPES # equivalent to True
1317 @classmethod
1318 def _missing_(cls, value: object) -> Any:
1319 if isinstance(value, bool):
1320 return cls.ALL if value else cls.NONE
1321 return super()._missing_(value)
1324class SchemaGenerator(InvokeCreateDDLBase):
1325 def __init__(
1326 self,
1327 dialect,
1328 connection,
1329 checkfirst=CheckFirst.NONE,
1330 tables=None,
1331 **kwargs,
1332 ):
1333 super().__init__(connection, **kwargs)
1334 self.checkfirst = CheckFirst(checkfirst)
1335 self.tables = tables
1336 self.preparer = dialect.identifier_preparer
1337 self.dialect = dialect
1338 self.memo = {}
1340 def _can_create_table(self, table):
1341 self.dialect.validate_identifier(table.name)
1342 effective_schema = self.connection.schema_for_object(table)
1343 if effective_schema:
1344 self.dialect.validate_identifier(effective_schema)
1346 bool_to_check = (
1347 CheckFirst.TABLES if not table.is_view else CheckFirst.VIEWS
1348 )
1349 return (
1350 not self.checkfirst & bool_to_check
1351 or not self.dialect.has_table(
1352 self.connection, table.name, schema=effective_schema
1353 )
1354 )
1356 def _can_create_index(self, index):
1357 effective_schema = self.connection.schema_for_object(index.table)
1358 if effective_schema:
1359 self.dialect.validate_identifier(effective_schema)
1360 return (
1361 not self.checkfirst & CheckFirst.INDEXES
1362 or not self.dialect.has_index(
1363 self.connection,
1364 index.table.name,
1365 index.name,
1366 schema=effective_schema,
1367 )
1368 )
1370 def _can_create_sequence(self, sequence):
1371 effective_schema = self.connection.schema_for_object(sequence)
1373 return self.dialect.supports_sequences and (
1374 (not self.dialect.sequences_optional or not sequence.optional)
1375 and (
1376 not self.checkfirst & CheckFirst.SEQUENCES
1377 or not self.dialect.has_sequence(
1378 self.connection, sequence.name, schema=effective_schema
1379 )
1380 )
1381 )
1383 def visit_metadata(self, metadata):
1384 if self.tables is not None:
1385 tables = self.tables
1386 else:
1387 tables = list(metadata.tables.values())
1389 collection = sort_tables_and_constraints(
1390 [t for t in tables if self._can_create_table(t)]
1391 )
1393 seq_coll = [
1394 s
1395 for s in metadata._sequences.values()
1396 if s.column is None and self._can_create_sequence(s)
1397 ]
1399 event_collection = [t for (t, fks) in collection if t is not None]
1401 with self.with_ddl_events(
1402 metadata,
1403 tables=event_collection,
1404 checkfirst=self.checkfirst,
1405 ):
1406 for seq in seq_coll:
1407 self.traverse_single(seq, create_ok=True)
1409 for table, fkcs in collection:
1410 if table is not None:
1411 self.traverse_single(
1412 table,
1413 create_ok=True,
1414 include_foreign_key_constraints=fkcs,
1415 _is_metadata_operation=True,
1416 )
1417 else:
1418 for fkc in fkcs:
1419 self.traverse_single(fkc)
1421 def visit_table(
1422 self,
1423 table,
1424 create_ok=False,
1425 include_foreign_key_constraints=None,
1426 _is_metadata_operation=False,
1427 ):
1428 if not create_ok and not self._can_create_table(table):
1429 return
1431 with self.with_ddl_events(
1432 table,
1433 checkfirst=self.checkfirst,
1434 _is_metadata_operation=_is_metadata_operation,
1435 ):
1436 for column in table.columns:
1437 if column.default is not None:
1438 self.traverse_single(column.default)
1440 if not self.dialect.supports_alter:
1441 # e.g., don't omit any foreign key constraints
1442 include_foreign_key_constraints = None
1444 if table._creator_ddl is not None:
1445 table_create_ddl = table._creator_ddl
1446 else:
1447 table_create_ddl = CreateTable(
1448 table,
1449 include_foreign_key_constraints=(
1450 include_foreign_key_constraints
1451 ),
1452 )
1454 table_create_ddl._invoke_with(self.connection)
1456 if hasattr(table, "indexes"):
1457 for index in table.indexes:
1458 self.traverse_single(index, create_ok=True)
1460 if (
1461 self.dialect.supports_comments
1462 and not self.dialect.inline_comments
1463 ):
1464 if table.comment is not None:
1465 SetTableComment(table)._invoke_with(self.connection)
1467 for column in table.columns:
1468 if column.comment is not None:
1469 SetColumnComment(column)._invoke_with(self.connection)
1471 if self.dialect.supports_constraint_comments:
1472 for constraint in table.constraints:
1473 if constraint.comment is not None:
1474 self.connection.execute(
1475 SetConstraintComment(constraint)
1476 )
1478 def visit_foreign_key_constraint(self, constraint):
1479 if not self.dialect.supports_alter:
1480 return
1482 with self.with_ddl_events(constraint):
1483 AddConstraint(constraint)._invoke_with(self.connection)
1485 def visit_sequence(self, sequence, create_ok=False):
1486 if not create_ok and not self._can_create_sequence(sequence):
1487 return
1488 with self.with_ddl_events(sequence):
1489 CreateSequence(sequence)._invoke_with(self.connection)
1491 def visit_index(self, index, create_ok=False):
1492 if not create_ok and not self._can_create_index(index):
1493 return
1494 with self.with_ddl_events(index):
1495 CreateIndex(index)._invoke_with(self.connection)
1498class SchemaDropper(InvokeDropDDLBase):
1499 def __init__(
1500 self,
1501 dialect,
1502 connection,
1503 checkfirst=CheckFirst.NONE,
1504 tables=None,
1505 **kwargs,
1506 ):
1507 super().__init__(connection, **kwargs)
1508 self.checkfirst = CheckFirst(checkfirst)
1509 self.tables = tables
1510 self.preparer = dialect.identifier_preparer
1511 self.dialect = dialect
1512 self.memo = {}
1514 def visit_metadata(self, metadata):
1515 if self.tables is not None:
1516 tables = self.tables
1517 else:
1518 tables = list(metadata.tables.values())
1520 try:
1521 unsorted_tables = [t for t in tables if self._can_drop_table(t)]
1522 collection = list(
1523 reversed(
1524 sort_tables_and_constraints(
1525 unsorted_tables,
1526 filter_fn=lambda constraint: (
1527 False
1528 if not self.dialect.supports_alter
1529 or constraint.name is None
1530 else None
1531 ),
1532 )
1533 )
1534 )
1535 except exc.CircularDependencyError as err2:
1536 if not self.dialect.supports_alter:
1537 util.warn(
1538 "Can't sort tables for DROP; an "
1539 "unresolvable foreign key "
1540 "dependency exists between tables: %s; and backend does "
1541 "not support ALTER. To restore at least a partial sort, "
1542 "apply use_alter=True to ForeignKey and "
1543 "ForeignKeyConstraint "
1544 "objects involved in the cycle to mark these as known "
1545 "cycles that will be ignored."
1546 % (", ".join(sorted([t.fullname for t in err2.cycles])))
1547 )
1548 collection = [(t, ()) for t in unsorted_tables]
1549 else:
1550 raise exc.CircularDependencyError(
1551 err2.args[0],
1552 err2.cycles,
1553 err2.edges,
1554 msg="Can't sort tables for DROP; an "
1555 "unresolvable foreign key "
1556 "dependency exists between tables: %s. Please ensure "
1557 "that the ForeignKey and ForeignKeyConstraint objects "
1558 "involved in the cycle have "
1559 "names so that they can be dropped using "
1560 "DROP CONSTRAINT."
1561 % (", ".join(sorted([t.fullname for t in err2.cycles]))),
1562 ) from err2
1564 seq_coll = [
1565 s
1566 for s in metadata._sequences.values()
1567 if self._can_drop_sequence(s)
1568 ]
1570 event_collection = [t for (t, fks) in collection if t is not None]
1572 with self.with_ddl_events(
1573 metadata,
1574 tables=event_collection,
1575 checkfirst=self.checkfirst,
1576 ):
1577 for table, fkcs in collection:
1578 if table is not None:
1579 self.traverse_single(
1580 table,
1581 drop_ok=True,
1582 _is_metadata_operation=True,
1583 _ignore_sequences=seq_coll,
1584 )
1585 else:
1586 for fkc in fkcs:
1587 self.traverse_single(fkc)
1589 for seq in seq_coll:
1590 self.traverse_single(seq, drop_ok=seq.column is None)
1592 def _can_drop_table(self, table):
1593 self.dialect.validate_identifier(table.name)
1594 effective_schema = self.connection.schema_for_object(table)
1595 if effective_schema:
1596 self.dialect.validate_identifier(effective_schema)
1597 bool_to_check = (
1598 CheckFirst.TABLES if not table.is_view else CheckFirst.VIEWS
1599 )
1601 return not self.checkfirst & bool_to_check or self.dialect.has_table(
1602 self.connection, table.name, schema=effective_schema
1603 )
1605 def _can_drop_index(self, index):
1606 effective_schema = self.connection.schema_for_object(index.table)
1607 if effective_schema:
1608 self.dialect.validate_identifier(effective_schema)
1609 return (
1610 not self.checkfirst & CheckFirst.INDEXES
1611 or self.dialect.has_index(
1612 self.connection,
1613 index.table.name,
1614 index.name,
1615 schema=effective_schema,
1616 )
1617 )
1619 def _can_drop_sequence(self, sequence):
1620 effective_schema = self.connection.schema_for_object(sequence)
1621 return self.dialect.supports_sequences and (
1622 (not self.dialect.sequences_optional or not sequence.optional)
1623 and (
1624 not self.checkfirst & CheckFirst.SEQUENCES
1625 or self.dialect.has_sequence(
1626 self.connection, sequence.name, schema=effective_schema
1627 )
1628 )
1629 )
1631 def visit_index(self, index, drop_ok=False):
1632 if not drop_ok and not self._can_drop_index(index):
1633 return
1635 with self.with_ddl_events(index):
1636 DropIndex(index)(index, self.connection)
1638 def visit_table(
1639 self,
1640 table,
1641 drop_ok=False,
1642 _is_metadata_operation=False,
1643 _ignore_sequences=(),
1644 ):
1645 if not drop_ok and not self._can_drop_table(table):
1646 return
1648 with self.with_ddl_events(
1649 table,
1650 checkfirst=self.checkfirst,
1651 _is_metadata_operation=_is_metadata_operation,
1652 ):
1653 if table._dropper_ddl is not None:
1654 table_dropper_ddl = table._dropper_ddl
1655 else:
1656 table_dropper_ddl = DropTable(table)
1657 table_dropper_ddl._invoke_with(self.connection)
1659 # traverse client side defaults which may refer to server-side
1660 # sequences. noting that some of these client side defaults may
1661 # also be set up as server side defaults
1662 # (see https://docs.sqlalchemy.org/en/
1663 # latest/core/defaults.html
1664 # #associating-a-sequence-as-the-server-side-
1665 # default), so have to be dropped after the table is dropped.
1666 for column in table.columns:
1667 if (
1668 column.default is not None
1669 and column.default not in _ignore_sequences
1670 ):
1671 self.traverse_single(column.default)
1673 def visit_foreign_key_constraint(self, constraint):
1674 if not self.dialect.supports_alter:
1675 return
1676 with self.with_ddl_events(constraint):
1677 DropConstraint(constraint)._invoke_with(self.connection)
1679 def visit_sequence(self, sequence, drop_ok=False):
1680 if not drop_ok and not self._can_drop_sequence(sequence):
1681 return
1682 with self.with_ddl_events(sequence):
1683 DropSequence(sequence)._invoke_with(self.connection)
1686def sort_tables(
1687 tables: Iterable[TableClause],
1688 skip_fn: Optional[Callable[[ForeignKeyConstraint], bool]] = None,
1689 extra_dependencies: Optional[
1690 typing_Sequence[Tuple[TableClause, TableClause]]
1691 ] = None,
1692) -> List[Table]:
1693 """Sort a collection of :class:`_schema.Table` objects based on
1694 dependency.
1696 This is a dependency-ordered sort which will emit :class:`_schema.Table`
1697 objects such that they will follow their dependent :class:`_schema.Table`
1698 objects.
1699 Tables are dependent on another based on the presence of
1700 :class:`_schema.ForeignKeyConstraint`
1701 objects as well as explicit dependencies
1702 added by :meth:`_schema.Table.add_is_dependent_on`.
1704 .. warning::
1706 The :func:`._schema.sort_tables` function cannot by itself
1707 accommodate automatic resolution of dependency cycles between
1708 tables, which are usually caused by mutually dependent foreign key
1709 constraints. When these cycles are detected, the foreign keys
1710 of these tables are omitted from consideration in the sort.
1711 A warning is emitted when this condition occurs, which will be an
1712 exception raise in a future release. Tables which are not part
1713 of the cycle will still be returned in dependency order.
1715 To resolve these cycles, the
1716 :paramref:`_schema.ForeignKeyConstraint.use_alter` parameter may be
1717 applied to those constraints which create a cycle. Alternatively,
1718 the :func:`_schema.sort_tables_and_constraints` function will
1719 automatically return foreign key constraints in a separate
1720 collection when cycles are detected so that they may be applied
1721 to a schema separately.
1723 :param tables: a sequence of :class:`_schema.Table` objects.
1725 :param skip_fn: optional callable which will be passed a
1726 :class:`_schema.ForeignKeyConstraint` object; if it returns True, this
1727 constraint will not be considered as a dependency. Note this is
1728 **different** from the same parameter in
1729 :func:`.sort_tables_and_constraints`, which is
1730 instead passed the owning :class:`_schema.ForeignKeyConstraint` object.
1732 :param extra_dependencies: a sequence of 2-tuples of tables which will
1733 also be considered as dependent on each other.
1735 .. seealso::
1737 :func:`.sort_tables_and_constraints`
1739 :attr:`_schema.MetaData.sorted_tables` - uses this function to sort
1742 """
1744 if skip_fn is not None:
1745 fixed_skip_fn = skip_fn
1747 def _skip_fn(fkc):
1748 for fk in fkc.elements:
1749 if fixed_skip_fn(fk):
1750 return True
1751 else:
1752 return None
1754 else:
1755 _skip_fn = None # type: ignore
1757 return [
1758 t
1759 for (t, fkcs) in sort_tables_and_constraints(
1760 tables,
1761 filter_fn=_skip_fn,
1762 extra_dependencies=extra_dependencies,
1763 _warn_for_cycles=True,
1764 )
1765 if t is not None
1766 ]
1769@util.preload_module("sqlalchemy.sql.schema")
1770def sort_tables_and_constraints(
1771 tables, filter_fn=None, extra_dependencies=None, _warn_for_cycles=False
1772):
1773 """Sort a collection of :class:`_schema.Table` /
1774 :class:`_schema.ForeignKeyConstraint`
1775 objects.
1777 This is a dependency-ordered sort which will emit tuples of
1778 ``(Table, [ForeignKeyConstraint, ...])`` such that each
1779 :class:`_schema.Table` follows its dependent :class:`_schema.Table`
1780 objects.
1781 Remaining :class:`_schema.ForeignKeyConstraint`
1782 objects that are separate due to
1783 dependency rules not satisfied by the sort are emitted afterwards
1784 as ``(None, [ForeignKeyConstraint ...])``.
1786 Tables are dependent on another based on the presence of
1787 :class:`_schema.ForeignKeyConstraint` objects, explicit dependencies
1788 added by :meth:`_schema.Table.add_is_dependent_on`,
1789 as well as dependencies
1790 stated here using the :paramref:`~.sort_tables_and_constraints.skip_fn`
1791 and/or :paramref:`~.sort_tables_and_constraints.extra_dependencies`
1792 parameters.
1794 :param tables: a sequence of :class:`_schema.Table` objects.
1796 :param filter_fn: optional callable which will be passed a
1797 :class:`_schema.ForeignKeyConstraint` object,
1798 and returns a value based on
1799 whether this constraint should definitely be included or excluded as
1800 an inline constraint, or neither. If it returns False, the constraint
1801 will definitely be included as a dependency that cannot be subject
1802 to ALTER; if True, it will **only** be included as an ALTER result at
1803 the end. Returning None means the constraint is included in the
1804 table-based result unless it is detected as part of a dependency cycle.
1806 :param extra_dependencies: a sequence of 2-tuples of tables which will
1807 also be considered as dependent on each other.
1809 .. seealso::
1811 :func:`.sort_tables`
1814 """
1815 Table = util.preloaded.sql_schema.Table
1817 fixed_dependencies = set()
1818 mutable_dependencies = set()
1820 if extra_dependencies is not None:
1821 fixed_dependencies.update(extra_dependencies)
1823 remaining_fkcs = set()
1824 for table in tables:
1825 for fkc in table.foreign_key_constraints:
1826 if fkc.use_alter is True:
1827 remaining_fkcs.add(fkc)
1828 continue
1830 if filter_fn:
1831 filtered = filter_fn(fkc)
1833 if filtered is True:
1834 remaining_fkcs.add(fkc)
1835 continue
1837 dependent_on = fkc.referred_table
1838 if dependent_on is not table:
1839 mutable_dependencies.add((dependent_on, table))
1841 if isinstance(table._creator_ddl, _TableViaSelect):
1842 selectable = table._creator_ddl.selectable
1843 for selected_table in sql_util.find_tables(
1844 selectable,
1845 check_columns=True,
1846 include_aliases=True,
1847 include_joins=True,
1848 include_selects=True,
1849 include_crud=True,
1850 ):
1851 if (
1852 isinstance(selected_table, Table)
1853 and selected_table.metadata is table.metadata
1854 ):
1855 fixed_dependencies.add((selected_table, table))
1857 fixed_dependencies.update(
1858 (parent, table) for parent in table._extra_dependencies
1859 )
1861 try:
1862 candidate_sort = list(
1863 topological.sort(
1864 fixed_dependencies.union(mutable_dependencies),
1865 tables,
1866 )
1867 )
1868 except exc.CircularDependencyError as err:
1869 if _warn_for_cycles:
1870 util.warn(
1871 "Cannot correctly sort tables; there are unresolvable cycles "
1872 'between tables "%s", which is usually caused by mutually '
1873 "dependent foreign key constraints. Foreign key constraints "
1874 "involving these tables will not be considered; this warning "
1875 "may raise an error in a future release."
1876 % (", ".join(sorted(t.fullname for t in err.cycles)),)
1877 )
1878 for edge in err.edges:
1879 if edge in mutable_dependencies:
1880 table = edge[1]
1881 if table not in err.cycles:
1882 continue
1883 can_remove = [
1884 fkc
1885 for fkc in table.foreign_key_constraints
1886 if filter_fn is None or filter_fn(fkc) is not False
1887 ]
1888 remaining_fkcs.update(can_remove)
1889 for fkc in can_remove:
1890 dependent_on = fkc.referred_table
1891 if dependent_on is not table:
1892 mutable_dependencies.discard((dependent_on, table))
1893 candidate_sort = list(
1894 topological.sort(
1895 fixed_dependencies.union(mutable_dependencies),
1896 tables,
1897 )
1898 )
1900 return [
1901 (table, table.foreign_key_constraints.difference(remaining_fkcs))
1902 for table in candidate_sort
1903 ] + [(None, list(remaining_fkcs))]