Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/sql/ddl.py: 54%
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-2026 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"""
15from __future__ import annotations
17import contextlib
18from enum import auto
19from enum import Flag
20import typing
21from typing import Any
22from typing import Callable
23from typing import Generic
24from typing import Iterable
25from typing import List
26from typing import Optional
27from typing import Protocol
28from typing import Sequence as typing_Sequence
29from typing import Tuple
30from typing import TypeVar
31from typing import Union
33from . import coercions
34from . import roles
35from . import util as sql_util
36from .base import _generative
37from .base import _NoArg
38from .base import DialectKWArgs
39from .base import Executable
40from .base import NO_ARG
41from .base import SchemaVisitor
42from .elements import ClauseElement
43from .selectable import SelectBase
44from .selectable import TableClause
45from .. import exc
46from .. import util
47from ..util import topological
48from ..util.typing import Self
50if typing.TYPE_CHECKING:
51 from .compiler import Compiled
52 from .compiler import DDLCompiler
53 from .elements import BindParameter
54 from .schema import Column
55 from .schema import Constraint
56 from .schema import ForeignKeyConstraint
57 from .schema import Index
58 from .schema import MetaData
59 from .schema import SchemaItem
60 from .schema import Sequence as Sequence # noqa: F401
61 from .schema import Table
62 from ..engine.base import Connection
63 from ..engine.interfaces import _CoreSingleExecuteParams
64 from ..engine.interfaces import CacheStats
65 from ..engine.interfaces import CompiledCacheType
66 from ..engine.interfaces import Dialect
67 from ..engine.interfaces import SchemaTranslateMapType
69_SI = TypeVar("_SI", bound=Union["SchemaItem", str])
72class BaseDDLElement(ClauseElement):
73 """The root of DDL constructs, including those that are sub-elements
74 within the "create table" and other processes.
76 .. versionadded:: 2.0
78 """
80 _hierarchy_supports_caching = False
81 """disable cache warnings for all _DDLCompiles subclasses. """
83 def _compiler(self, dialect, **kw):
84 """Return a compiler appropriate for this ClauseElement, given a
85 Dialect."""
87 return dialect.ddl_compiler(dialect, self, **kw)
89 def _compile_w_cache(
90 self,
91 dialect: Dialect,
92 *,
93 compiled_cache: Optional[CompiledCacheType],
94 column_keys: List[str],
95 for_executemany: bool = False,
96 schema_translate_map: Optional[SchemaTranslateMapType] = None,
97 **kw: Any,
98 ) -> tuple[
99 Compiled,
100 typing_Sequence[BindParameter[Any]] | None,
101 _CoreSingleExecuteParams | None,
102 CacheStats,
103 ]:
104 raise NotImplementedError()
107class DDLIfCallable(Protocol):
108 def __call__(
109 self,
110 ddl: BaseDDLElement,
111 target: Union[SchemaItem, str],
112 bind: Optional[Connection],
113 tables: Optional[List[Table]] = None,
114 state: Optional[Any] = None,
115 *,
116 dialect: Dialect,
117 compiler: Optional[DDLCompiler] = ...,
118 checkfirst: bool,
119 ) -> bool: ...
122class DDLIf(typing.NamedTuple):
123 dialect: Optional[str]
124 callable_: Optional[DDLIfCallable]
125 state: Optional[Any]
127 def _should_execute(
128 self,
129 ddl: BaseDDLElement,
130 target: Union[SchemaItem, str],
131 bind: Optional[Connection],
132 compiler: Optional[DDLCompiler] = None,
133 **kw: Any,
134 ) -> bool:
135 if bind is not None:
136 dialect = bind.dialect
137 elif compiler is not None:
138 dialect = compiler.dialect
139 else:
140 assert False, "compiler or dialect is required"
142 if isinstance(self.dialect, str):
143 if self.dialect != dialect.name:
144 return False
145 elif isinstance(self.dialect, (tuple, list, set)):
146 if dialect.name not in self.dialect:
147 return False
148 if self.callable_ is not None and not self.callable_(
149 ddl,
150 target,
151 bind,
152 state=self.state,
153 dialect=dialect,
154 compiler=compiler,
155 **kw,
156 ):
157 return False
159 return True
162class ExecutableDDLElement(roles.DDLRole, Executable, BaseDDLElement):
163 """Base class for standalone executable DDL expression constructs.
165 This class is the base for the general purpose :class:`.DDL` class,
166 as well as the various create/drop clause constructs such as
167 :class:`.CreateTable`, :class:`.DropTable`, :class:`.AddConstraint`,
168 etc.
170 .. versionchanged:: 2.0 :class:`.ExecutableDDLElement` is renamed from
171 :class:`.DDLElement`, which still exists for backwards compatibility.
173 :class:`.ExecutableDDLElement` integrates closely with SQLAlchemy events,
174 introduced in :ref:`event_toplevel`. An instance of one is
175 itself an event receiving callable::
177 event.listen(
178 users,
179 "after_create",
180 AddConstraint(constraint, isolate_from_table=True).execute_if(
181 dialect="postgresql"
182 ),
183 )
185 .. seealso::
187 :class:`.DDL`
189 :class:`.DDLEvents`
191 :ref:`event_toplevel`
193 :ref:`schema_ddl_sequences`
195 """
197 _ddl_if: Optional[DDLIf] = None
198 target: Union[SchemaItem, str, None] = None
200 def _execute_on_connection(
201 self, connection, distilled_params, execution_options
202 ):
203 return connection._execute_ddl(
204 self, distilled_params, execution_options
205 )
207 @_generative
208 def against(self, target: SchemaItem) -> Self:
209 """Return a copy of this :class:`_schema.ExecutableDDLElement` which
210 will include the given target.
212 This essentially applies the given item to the ``.target`` attribute of
213 the returned :class:`_schema.ExecutableDDLElement` object. This target
214 is then usable by event handlers and compilation routines in order to
215 provide services such as tokenization of a DDL string in terms of a
216 particular :class:`_schema.Table`.
218 When a :class:`_schema.ExecutableDDLElement` object is established as
219 an event handler for the :meth:`_events.DDLEvents.before_create` or
220 :meth:`_events.DDLEvents.after_create` events, and the event then
221 occurs for a given target such as a :class:`_schema.Constraint` or
222 :class:`_schema.Table`, that target is established with a copy of the
223 :class:`_schema.ExecutableDDLElement` object using this method, which
224 then proceeds to the :meth:`_schema.ExecutableDDLElement.execute`
225 method in order to invoke the actual DDL instruction.
227 :param target: a :class:`_schema.SchemaItem` that will be the subject
228 of a DDL operation.
230 :return: a copy of this :class:`_schema.ExecutableDDLElement` with the
231 ``.target`` attribute assigned to the given
232 :class:`_schema.SchemaItem`.
234 .. seealso::
236 :class:`_schema.DDL` - uses tokenization against the "target" when
237 processing the DDL string.
239 """
240 self.target = target
241 return self
243 @_generative
244 def execute_if(
245 self,
246 dialect: Optional[str] = None,
247 callable_: Optional[DDLIfCallable] = None,
248 state: Optional[Any] = None,
249 ) -> Self:
250 r"""Return a callable that will execute this
251 :class:`_ddl.ExecutableDDLElement` conditionally within an event
252 handler.
254 Used to provide a wrapper for event listening::
256 event.listen(
257 metadata,
258 "before_create",
259 DDL("my_ddl").execute_if(dialect="postgresql"),
260 )
262 :param dialect: May be a string or tuple of strings.
263 If a string, it will be compared to the name of the
264 executing database dialect::
266 DDL("something").execute_if(dialect="postgresql")
268 If a tuple, specifies multiple dialect names::
270 DDL("something").execute_if(dialect=("postgresql", "mysql"))
272 :param callable\_: A callable, which will be invoked with
273 three positional arguments as well as optional keyword
274 arguments:
276 :ddl:
277 This DDL element.
279 :target:
280 The :class:`_schema.Table` or :class:`_schema.MetaData`
281 object which is the
282 target of this event. May be None if the DDL is executed
283 explicitly.
285 :bind:
286 The :class:`_engine.Connection` being used for DDL execution.
287 May be None if this construct is being created inline within
288 a table, in which case ``compiler`` will be present.
290 :tables:
291 Optional keyword argument - a list of Table objects which are to
292 be created/ dropped within a MetaData.create_all() or drop_all()
293 method call.
295 :dialect: keyword argument, but always present - the
296 :class:`.Dialect` involved in the operation.
298 :compiler: keyword argument. Will be ``None`` for an engine
299 level DDL invocation, but will refer to a :class:`.DDLCompiler`
300 if this DDL element is being created inline within a table.
302 :state:
303 Optional keyword argument - will be the ``state`` argument
304 passed to this function.
306 :checkfirst:
307 Keyword argument, will be True if the 'checkfirst' flag was
308 set during the call to ``create()``, ``create_all()``,
309 ``drop()``, ``drop_all()``.
311 If the callable returns a True value, the DDL statement will be
312 executed.
314 :param state: any value which will be passed to the callable\_
315 as the ``state`` keyword argument.
317 .. seealso::
319 :meth:`.SchemaItem.ddl_if`
321 :class:`.DDLEvents`
323 :ref:`event_toplevel`
325 """
326 self._ddl_if = DDLIf(dialect, callable_, state)
327 return self
329 def _should_execute(self, target, bind, **kw):
330 if self._ddl_if is None:
331 return True
332 else:
333 return self._ddl_if._should_execute(self, target, bind, **kw)
335 def _invoke_with(self, bind):
336 if self._should_execute(self.target, bind):
337 return bind.execute(self)
339 def __call__(self, target, bind, **kw):
340 """Execute the DDL as a ddl_listener."""
342 self.against(target)._invoke_with(bind)
344 def _generate(self):
345 s = self.__class__.__new__(self.__class__)
346 s.__dict__ = self.__dict__.copy()
347 return s
350DDLElement = ExecutableDDLElement
351""":class:`.DDLElement` is renamed to :class:`.ExecutableDDLElement`."""
354class DDL(ExecutableDDLElement):
355 """A literal DDL statement.
357 Specifies literal SQL DDL to be executed by the database. DDL objects
358 function as DDL event listeners, and can be subscribed to those events
359 listed in :class:`.DDLEvents`, using either :class:`_schema.Table` or
360 :class:`_schema.MetaData` objects as targets.
361 Basic templating support allows
362 a single DDL instance to handle repetitive tasks for multiple tables.
364 Examples::
366 from sqlalchemy import event, DDL
368 tbl = Table("users", metadata, Column("uid", Integer))
369 event.listen(tbl, "before_create", DDL("DROP TRIGGER users_trigger"))
371 spow = DDL("ALTER TABLE %(table)s SET secretpowers TRUE")
372 event.listen(tbl, "after_create", spow.execute_if(dialect="somedb"))
374 drop_spow = DDL("ALTER TABLE users SET secretpowers FALSE")
375 connection.execute(drop_spow)
377 When operating on Table events, the following ``statement``
378 string substitutions are available:
380 .. sourcecode:: text
382 %(table)s - the Table name, with any required quoting applied
383 %(schema)s - the schema name, with any required quoting applied
384 %(fullname)s - the Table name including schema, quoted if needed
386 The DDL's "context", if any, will be combined with the standard
387 substitutions noted above. Keys present in the context will override
388 the standard substitutions.
390 """
392 __visit_name__ = "ddl"
394 def __init__(self, statement, context=None):
395 """Create a DDL statement.
397 :param statement:
398 A string or unicode string to be executed. Statements will be
399 processed with Python's string formatting operator using
400 a fixed set of string substitutions, as well as additional
401 substitutions provided by the optional :paramref:`.DDL.context`
402 parameter.
404 A literal '%' in a statement must be escaped as '%%'.
406 SQL bind parameters are not available in DDL statements.
408 :param context:
409 Optional dictionary, defaults to None. These values will be
410 available for use in string substitutions on the DDL statement.
412 .. seealso::
414 :class:`.DDLEvents`
416 :ref:`event_toplevel`
418 """
420 if not isinstance(statement, str):
421 raise exc.ArgumentError(
422 "Expected a string or unicode SQL statement, got '%r'"
423 % statement
424 )
426 self.statement = statement
427 self.context = context or {}
429 def __repr__(self):
430 parts = [repr(self.statement)]
431 if self.context:
432 parts.append(f"context={self.context}")
434 return "<%s@%s; %s>" % (
435 type(self).__name__,
436 id(self),
437 ", ".join(parts),
438 )
441class _CreateDropBase(ExecutableDDLElement, Generic[_SI]):
442 """Base class for DDL constructs that represent CREATE and DROP or
443 equivalents.
445 The common theme of _CreateDropBase is a single
446 ``element`` attribute which refers to the element
447 to be created or dropped.
449 """
451 element: _SI
453 def __init__(self, element: _SI) -> None:
454 self.element = self.target = element
455 self._ddl_if = getattr(element, "_ddl_if", None)
457 @property
458 def stringify_dialect(self): # type: ignore[override]
459 assert not isinstance(self.element, str)
460 return self.element.create_drop_stringify_dialect
462 def _create_rule_disable(self, compiler):
463 """Allow disable of _create_rule using a callable.
465 Pass to _create_rule using
466 util.portable_instancemethod(self._create_rule_disable)
467 to retain serializability.
469 """
470 return False
473class _CreateBase(_CreateDropBase[_SI]):
474 def __init__(self, element: _SI, if_not_exists: bool = False) -> None:
475 super().__init__(element)
476 self.if_not_exists = if_not_exists
479class TableCreateDDL(_CreateBase["Table"]):
481 def to_metadata(self, metadata: MetaData, table: Table) -> Self:
482 raise NotImplementedError()
485class _DropBase(_CreateDropBase[_SI]):
486 def __init__(self, element: _SI, if_exists: bool = False) -> None:
487 super().__init__(element)
488 self.if_exists = if_exists
491class TableDropDDL(_DropBase["Table"]):
493 def to_metadata(self, metadata: MetaData, table: Table) -> Self:
494 raise NotImplementedError()
497class CreateSchema(_CreateBase[str]):
498 """Represent a CREATE SCHEMA statement.
500 The argument here is the string name of the schema.
502 """
504 __visit_name__ = "create_schema"
506 stringify_dialect = "default"
508 def __init__(
509 self,
510 name: str,
511 if_not_exists: bool = False,
512 ) -> None:
513 """Create a new :class:`.CreateSchema` construct."""
515 super().__init__(element=name, if_not_exists=if_not_exists)
518class DropSchema(_DropBase[str]):
519 """Represent a DROP SCHEMA statement.
521 The argument here is the string name of the schema.
523 """
525 __visit_name__ = "drop_schema"
527 stringify_dialect = "default"
529 def __init__(
530 self,
531 name: str,
532 cascade: bool = False,
533 if_exists: bool = False,
534 ) -> None:
535 """Create a new :class:`.DropSchema` construct."""
537 super().__init__(element=name, if_exists=if_exists)
538 self.cascade = cascade
541class CreateTable(TableCreateDDL):
542 """Represent a CREATE TABLE statement."""
544 __visit_name__ = "create_table"
546 def __init__(
547 self,
548 element: Table,
549 include_foreign_key_constraints: Optional[
550 typing_Sequence[ForeignKeyConstraint]
551 ] = None,
552 if_not_exists: bool = False,
553 ) -> None:
554 """Create a :class:`.CreateTable` construct.
556 :param element: a :class:`_schema.Table` that's the subject
557 of the CREATE
558 :param on: See the description for 'on' in :class:`.DDL`.
559 :param include_foreign_key_constraints: optional sequence of
560 :class:`_schema.ForeignKeyConstraint` objects that will be included
561 inline within the CREATE construct; if omitted, all foreign key
562 constraints that do not specify use_alter=True are included.
564 :param if_not_exists: if True, an IF NOT EXISTS operator will be
565 applied to the construct.
567 .. versionadded:: 1.4.0b2
569 """
570 super().__init__(element, if_not_exists=if_not_exists)
571 self.columns = [CreateColumn(column) for column in element.columns]
572 self.include_foreign_key_constraints = include_foreign_key_constraints
574 def to_metadata(self, metadata: MetaData, table: Table) -> Self:
575 return self.__class__(table, if_not_exists=self.if_not_exists)
578class _TableViaSelect(TableCreateDDL, ExecutableDDLElement):
579 """Common base class for DDL constructs that generate and render for a
580 :class:`.Table` given a :class:`.Select`
582 .. versionadded:: 2.1
584 """
586 table: Table
587 """:class:`.Table` object representing the table that this
588 :class:`.CreateTableAs` would generate when executed."""
590 def __init__(
591 self,
592 selectable: SelectBase,
593 name: str,
594 *,
595 metadata: Optional["MetaData"] = None,
596 schema: Optional[str] = None,
597 temporary: bool = False,
598 if_not_exists: bool = False,
599 ):
600 # Coerce selectable to a Select statement
601 selectable = coercions.expect(roles.DMLSelectRole, selectable)
603 self.schema = schema
604 self.selectable = selectable
605 self.temporary = bool(temporary)
606 self.if_not_exists = bool(if_not_exists)
607 self.metadata = metadata
608 self.table_name = name
609 self._gen_table()
611 @property
612 def element(self): # type: ignore[override]
613 return self.table
615 def to_metadata(self, metadata: MetaData, table: Table) -> Self:
616 new = self.__class__.__new__(self.__class__)
617 new.__dict__.update(self.__dict__)
618 new.metadata = metadata
619 new.table = table
620 return new
622 @util.preload_module("sqlalchemy.sql.schema")
623 def _gen_table(self) -> None:
624 MetaData = util.preloaded.sql_schema.MetaData
625 Column = util.preloaded.sql_schema.Column
626 Table = util.preloaded.sql_schema.Table
627 MetaData = util.preloaded.sql_schema.MetaData
629 column_name_type_pairs = (
630 (name, col_element.type)
631 for _, name, _, col_element, _ in (
632 self.selectable._generate_columns_plus_names(
633 anon_for_dupe_key=False
634 )
635 )
636 )
638 if self.metadata is None:
639 self.metadata = metadata = MetaData()
640 else:
641 metadata = self.metadata
643 self.table = Table(
644 self.table_name,
645 metadata,
646 *(Column(name, typ) for name, typ in column_name_type_pairs),
647 schema=self.schema,
648 _creator_ddl=self,
649 )
652class CreateTableAs(DialectKWArgs, _TableViaSelect):
653 """Represent a CREATE TABLE ... AS statement.
655 This creates a new table directly from the output of a SELECT, including
656 its schema and its initial set of data. Unlike a view, the
657 new table is fixed and does not synchronize further with the originating
658 SELECT statement.
660 The example below illustrates basic use of :class:`.CreateTableAs`; given a
661 :class:`.Select` and optional :class:`.MetaData`, the
662 :class:`.CreateTableAs` may be invoked directly via
663 :meth:`.Connection.execute` or indirectly via :meth:`.MetaData.create_all`;
664 the :attr:`.CreateTableAs.table` attribute provides a :class:`.Table`
665 object with which to generate new queries::
667 from sqlalchemy import CreateTableAs
668 from sqlalchemy import select
670 # instantiate CreateTableAs given a select() and optional MetaData
671 cas = CreateTableAs(
672 select(users.c.id, users.c.name).where(users.c.status == "active"),
673 "active_users",
674 metadata=some_metadata,
675 )
677 # a Table object is available immediately via the .table attribute
678 new_statement = select(cas.table)
680 # to emit CREATE TABLE AS, either invoke CreateTableAs directly...
681 with engine.begin() as conn:
682 conn.execute(cas)
684 # or alternatively, invoke metadata.create_all()
685 some_metdata.create_all(engine)
687 # drop is performed in the usual way, via drop_all
688 # or table.drop()
689 some_metdata.drop_all(engine)
691 For detailed background on :class:`.CreateTableAs` see
692 :ref:`metadata_create_table_as`.
694 .. versionadded:: 2.1
696 :param selectable: :class:`_sql.Select`
697 The SELECT statement providing the columns and rows.
699 :param table_name: table name as a string. Combine with the optional
700 :paramref:`.CreateTableAs.schema` parameter to indicate a
701 schema-qualified table name.
703 :param metadata: :class:`_schema.MetaData`, optional
704 If provided, the :class:`_schema.Table` object available via the
705 :attr:`.table` attribute will be associated with this
706 :class:`.MetaData`. Otherwise, a new, empty :class:`.MetaData`
707 is created.
709 :param schema: str, optional schema or owner name.
711 :param temporary: bool, default False.
712 If True, render ``TEMPORARY``
714 :param if_not_exists: bool, default False.
715 If True, render ``IF NOT EXISTS``
717 .. seealso::
719 :ref:`metadata_create_table_as` - in :ref:`metadata_toplevel`
721 :meth:`_sql.SelectBase.into` - convenience method to create a
722 :class:`_schema.CreateTableAs` from a SELECT statement
724 :class:`.CreateView`
727 """
729 __visit_name__ = "create_table_as"
730 inherit_cache = False
732 table: Table
733 """:class:`.Table` object representing the table that this
734 :class:`.CreateTableAs` would generate when executed."""
736 def __init__(
737 self,
738 selectable: SelectBase,
739 table_name: str,
740 *,
741 metadata: Optional["MetaData"] = None,
742 schema: Optional[str] = None,
743 temporary: bool = False,
744 if_not_exists: bool = False,
745 **dialect_kwargs: Any,
746 ):
747 self._validate_dialect_kwargs(dialect_kwargs)
748 super().__init__(
749 selectable=selectable,
750 name=table_name,
751 metadata=metadata,
752 schema=schema,
753 temporary=temporary,
754 if_not_exists=if_not_exists,
755 )
758class CreateView(DialectKWArgs, _TableViaSelect):
759 """Represent a CREATE VIEW statement.
761 This creates a new view based on a particular SELECT statement. The schema
762 of the view is based on the columns of the SELECT statement, and the data
763 present in the view is derived from the rows represented by the
764 SELECT. A non-materialized view will evaluate the SELECT statement
765 dynamically as it is queried, whereas a materialized view represents a
766 snapshot of the SELECT statement at a particular point in time and
767 typically needs to be refreshed manually using database-specific commands.
769 The example below illustrates basic use of :class:`.CreateView`; given a
770 :class:`.Select` and optional :class:`.MetaData`, the
771 :class:`.CreateView` may be invoked directly via
772 :meth:`.Connection.execute` or indirectly via :meth:`.MetaData.create_all`;
773 the :attr:`.CreateView.table` attribute provides a :class:`.Table`
774 object with which to generate new queries::
777 from sqlalchemy import select
778 from sqlalchemy.sql.ddl import CreateView
780 # instantiate CreateView given a select() and optional MetaData
781 create_view = CreateView(
782 select(users.c.id, users.c.name).where(users.c.status == "active"),
783 "active_users_view",
784 metadata=some_metadata,
785 )
787 # a Table object is available immediately via the .table attribute
788 new_statement = select(create_view.table)
790 # to emit CREATE VIEW, either invoke CreateView directly...
791 with engine.begin() as conn:
792 conn.execute(create_view)
794 # or alternatively, invoke metadata.create_all()
795 some_metdata.create_all(engine)
797 # drop is performed in the usual way, via drop_all
798 # or table.drop() (will emit DROP VIEW)
799 some_metdata.drop_all(engine)
801 For detailed background on :class:`.CreateView` see
802 :ref:`metadata_create_view`.
804 .. versionadded:: 2.1
806 :param selectable: :class:`_sql.Select`
807 The SELECT statement defining the view.
809 :param view_name: table name as a string. Combine with the optional
810 :paramref:`.CreateView.schema` parameter to indicate a
811 schema-qualified table name.
813 :param metadata: :class:`_schema.MetaData`, optional
814 If provided, the :class:`_schema.Table` object available via the
815 :attr:`.table` attribute will be associated with this
816 :class:`.MetaData`. Otherwise, a new, empty :class:`.MetaData`
817 is created.
819 :param schema: str, optional schema or owner name.
821 :param temporary: bool, default False.
822 If True, render ``TEMPORARY``
824 :param or_replace: bool, default False.
825 If True, render ``OR REPLACE`` to replace an existing view if it
826 exists. Supported by PostgreSQL, MySQL, MariaDB, and Oracle.
827 Not supported by SQLite or SQL Server.
829 .. versionadded:: 2.1
831 :param materialized: bool, default False.
832 If True, render ``MATERIALIZED`` to create a materialized view.
833 Materialized views store the query results physically and can be
834 refreshed periodically. Not supported by all database backends.
836 .. versionadded:: 2.1
838 :param dialect_kw: Additional keyword arguments are dialect-specific and
839 are passed as keyword arguments to the dialect's compiler.
841 .. note::
843 For SQLite, the ``sqlite_if_not_exists`` boolean parameter
844 is supported to render ``CREATE VIEW IF NOT EXISTS``.
846 .. versionadded:: 2.1
848 .. seealso::
850 :ref:`metadata_create_view` - in :ref:`metadata_toplevel`
852 :class:`.CreateTableAs` - for creating a table from a SELECT statement
854 """
856 __visit_name__ = "create_view"
858 inherit_cache = False
860 table: Table
861 """:class:`.Table` object representing the view that this
862 :class:`.CreateView` would generate when executed."""
864 materialized: bool
865 """Boolean flag indicating if this is a materialized view."""
867 or_replace: bool
868 """Boolean flag indicating if OR REPLACE should be used."""
870 def __init__(
871 self,
872 selectable: SelectBase,
873 view_name: str,
874 *,
875 metadata: Optional["MetaData"] = None,
876 schema: Optional[str] = None,
877 temporary: bool = False,
878 or_replace: bool = False,
879 materialized: bool = False,
880 **dialect_kwargs: Any,
881 ):
882 self._validate_dialect_kwargs(dialect_kwargs)
883 super().__init__(
884 selectable=selectable,
885 name=view_name,
886 metadata=metadata,
887 schema=schema,
888 temporary=temporary,
889 if_not_exists=False,
890 )
891 self.materialized = materialized
892 self.or_replace = or_replace
893 self.table._dropper_ddl = DropView(
894 self.table, materialized=materialized
895 )
898class DropView(TableDropDDL):
899 """'DROP VIEW' construct.
901 .. versionadded:: 2.1 the :class:`.DropView` construct became public
902 and was renamed from ``_DropView``.
904 """
906 __visit_name__ = "drop_view"
908 materialized: bool
909 """Boolean flag indicating if this is a materialized view."""
911 def __init__(
912 self,
913 element: Table,
914 *,
915 if_exists: bool = False,
916 materialized: bool = False,
917 ) -> None:
918 super().__init__(element, if_exists=if_exists)
919 self.materialized = materialized
921 def to_metadata(self, metadata: MetaData, table: Table) -> Self:
922 new = self.__class__.__new__(self.__class__)
923 new.__dict__.update(self.__dict__)
924 new.element = table
925 return new
928class CreateConstraint(BaseDDLElement):
929 element: Constraint
931 def __init__(self, element: Constraint) -> None:
932 self.element = element
935class CreateColumn(BaseDDLElement):
936 """Represent a :class:`_schema.Column`
937 as rendered in a CREATE TABLE statement,
938 via the :class:`.CreateTable` construct.
940 This is provided to support custom column DDL within the generation
941 of CREATE TABLE statements, by using the
942 compiler extension documented in :ref:`sqlalchemy.ext.compiler_toplevel`
943 to extend :class:`.CreateColumn`.
945 Typical integration is to examine the incoming :class:`_schema.Column`
946 object, and to redirect compilation if a particular flag or condition
947 is found::
949 from sqlalchemy import schema
950 from sqlalchemy.ext.compiler import compiles
953 @compiles(schema.CreateColumn)
954 def compile(element, compiler, **kw):
955 column = element.element
957 if "special" not in column.info:
958 return compiler.visit_create_column(element, **kw)
960 text = "%s SPECIAL DIRECTIVE %s" % (
961 column.name,
962 compiler.type_compiler.process(column.type),
963 )
964 default = compiler.get_column_default_string(column)
965 if default is not None:
966 text += " DEFAULT " + default
968 if not column.nullable:
969 text += " NOT NULL"
971 if column.constraints:
972 text += " ".join(
973 compiler.process(const) for const in column.constraints
974 )
975 return text
977 The above construct can be applied to a :class:`_schema.Table`
978 as follows::
980 from sqlalchemy import Table, Metadata, Column, Integer, String
981 from sqlalchemy import schema
983 metadata = MetaData()
985 table = Table(
986 "mytable",
987 MetaData(),
988 Column("x", Integer, info={"special": True}, primary_key=True),
989 Column("y", String(50)),
990 Column("z", String(20), info={"special": True}),
991 )
993 metadata.create_all(conn)
995 Above, the directives we've added to the :attr:`_schema.Column.info`
996 collection
997 will be detected by our custom compilation scheme:
999 .. sourcecode:: sql
1001 CREATE TABLE mytable (
1002 x SPECIAL DIRECTIVE INTEGER NOT NULL,
1003 y VARCHAR(50),
1004 z SPECIAL DIRECTIVE VARCHAR(20),
1005 PRIMARY KEY (x)
1006 )
1008 The :class:`.CreateColumn` construct can also be used to skip certain
1009 columns when producing a ``CREATE TABLE``. This is accomplished by
1010 creating a compilation rule that conditionally returns ``None``.
1011 This is essentially how to produce the same effect as using the
1012 ``system=True`` argument on :class:`_schema.Column`, which marks a column
1013 as an implicitly-present "system" column.
1015 For example, suppose we wish to produce a :class:`_schema.Table`
1016 which skips
1017 rendering of the PostgreSQL ``xmin`` column against the PostgreSQL
1018 backend, but on other backends does render it, in anticipation of a
1019 triggered rule. A conditional compilation rule could skip this name only
1020 on PostgreSQL::
1022 from sqlalchemy.schema import CreateColumn
1025 @compiles(CreateColumn, "postgresql")
1026 def skip_xmin(element, compiler, **kw):
1027 if element.element.name == "xmin":
1028 return None
1029 else:
1030 return compiler.visit_create_column(element, **kw)
1033 my_table = Table(
1034 "mytable",
1035 metadata,
1036 Column("id", Integer, primary_key=True),
1037 Column("xmin", Integer),
1038 )
1040 Above, a :class:`.CreateTable` construct will generate a ``CREATE TABLE``
1041 which only includes the ``id`` column in the string; the ``xmin`` column
1042 will be omitted, but only against the PostgreSQL backend.
1044 """
1046 __visit_name__ = "create_column"
1048 element: Column[Any]
1050 def __init__(self, element: Column[Any]) -> None:
1051 self.element = element
1054class DropTable(TableDropDDL):
1055 """Represent a DROP TABLE statement."""
1057 __visit_name__ = "drop_table"
1059 def __init__(self, element: Table, if_exists: bool = False) -> None:
1060 """Create a :class:`.DropTable` construct.
1062 :param element: a :class:`_schema.Table` that's the subject
1063 of the DROP.
1064 :param on: See the description for 'on' in :class:`.DDL`.
1065 :param if_exists: if True, an IF EXISTS operator will be applied to the
1066 construct.
1068 .. versionadded:: 1.4.0b2
1070 """
1071 super().__init__(element, if_exists=if_exists)
1073 def to_metadata(self, metadata: MetaData, table: Table) -> Self:
1074 return self.__class__(table, if_exists=self.if_exists)
1077class CreateSequence(_CreateBase["Sequence"]):
1078 """Represent a CREATE SEQUENCE statement."""
1080 __visit_name__ = "create_sequence"
1083class DropSequence(_DropBase["Sequence"]):
1084 """Represent a DROP SEQUENCE statement."""
1086 __visit_name__ = "drop_sequence"
1089class CreateIndex(_CreateBase["Index"]):
1090 """Represent a CREATE INDEX statement."""
1092 __visit_name__ = "create_index"
1094 def __init__(self, element: Index, if_not_exists: bool = False) -> None:
1095 """Create a :class:`.Createindex` construct.
1097 :param element: a :class:`_schema.Index` that's the subject
1098 of the CREATE.
1099 :param if_not_exists: if True, an IF NOT EXISTS operator will be
1100 applied to the construct.
1102 .. versionadded:: 1.4.0b2
1104 """
1105 super().__init__(element, if_not_exists=if_not_exists)
1108class DropIndex(_DropBase["Index"]):
1109 """Represent a DROP INDEX statement."""
1111 __visit_name__ = "drop_index"
1113 def __init__(self, element: Index, if_exists: bool = False) -> None:
1114 """Create a :class:`.DropIndex` construct.
1116 :param element: a :class:`_schema.Index` that's the subject
1117 of the DROP.
1118 :param if_exists: if True, an IF EXISTS operator will be applied to the
1119 construct.
1121 .. versionadded:: 1.4.0b2
1123 """
1124 super().__init__(element, if_exists=if_exists)
1127class AddConstraint(_CreateBase["Constraint"]):
1128 """Represent an ALTER TABLE ADD CONSTRAINT statement."""
1130 __visit_name__ = "add_constraint"
1132 def __init__(
1133 self, element: Constraint, *, isolate_from_table: bool = True
1134 ) -> None:
1135 """Construct a new :class:`.AddConstraint` construct.
1137 :param element: a :class:`.Constraint` object
1139 :param isolate_from_table: optional boolean. Prevents the target
1140 :class:`.Constraint` from being rendered inline in a "CONSTRAINT"
1141 clause within a CREATE TABLE statement, in the case that the
1142 constraint is associated with a :class:`.Table` which is later
1143 created using :meth:`.Table.create` or :meth:`.MetaData.create_all`.
1144 This occurs by modifying the state of the :class:`.Constraint`
1145 object itself such that the CREATE TABLE DDL process will skip it.
1146 Used for the case when a separate `ALTER TABLE...ADD CONSTRAINT`
1147 call will be emitted after the `CREATE TABLE` has already occurred.
1148 ``True`` by default.
1150 .. versionadded:: 2.0.39 - added
1151 :paramref:`.AddConstraint.isolate_from_table`, defaulting
1152 to True. Previously, the behavior of this parameter was implicitly
1153 turned on in all cases.
1155 """
1156 super().__init__(element)
1158 if isolate_from_table:
1159 element._create_rule = self._create_rule_disable
1162class DropConstraint(_DropBase["Constraint"]):
1163 """Represent an ALTER TABLE DROP CONSTRAINT statement."""
1165 __visit_name__ = "drop_constraint"
1167 def __init__(
1168 self,
1169 element: Constraint,
1170 *,
1171 cascade: bool = False,
1172 if_exists: bool = False,
1173 isolate_from_table: bool | _NoArg = NO_ARG,
1174 **kw: Any,
1175 ) -> None:
1176 """Construct a new :class:`.DropConstraint` construct.
1178 :param element: a :class:`.Constraint` object
1180 :param cascade: optional boolean, indicates backend-specific
1181 "CASCADE CONSTRAINT" directive should be rendered if available
1183 :param if_exists: optional boolean, indicates backend-specific
1184 "IF EXISTS" directive should be rendered if available
1186 :param isolate_from_table: optional boolean. This is a deprecated
1187 setting that when ``True``, does the same thing that
1188 :paramref:`.AddConstraint.isolate_from_table` does, which is prevents
1189 the constraint from being associated with an inline ``CREATE TABLE``
1190 statement. It does not have any effect on the DROP process for a
1191 table and is an artifact of older SQLAlchemy versions,
1192 and will be removed in a future release.
1194 .. versionadded:: 2.0.39 - added
1195 :paramref:`.DropConstraint.isolate_from_table`, defaulting
1196 to True. Previously, the behavior of this parameter was implicitly
1197 turned on in all cases.
1199 .. versionchanged:: 2.1 - This parameter has been deprecated and
1200 the default value of the flag was changed to ``False``.
1202 """
1203 self.cascade = cascade
1204 super().__init__(element, if_exists=if_exists, **kw)
1206 if isolate_from_table is not NO_ARG:
1207 util.warn_deprecated(
1208 "The ``isolate_from_table`` is deprecated and it be removed "
1209 "in a future release.",
1210 "2.1",
1211 )
1213 if isolate_from_table:
1214 element._create_rule = self._create_rule_disable
1217class SetTableComment(_CreateDropBase["Table"]):
1218 """Represent a COMMENT ON TABLE IS statement."""
1220 __visit_name__ = "set_table_comment"
1223class DropTableComment(_CreateDropBase["Table"]):
1224 """Represent a COMMENT ON TABLE '' statement.
1226 Note this varies a lot across database backends.
1228 """
1230 __visit_name__ = "drop_table_comment"
1233class SetColumnComment(_CreateDropBase["Column[Any]"]):
1234 """Represent a COMMENT ON COLUMN IS statement."""
1236 __visit_name__ = "set_column_comment"
1239class DropColumnComment(_CreateDropBase["Column[Any]"]):
1240 """Represent a COMMENT ON COLUMN IS NULL statement."""
1242 __visit_name__ = "drop_column_comment"
1245class SetConstraintComment(_CreateDropBase["Constraint"]):
1246 """Represent a COMMENT ON CONSTRAINT IS statement."""
1248 __visit_name__ = "set_constraint_comment"
1251class DropConstraintComment(_CreateDropBase["Constraint"]):
1252 """Represent a COMMENT ON CONSTRAINT IS NULL statement."""
1254 __visit_name__ = "drop_constraint_comment"
1257class InvokeDDLBase(SchemaVisitor):
1258 def __init__(self, connection, **kw):
1259 self.connection = connection
1260 assert not kw, f"Unexpected keywords: {kw.keys()}"
1262 @contextlib.contextmanager
1263 def with_ddl_events(self, target, **kw):
1264 """helper context manager that will apply appropriate DDL events
1265 to a CREATE or DROP operation."""
1267 raise NotImplementedError()
1270class InvokeCreateDDLBase(InvokeDDLBase):
1271 @contextlib.contextmanager
1272 def with_ddl_events(self, target, **kw):
1273 """helper context manager that will apply appropriate DDL events
1274 to a CREATE or DROP operation."""
1276 target.dispatch.before_create(
1277 target, self.connection, _ddl_runner=self, **kw
1278 )
1279 yield
1280 target.dispatch.after_create(
1281 target, self.connection, _ddl_runner=self, **kw
1282 )
1285class InvokeDropDDLBase(InvokeDDLBase):
1286 @contextlib.contextmanager
1287 def with_ddl_events(self, target, **kw):
1288 """helper context manager that will apply appropriate DDL events
1289 to a CREATE or DROP operation."""
1291 target.dispatch.before_drop(
1292 target, self.connection, _ddl_runner=self, **kw
1293 )
1294 yield
1295 target.dispatch.after_drop(
1296 target, self.connection, _ddl_runner=self, **kw
1297 )
1300class CheckFirst(Flag):
1301 """Enumeration for the :paramref:`.MetaData.create_all.checkfirst`
1302 parameter passed to methods like :meth:`.MetaData.create_all`,
1303 :meth:`.MetaData.drop_all`, :meth:`.Table.create`, :meth:`.Table.drop` and
1304 others.
1306 This enumeration indicates what kinds of objects should be "checked"
1307 with a separate query before emitting CREATE or DROP for that object.
1309 Can use ``CheckFirst(bool_value)`` to convert from a boolean value.
1311 .. versionadded:: 2.1
1313 """
1315 NONE = 0 # equivalent to False
1316 """No items should be checked"""
1318 # avoid 1 so that bool True doesn't match by value
1319 TABLES = 2
1320 """Check for tables"""
1322 VIEWS = auto()
1323 """Check for views"""
1325 INDEXES = auto()
1326 """Check for indexes"""
1328 SEQUENCES = auto()
1329 """Check for sequences"""
1331 TYPES = auto()
1332 """Check for custom datatypes that are created server-side
1334 This is currently used by PostgreSQL.
1336 """
1338 ALL = TABLES | VIEWS | INDEXES | SEQUENCES | TYPES # equivalent to True
1340 @classmethod
1341 def _missing_(cls, value: object) -> Any:
1342 if isinstance(value, bool):
1343 return cls.ALL if value else cls.NONE
1344 return super()._missing_(value)
1347class _SchemaTableReflector:
1348 _effective_tables: typing_Sequence[Table] | None = None
1349 _has_tables: dict[tuple[Optional[str], str], bool]
1351 def __init__(
1352 self,
1353 dialect: Dialect,
1354 connection: Connection,
1355 tables: typing_Sequence[Table] | None,
1356 ):
1357 self.dialect = dialect
1358 self.connection = connection
1359 self.tables = self._effective_tables = tables
1360 self._has_tables = {}
1362 def _update_effective_tables(self, metadata):
1363 if self.tables is not None:
1364 self._effective_tables = self.tables
1365 else:
1366 self._effective_tables = list(metadata.tables.values())
1367 return self._effective_tables
1369 def _has_table(self, table: Table, schema: str | None) -> bool:
1370 res = self._has_tables.get((schema, table.name))
1371 if res is not None:
1372 return res
1373 if self._effective_tables is None:
1374 to_check = [table.name]
1375 else:
1376 is_view = table.is_view
1377 to_check = [
1378 t.name
1379 for t in self._effective_tables
1380 if t.is_view == is_view
1381 and self.connection.schema_for_object(t) == schema
1382 ]
1383 assert table.name in to_check
1384 self._has_tables.update(
1385 self.dialect.has_multi_table(
1386 self.connection, to_check, schema=schema
1387 )
1388 )
1389 return self._has_tables[(schema, table.name)]
1392class SchemaGenerator(InvokeCreateDDLBase, _SchemaTableReflector):
1393 def __init__(
1394 self,
1395 dialect,
1396 connection,
1397 checkfirst=CheckFirst.NONE,
1398 tables=None,
1399 **kwargs,
1400 ):
1401 InvokeCreateDDLBase.__init__(self, connection, **kwargs)
1402 _SchemaTableReflector.__init__(self, dialect, connection, tables)
1403 self.checkfirst = CheckFirst(checkfirst)
1404 self.preparer = dialect.identifier_preparer
1405 self.memo = {}
1407 def _can_create_table(self, table):
1408 self.dialect.validate_identifier(table.name)
1409 effective_schema = self.connection.schema_for_object(table)
1410 if effective_schema:
1411 self.dialect.validate_identifier(effective_schema)
1413 bool_to_check = (
1414 CheckFirst.TABLES if not table.is_view else CheckFirst.VIEWS
1415 )
1416 return not self.checkfirst & bool_to_check or not self._has_table(
1417 table, effective_schema
1418 )
1420 def _can_create_index(self, index):
1421 effective_schema = self.connection.schema_for_object(index.table)
1422 if effective_schema:
1423 self.dialect.validate_identifier(effective_schema)
1424 return (
1425 not self.checkfirst & CheckFirst.INDEXES
1426 or not self.dialect.has_index(
1427 self.connection,
1428 index.table.name,
1429 index.name,
1430 schema=effective_schema,
1431 )
1432 )
1434 def _can_create_sequence(self, sequence):
1435 effective_schema = self.connection.schema_for_object(sequence)
1437 return self.dialect.supports_sequences and (
1438 (not self.dialect.sequences_optional or not sequence.optional)
1439 and (
1440 not self.checkfirst & CheckFirst.SEQUENCES
1441 or not self.dialect.has_sequence(
1442 self.connection, sequence.name, schema=effective_schema
1443 )
1444 )
1445 )
1447 def visit_metadata(self, metadata):
1448 tables = self._update_effective_tables(metadata)
1450 collection = sort_tables_and_constraints(
1451 [t for t in tables if self._can_create_table(t)]
1452 )
1454 seq_coll = [
1455 s
1456 for s in metadata._sequences.values()
1457 if s.column is None and self._can_create_sequence(s)
1458 ]
1460 event_collection = [t for (t, fks) in collection if t is not None]
1462 with self.with_ddl_events(
1463 metadata,
1464 tables=event_collection,
1465 checkfirst=self.checkfirst,
1466 ):
1467 for seq in seq_coll:
1468 self.traverse_single(seq, create_ok=True)
1470 for table, fkcs in collection:
1471 if table is not None:
1472 self.traverse_single(
1473 table,
1474 create_ok=True,
1475 include_foreign_key_constraints=fkcs,
1476 _is_metadata_operation=True,
1477 )
1478 else:
1479 for fkc in fkcs:
1480 self.traverse_single(fkc)
1482 def visit_table(
1483 self,
1484 table,
1485 create_ok=False,
1486 include_foreign_key_constraints=None,
1487 _is_metadata_operation=False,
1488 ):
1489 if not create_ok and not self._can_create_table(table):
1490 return
1492 with self.with_ddl_events(
1493 table,
1494 checkfirst=self.checkfirst,
1495 _is_metadata_operation=_is_metadata_operation,
1496 ):
1497 for column in table.columns:
1498 if column.default is not None:
1499 self.traverse_single(column.default)
1501 if not self.dialect.supports_alter:
1502 # e.g., don't omit any foreign key constraints
1503 include_foreign_key_constraints = None
1505 if table._creator_ddl is not None:
1506 table_create_ddl = table._creator_ddl
1507 else:
1508 table_create_ddl = CreateTable(
1509 table,
1510 include_foreign_key_constraints=(
1511 include_foreign_key_constraints
1512 ),
1513 )
1515 table_create_ddl._invoke_with(self.connection)
1517 if hasattr(table, "indexes"):
1518 for index in table.indexes:
1519 self.traverse_single(index, create_ok=True)
1521 if (
1522 self.dialect.supports_comments
1523 and not self.dialect.inline_comments
1524 ):
1525 if table.comment is not None:
1526 SetTableComment(table)._invoke_with(self.connection)
1528 for column in table.columns:
1529 if column.comment is not None:
1530 SetColumnComment(column)._invoke_with(self.connection)
1532 if self.dialect.supports_constraint_comments:
1533 for constraint in table.constraints:
1534 if constraint.comment is not None:
1535 self.connection.execute(
1536 SetConstraintComment(constraint)
1537 )
1539 def visit_foreign_key_constraint(self, constraint):
1540 if not self.dialect.supports_alter:
1541 return
1543 with self.with_ddl_events(constraint):
1544 AddConstraint(constraint, isolate_from_table=True)._invoke_with(
1545 self.connection
1546 )
1548 def visit_sequence(self, sequence, create_ok=False):
1549 if not create_ok and not self._can_create_sequence(sequence):
1550 return
1551 with self.with_ddl_events(sequence):
1552 CreateSequence(sequence)._invoke_with(self.connection)
1554 def visit_index(self, index, create_ok=False):
1555 if not create_ok and not self._can_create_index(index):
1556 return
1557 with self.with_ddl_events(index):
1558 CreateIndex(index)._invoke_with(self.connection)
1561class SchemaDropper(InvokeDropDDLBase, _SchemaTableReflector):
1562 def __init__(
1563 self,
1564 dialect,
1565 connection,
1566 checkfirst=CheckFirst.NONE,
1567 tables=None,
1568 **kwargs,
1569 ):
1570 InvokeDropDDLBase.__init__(self, connection, **kwargs)
1571 _SchemaTableReflector.__init__(self, dialect, connection, tables)
1572 self.checkfirst = CheckFirst(checkfirst)
1573 self.preparer = dialect.identifier_preparer
1574 self.memo = {}
1576 def visit_metadata(self, metadata):
1577 tables = self._update_effective_tables(metadata)
1579 try:
1580 unsorted_tables = [t for t in tables if self._can_drop_table(t)]
1581 collection = list(
1582 reversed(
1583 sort_tables_and_constraints(
1584 unsorted_tables,
1585 filter_fn=lambda constraint: (
1586 False
1587 if not self.dialect.supports_alter
1588 or constraint.name is None
1589 else None
1590 ),
1591 )
1592 )
1593 )
1594 except exc.CircularDependencyError as err2:
1595 if not self.dialect.supports_alter:
1596 util.warn(
1597 "Can't sort tables for DROP; an "
1598 "unresolvable foreign key "
1599 "dependency exists between tables: %s; and backend does "
1600 "not support ALTER. To restore at least a partial sort, "
1601 "apply use_alter=True to ForeignKey and "
1602 "ForeignKeyConstraint "
1603 "objects involved in the cycle to mark these as known "
1604 "cycles that will be ignored."
1605 % (", ".join(sorted([t.fullname for t in err2.cycles])))
1606 )
1607 collection = [(t, ()) for t in unsorted_tables]
1608 else:
1609 raise exc.CircularDependencyError(
1610 err2.args[0],
1611 err2.cycles,
1612 err2.edges,
1613 msg="Can't sort tables for DROP; an "
1614 "unresolvable foreign key "
1615 "dependency exists between tables: %s. Please ensure "
1616 "that the ForeignKey and ForeignKeyConstraint objects "
1617 "involved in the cycle have "
1618 "names so that they can be dropped using "
1619 "DROP CONSTRAINT."
1620 % (", ".join(sorted([t.fullname for t in err2.cycles]))),
1621 ) from err2
1623 seq_coll = [
1624 s
1625 for s in metadata._sequences.values()
1626 if self._can_drop_sequence(s)
1627 ]
1629 event_collection = [t for (t, fks) in collection if t is not None]
1631 with self.with_ddl_events(
1632 metadata,
1633 tables=event_collection,
1634 checkfirst=self.checkfirst,
1635 ):
1636 for table, fkcs in collection:
1637 if table is not None:
1638 self.traverse_single(
1639 table,
1640 drop_ok=True,
1641 _is_metadata_operation=True,
1642 _ignore_sequences=seq_coll,
1643 )
1644 else:
1645 for fkc in fkcs:
1646 self.traverse_single(fkc)
1648 for seq in seq_coll:
1649 self.traverse_single(seq, drop_ok=seq.column is None)
1651 def _can_drop_table(self, table):
1652 self.dialect.validate_identifier(table.name)
1653 effective_schema = self.connection.schema_for_object(table)
1654 if effective_schema:
1655 self.dialect.validate_identifier(effective_schema)
1656 bool_to_check = (
1657 CheckFirst.TABLES if not table.is_view else CheckFirst.VIEWS
1658 )
1660 return not self.checkfirst & bool_to_check or self._has_table(
1661 table, effective_schema
1662 )
1664 def _can_drop_index(self, index):
1665 effective_schema = self.connection.schema_for_object(index.table)
1666 if effective_schema:
1667 self.dialect.validate_identifier(effective_schema)
1668 return (
1669 not self.checkfirst & CheckFirst.INDEXES
1670 or self.dialect.has_index(
1671 self.connection,
1672 index.table.name,
1673 index.name,
1674 schema=effective_schema,
1675 )
1676 )
1678 def _can_drop_sequence(self, sequence):
1679 effective_schema = self.connection.schema_for_object(sequence)
1680 return self.dialect.supports_sequences and (
1681 (not self.dialect.sequences_optional or not sequence.optional)
1682 and (
1683 not self.checkfirst & CheckFirst.SEQUENCES
1684 or self.dialect.has_sequence(
1685 self.connection, sequence.name, schema=effective_schema
1686 )
1687 )
1688 )
1690 def visit_index(self, index, drop_ok=False):
1691 if not drop_ok and not self._can_drop_index(index):
1692 return
1694 with self.with_ddl_events(index):
1695 DropIndex(index)(index, self.connection)
1697 def visit_table(
1698 self,
1699 table,
1700 drop_ok=False,
1701 _is_metadata_operation=False,
1702 _ignore_sequences=(),
1703 ):
1704 if not drop_ok and not self._can_drop_table(table):
1705 return
1707 with self.with_ddl_events(
1708 table,
1709 checkfirst=self.checkfirst,
1710 _is_metadata_operation=_is_metadata_operation,
1711 ):
1712 if table._dropper_ddl is not None:
1713 table_dropper_ddl = table._dropper_ddl
1714 else:
1715 table_dropper_ddl = DropTable(table)
1716 table_dropper_ddl._invoke_with(self.connection)
1718 # traverse client side defaults which may refer to server-side
1719 # sequences. noting that some of these client side defaults may
1720 # also be set up as server side defaults
1721 # (see https://docs.sqlalchemy.org/en/
1722 # latest/core/defaults.html
1723 # #associating-a-sequence-as-the-server-side-
1724 # default), so have to be dropped after the table is dropped.
1725 for column in table.columns:
1726 if (
1727 column.default is not None
1728 and column.default not in _ignore_sequences
1729 ):
1730 self.traverse_single(column.default)
1732 def visit_foreign_key_constraint(self, constraint):
1733 if not self.dialect.supports_alter:
1734 return
1735 with self.with_ddl_events(constraint):
1736 DropConstraint(constraint)._invoke_with(self.connection)
1738 def visit_sequence(self, sequence, drop_ok=False):
1739 if not drop_ok and not self._can_drop_sequence(sequence):
1740 return
1741 with self.with_ddl_events(sequence):
1742 DropSequence(sequence)._invoke_with(self.connection)
1745def sort_tables(
1746 tables: Iterable[TableClause],
1747 skip_fn: Optional[Callable[[ForeignKeyConstraint], bool]] = None,
1748 extra_dependencies: Optional[
1749 typing_Sequence[Tuple[TableClause, TableClause]]
1750 ] = None,
1751) -> List[Table]:
1752 """Sort a collection of :class:`_schema.Table` objects based on
1753 dependency.
1755 This is a dependency-ordered sort which will emit :class:`_schema.Table`
1756 objects such that they will follow their dependent :class:`_schema.Table`
1757 objects.
1758 Tables are dependent on another based on the presence of
1759 :class:`_schema.ForeignKeyConstraint`
1760 objects as well as explicit dependencies
1761 added by :meth:`_schema.Table.add_is_dependent_on`.
1763 .. warning::
1765 The :func:`._schema.sort_tables` function cannot by itself
1766 accommodate automatic resolution of dependency cycles between
1767 tables, which are usually caused by mutually dependent foreign key
1768 constraints. When these cycles are detected, the foreign keys
1769 of these tables are omitted from consideration in the sort.
1770 A warning is emitted when this condition occurs, which will be an
1771 exception raise in a future release. Tables which are not part
1772 of the cycle will still be returned in dependency order.
1774 To resolve these cycles, the
1775 :paramref:`_schema.ForeignKeyConstraint.use_alter` parameter may be
1776 applied to those constraints which create a cycle. Alternatively,
1777 the :func:`_schema.sort_tables_and_constraints` function will
1778 automatically return foreign key constraints in a separate
1779 collection when cycles are detected so that they may be applied
1780 to a schema separately.
1782 :param tables: a sequence of :class:`_schema.Table` objects.
1784 :param skip_fn: optional callable which will be passed a
1785 :class:`_schema.ForeignKeyConstraint` object; if it returns True, this
1786 constraint will not be considered as a dependency. Note this is
1787 **different** from the same parameter in
1788 :func:`.sort_tables_and_constraints`, which is
1789 instead passed the owning :class:`_schema.ForeignKeyConstraint` object.
1791 :param extra_dependencies: a sequence of 2-tuples of tables which will
1792 also be considered as dependent on each other.
1794 .. seealso::
1796 :func:`.sort_tables_and_constraints`
1798 :attr:`_schema.MetaData.sorted_tables` - uses this function to sort
1801 """
1803 if skip_fn is not None:
1804 fixed_skip_fn = skip_fn
1806 def _skip_fn(fkc):
1807 for fk in fkc.elements:
1808 if fixed_skip_fn(fk):
1809 return True
1810 else:
1811 return None
1813 else:
1814 _skip_fn = None # type: ignore[assignment]
1816 return [
1817 t
1818 for (t, fkcs) in sort_tables_and_constraints(
1819 tables,
1820 filter_fn=_skip_fn,
1821 extra_dependencies=extra_dependencies,
1822 _warn_for_cycles=True,
1823 )
1824 if t is not None
1825 ]
1828@util.preload_module("sqlalchemy.sql.schema")
1829def sort_tables_and_constraints(
1830 tables, filter_fn=None, extra_dependencies=None, _warn_for_cycles=False
1831):
1832 """Sort a collection of :class:`_schema.Table` /
1833 :class:`_schema.ForeignKeyConstraint`
1834 objects.
1836 This is a dependency-ordered sort which will emit tuples of
1837 ``(Table, [ForeignKeyConstraint, ...])`` such that each
1838 :class:`_schema.Table` follows its dependent :class:`_schema.Table`
1839 objects.
1840 Remaining :class:`_schema.ForeignKeyConstraint`
1841 objects that are separate due to
1842 dependency rules not satisfied by the sort are emitted afterwards
1843 as ``(None, [ForeignKeyConstraint ...])``.
1845 Tables are dependent on another based on the presence of
1846 :class:`_schema.ForeignKeyConstraint` objects, explicit dependencies
1847 added by :meth:`_schema.Table.add_is_dependent_on`,
1848 as well as dependencies
1849 stated here using the :paramref:`~.sort_tables_and_constraints.skip_fn`
1850 and/or :paramref:`~.sort_tables_and_constraints.extra_dependencies`
1851 parameters.
1853 :param tables: a sequence of :class:`_schema.Table` objects.
1855 :param filter_fn: optional callable which will be passed a
1856 :class:`_schema.ForeignKeyConstraint` object,
1857 and returns a value based on
1858 whether this constraint should definitely be included or excluded as
1859 an inline constraint, or neither. If it returns False, the constraint
1860 will definitely be included as a dependency that cannot be subject
1861 to ALTER; if True, it will **only** be included as an ALTER result at
1862 the end. Returning None means the constraint is included in the
1863 table-based result unless it is detected as part of a dependency cycle.
1865 :param extra_dependencies: a sequence of 2-tuples of tables which will
1866 also be considered as dependent on each other.
1868 .. seealso::
1870 :func:`.sort_tables`
1873 """
1874 Table = util.preloaded.sql_schema.Table
1876 fixed_dependencies = set()
1877 mutable_dependencies = set()
1879 if extra_dependencies is not None:
1880 fixed_dependencies.update(extra_dependencies)
1882 remaining_fkcs = set()
1883 for table in tables:
1884 for fkc in table.foreign_key_constraints:
1885 if fkc.use_alter is True:
1886 remaining_fkcs.add(fkc)
1887 continue
1889 if filter_fn:
1890 filtered = filter_fn(fkc)
1892 if filtered is True:
1893 remaining_fkcs.add(fkc)
1894 continue
1896 dependent_on = fkc.referred_table
1897 if dependent_on is not table:
1898 mutable_dependencies.add((dependent_on, table))
1900 if isinstance(table._creator_ddl, _TableViaSelect):
1901 selectable = table._creator_ddl.selectable
1902 for selected_table in sql_util.find_tables(
1903 selectable,
1904 check_columns=True,
1905 include_aliases=True,
1906 include_joins=True,
1907 include_selects=True,
1908 include_crud=True,
1909 ):
1910 if (
1911 isinstance(selected_table, Table)
1912 and selected_table.metadata is table.metadata
1913 ):
1914 fixed_dependencies.add((selected_table, table))
1916 fixed_dependencies.update(
1917 (parent, table) for parent in table._extra_dependencies
1918 )
1920 try:
1921 candidate_sort = list(
1922 topological.sort(
1923 fixed_dependencies.union(mutable_dependencies),
1924 tables,
1925 )
1926 )
1927 except exc.CircularDependencyError as err:
1928 if _warn_for_cycles:
1929 util.warn(
1930 "Cannot correctly sort tables; there are unresolvable cycles "
1931 'between tables "%s", which is usually caused by mutually '
1932 "dependent foreign key constraints. Foreign key constraints "
1933 "involving these tables will not be considered; this warning "
1934 "may raise an error in a future release."
1935 % (", ".join(sorted(t.fullname for t in err.cycles)),)
1936 )
1937 for edge in err.edges:
1938 if edge in mutable_dependencies:
1939 table = edge[1]
1940 if table not in err.cycles:
1941 continue
1942 can_remove = [
1943 fkc
1944 for fkc in table.foreign_key_constraints
1945 if filter_fn is None or filter_fn(fkc) is not False
1946 ]
1947 remaining_fkcs.update(can_remove)
1948 for fkc in can_remove:
1949 dependent_on = fkc.referred_table
1950 if dependent_on is not table:
1951 mutable_dependencies.discard((dependent_on, table))
1952 candidate_sort = list(
1953 topological.sort(
1954 fixed_dependencies.union(mutable_dependencies),
1955 tables,
1956 )
1957 )
1959 return [
1960 (table, table.foreign_key_constraints.difference(remaining_fkcs))
1961 for table in candidate_sort
1962 ] + [(None, list(remaining_fkcs))]