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-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"""
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 _NoArg
37from .base import DialectKWArgs
38from .base import Executable
39from .base import NO_ARG
40from .base import SchemaVisitor
41from .elements import ClauseElement
42from .selectable import SelectBase
43from .selectable import TableClause
44from .. import exc
45from .. import util
46from ..util import topological
47from ..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
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 SchemaGenerator(InvokeCreateDDLBase):
1348 def __init__(
1349 self,
1350 dialect,
1351 connection,
1352 checkfirst=CheckFirst.NONE,
1353 tables=None,
1354 **kwargs,
1355 ):
1356 super().__init__(connection, **kwargs)
1357 self.checkfirst = CheckFirst(checkfirst)
1358 self.tables = tables
1359 self.preparer = dialect.identifier_preparer
1360 self.dialect = dialect
1361 self.memo = {}
1363 def _can_create_table(self, table):
1364 self.dialect.validate_identifier(table.name)
1365 effective_schema = self.connection.schema_for_object(table)
1366 if effective_schema:
1367 self.dialect.validate_identifier(effective_schema)
1369 bool_to_check = (
1370 CheckFirst.TABLES if not table.is_view else CheckFirst.VIEWS
1371 )
1372 return (
1373 not self.checkfirst & bool_to_check
1374 or not self.dialect.has_table(
1375 self.connection, table.name, schema=effective_schema
1376 )
1377 )
1379 def _can_create_index(self, index):
1380 effective_schema = self.connection.schema_for_object(index.table)
1381 if effective_schema:
1382 self.dialect.validate_identifier(effective_schema)
1383 return (
1384 not self.checkfirst & CheckFirst.INDEXES
1385 or not self.dialect.has_index(
1386 self.connection,
1387 index.table.name,
1388 index.name,
1389 schema=effective_schema,
1390 )
1391 )
1393 def _can_create_sequence(self, sequence):
1394 effective_schema = self.connection.schema_for_object(sequence)
1396 return self.dialect.supports_sequences and (
1397 (not self.dialect.sequences_optional or not sequence.optional)
1398 and (
1399 not self.checkfirst & CheckFirst.SEQUENCES
1400 or not self.dialect.has_sequence(
1401 self.connection, sequence.name, schema=effective_schema
1402 )
1403 )
1404 )
1406 def visit_metadata(self, metadata):
1407 if self.tables is not None:
1408 tables = self.tables
1409 else:
1410 tables = list(metadata.tables.values())
1412 collection = sort_tables_and_constraints(
1413 [t for t in tables if self._can_create_table(t)]
1414 )
1416 seq_coll = [
1417 s
1418 for s in metadata._sequences.values()
1419 if s.column is None and self._can_create_sequence(s)
1420 ]
1422 event_collection = [t for (t, fks) in collection if t is not None]
1424 with self.with_ddl_events(
1425 metadata,
1426 tables=event_collection,
1427 checkfirst=self.checkfirst,
1428 ):
1429 for seq in seq_coll:
1430 self.traverse_single(seq, create_ok=True)
1432 for table, fkcs in collection:
1433 if table is not None:
1434 self.traverse_single(
1435 table,
1436 create_ok=True,
1437 include_foreign_key_constraints=fkcs,
1438 _is_metadata_operation=True,
1439 )
1440 else:
1441 for fkc in fkcs:
1442 self.traverse_single(fkc)
1444 def visit_table(
1445 self,
1446 table,
1447 create_ok=False,
1448 include_foreign_key_constraints=None,
1449 _is_metadata_operation=False,
1450 ):
1451 if not create_ok and not self._can_create_table(table):
1452 return
1454 with self.with_ddl_events(
1455 table,
1456 checkfirst=self.checkfirst,
1457 _is_metadata_operation=_is_metadata_operation,
1458 ):
1459 for column in table.columns:
1460 if column.default is not None:
1461 self.traverse_single(column.default)
1463 if not self.dialect.supports_alter:
1464 # e.g., don't omit any foreign key constraints
1465 include_foreign_key_constraints = None
1467 if table._creator_ddl is not None:
1468 table_create_ddl = table._creator_ddl
1469 else:
1470 table_create_ddl = CreateTable(
1471 table,
1472 include_foreign_key_constraints=(
1473 include_foreign_key_constraints
1474 ),
1475 )
1477 table_create_ddl._invoke_with(self.connection)
1479 if hasattr(table, "indexes"):
1480 for index in table.indexes:
1481 self.traverse_single(index, create_ok=True)
1483 if (
1484 self.dialect.supports_comments
1485 and not self.dialect.inline_comments
1486 ):
1487 if table.comment is not None:
1488 SetTableComment(table)._invoke_with(self.connection)
1490 for column in table.columns:
1491 if column.comment is not None:
1492 SetColumnComment(column)._invoke_with(self.connection)
1494 if self.dialect.supports_constraint_comments:
1495 for constraint in table.constraints:
1496 if constraint.comment is not None:
1497 self.connection.execute(
1498 SetConstraintComment(constraint)
1499 )
1501 def visit_foreign_key_constraint(self, constraint):
1502 if not self.dialect.supports_alter:
1503 return
1505 with self.with_ddl_events(constraint):
1506 AddConstraint(constraint, isolate_from_table=True)._invoke_with(
1507 self.connection
1508 )
1510 def visit_sequence(self, sequence, create_ok=False):
1511 if not create_ok and not self._can_create_sequence(sequence):
1512 return
1513 with self.with_ddl_events(sequence):
1514 CreateSequence(sequence)._invoke_with(self.connection)
1516 def visit_index(self, index, create_ok=False):
1517 if not create_ok and not self._can_create_index(index):
1518 return
1519 with self.with_ddl_events(index):
1520 CreateIndex(index)._invoke_with(self.connection)
1523class SchemaDropper(InvokeDropDDLBase):
1524 def __init__(
1525 self,
1526 dialect,
1527 connection,
1528 checkfirst=CheckFirst.NONE,
1529 tables=None,
1530 **kwargs,
1531 ):
1532 super().__init__(connection, **kwargs)
1533 self.checkfirst = CheckFirst(checkfirst)
1534 self.tables = tables
1535 self.preparer = dialect.identifier_preparer
1536 self.dialect = dialect
1537 self.memo = {}
1539 def visit_metadata(self, metadata):
1540 if self.tables is not None:
1541 tables = self.tables
1542 else:
1543 tables = list(metadata.tables.values())
1545 try:
1546 unsorted_tables = [t for t in tables if self._can_drop_table(t)]
1547 collection = list(
1548 reversed(
1549 sort_tables_and_constraints(
1550 unsorted_tables,
1551 filter_fn=lambda constraint: (
1552 False
1553 if not self.dialect.supports_alter
1554 or constraint.name is None
1555 else None
1556 ),
1557 )
1558 )
1559 )
1560 except exc.CircularDependencyError as err2:
1561 if not self.dialect.supports_alter:
1562 util.warn(
1563 "Can't sort tables for DROP; an "
1564 "unresolvable foreign key "
1565 "dependency exists between tables: %s; and backend does "
1566 "not support ALTER. To restore at least a partial sort, "
1567 "apply use_alter=True to ForeignKey and "
1568 "ForeignKeyConstraint "
1569 "objects involved in the cycle to mark these as known "
1570 "cycles that will be ignored."
1571 % (", ".join(sorted([t.fullname for t in err2.cycles])))
1572 )
1573 collection = [(t, ()) for t in unsorted_tables]
1574 else:
1575 raise exc.CircularDependencyError(
1576 err2.args[0],
1577 err2.cycles,
1578 err2.edges,
1579 msg="Can't sort tables for DROP; an "
1580 "unresolvable foreign key "
1581 "dependency exists between tables: %s. Please ensure "
1582 "that the ForeignKey and ForeignKeyConstraint objects "
1583 "involved in the cycle have "
1584 "names so that they can be dropped using "
1585 "DROP CONSTRAINT."
1586 % (", ".join(sorted([t.fullname for t in err2.cycles]))),
1587 ) from err2
1589 seq_coll = [
1590 s
1591 for s in metadata._sequences.values()
1592 if self._can_drop_sequence(s)
1593 ]
1595 event_collection = [t for (t, fks) in collection if t is not None]
1597 with self.with_ddl_events(
1598 metadata,
1599 tables=event_collection,
1600 checkfirst=self.checkfirst,
1601 ):
1602 for table, fkcs in collection:
1603 if table is not None:
1604 self.traverse_single(
1605 table,
1606 drop_ok=True,
1607 _is_metadata_operation=True,
1608 _ignore_sequences=seq_coll,
1609 )
1610 else:
1611 for fkc in fkcs:
1612 self.traverse_single(fkc)
1614 for seq in seq_coll:
1615 self.traverse_single(seq, drop_ok=seq.column is None)
1617 def _can_drop_table(self, table):
1618 self.dialect.validate_identifier(table.name)
1619 effective_schema = self.connection.schema_for_object(table)
1620 if effective_schema:
1621 self.dialect.validate_identifier(effective_schema)
1622 bool_to_check = (
1623 CheckFirst.TABLES if not table.is_view else CheckFirst.VIEWS
1624 )
1626 return not self.checkfirst & bool_to_check or self.dialect.has_table(
1627 self.connection, table.name, schema=effective_schema
1628 )
1630 def _can_drop_index(self, index):
1631 effective_schema = self.connection.schema_for_object(index.table)
1632 if effective_schema:
1633 self.dialect.validate_identifier(effective_schema)
1634 return (
1635 not self.checkfirst & CheckFirst.INDEXES
1636 or self.dialect.has_index(
1637 self.connection,
1638 index.table.name,
1639 index.name,
1640 schema=effective_schema,
1641 )
1642 )
1644 def _can_drop_sequence(self, sequence):
1645 effective_schema = self.connection.schema_for_object(sequence)
1646 return self.dialect.supports_sequences and (
1647 (not self.dialect.sequences_optional or not sequence.optional)
1648 and (
1649 not self.checkfirst & CheckFirst.SEQUENCES
1650 or self.dialect.has_sequence(
1651 self.connection, sequence.name, schema=effective_schema
1652 )
1653 )
1654 )
1656 def visit_index(self, index, drop_ok=False):
1657 if not drop_ok and not self._can_drop_index(index):
1658 return
1660 with self.with_ddl_events(index):
1661 DropIndex(index)(index, self.connection)
1663 def visit_table(
1664 self,
1665 table,
1666 drop_ok=False,
1667 _is_metadata_operation=False,
1668 _ignore_sequences=(),
1669 ):
1670 if not drop_ok and not self._can_drop_table(table):
1671 return
1673 with self.with_ddl_events(
1674 table,
1675 checkfirst=self.checkfirst,
1676 _is_metadata_operation=_is_metadata_operation,
1677 ):
1678 if table._dropper_ddl is not None:
1679 table_dropper_ddl = table._dropper_ddl
1680 else:
1681 table_dropper_ddl = DropTable(table)
1682 table_dropper_ddl._invoke_with(self.connection)
1684 # traverse client side defaults which may refer to server-side
1685 # sequences. noting that some of these client side defaults may
1686 # also be set up as server side defaults
1687 # (see https://docs.sqlalchemy.org/en/
1688 # latest/core/defaults.html
1689 # #associating-a-sequence-as-the-server-side-
1690 # default), so have to be dropped after the table is dropped.
1691 for column in table.columns:
1692 if (
1693 column.default is not None
1694 and column.default not in _ignore_sequences
1695 ):
1696 self.traverse_single(column.default)
1698 def visit_foreign_key_constraint(self, constraint):
1699 if not self.dialect.supports_alter:
1700 return
1701 with self.with_ddl_events(constraint):
1702 DropConstraint(constraint)._invoke_with(self.connection)
1704 def visit_sequence(self, sequence, drop_ok=False):
1705 if not drop_ok and not self._can_drop_sequence(sequence):
1706 return
1707 with self.with_ddl_events(sequence):
1708 DropSequence(sequence)._invoke_with(self.connection)
1711def sort_tables(
1712 tables: Iterable[TableClause],
1713 skip_fn: Optional[Callable[[ForeignKeyConstraint], bool]] = None,
1714 extra_dependencies: Optional[
1715 typing_Sequence[Tuple[TableClause, TableClause]]
1716 ] = None,
1717) -> List[Table]:
1718 """Sort a collection of :class:`_schema.Table` objects based on
1719 dependency.
1721 This is a dependency-ordered sort which will emit :class:`_schema.Table`
1722 objects such that they will follow their dependent :class:`_schema.Table`
1723 objects.
1724 Tables are dependent on another based on the presence of
1725 :class:`_schema.ForeignKeyConstraint`
1726 objects as well as explicit dependencies
1727 added by :meth:`_schema.Table.add_is_dependent_on`.
1729 .. warning::
1731 The :func:`._schema.sort_tables` function cannot by itself
1732 accommodate automatic resolution of dependency cycles between
1733 tables, which are usually caused by mutually dependent foreign key
1734 constraints. When these cycles are detected, the foreign keys
1735 of these tables are omitted from consideration in the sort.
1736 A warning is emitted when this condition occurs, which will be an
1737 exception raise in a future release. Tables which are not part
1738 of the cycle will still be returned in dependency order.
1740 To resolve these cycles, the
1741 :paramref:`_schema.ForeignKeyConstraint.use_alter` parameter may be
1742 applied to those constraints which create a cycle. Alternatively,
1743 the :func:`_schema.sort_tables_and_constraints` function will
1744 automatically return foreign key constraints in a separate
1745 collection when cycles are detected so that they may be applied
1746 to a schema separately.
1748 :param tables: a sequence of :class:`_schema.Table` objects.
1750 :param skip_fn: optional callable which will be passed a
1751 :class:`_schema.ForeignKeyConstraint` object; if it returns True, this
1752 constraint will not be considered as a dependency. Note this is
1753 **different** from the same parameter in
1754 :func:`.sort_tables_and_constraints`, which is
1755 instead passed the owning :class:`_schema.ForeignKeyConstraint` object.
1757 :param extra_dependencies: a sequence of 2-tuples of tables which will
1758 also be considered as dependent on each other.
1760 .. seealso::
1762 :func:`.sort_tables_and_constraints`
1764 :attr:`_schema.MetaData.sorted_tables` - uses this function to sort
1767 """
1769 if skip_fn is not None:
1770 fixed_skip_fn = skip_fn
1772 def _skip_fn(fkc):
1773 for fk in fkc.elements:
1774 if fixed_skip_fn(fk):
1775 return True
1776 else:
1777 return None
1779 else:
1780 _skip_fn = None # type: ignore
1782 return [
1783 t
1784 for (t, fkcs) in sort_tables_and_constraints(
1785 tables,
1786 filter_fn=_skip_fn,
1787 extra_dependencies=extra_dependencies,
1788 _warn_for_cycles=True,
1789 )
1790 if t is not None
1791 ]
1794@util.preload_module("sqlalchemy.sql.schema")
1795def sort_tables_and_constraints(
1796 tables, filter_fn=None, extra_dependencies=None, _warn_for_cycles=False
1797):
1798 """Sort a collection of :class:`_schema.Table` /
1799 :class:`_schema.ForeignKeyConstraint`
1800 objects.
1802 This is a dependency-ordered sort which will emit tuples of
1803 ``(Table, [ForeignKeyConstraint, ...])`` such that each
1804 :class:`_schema.Table` follows its dependent :class:`_schema.Table`
1805 objects.
1806 Remaining :class:`_schema.ForeignKeyConstraint`
1807 objects that are separate due to
1808 dependency rules not satisfied by the sort are emitted afterwards
1809 as ``(None, [ForeignKeyConstraint ...])``.
1811 Tables are dependent on another based on the presence of
1812 :class:`_schema.ForeignKeyConstraint` objects, explicit dependencies
1813 added by :meth:`_schema.Table.add_is_dependent_on`,
1814 as well as dependencies
1815 stated here using the :paramref:`~.sort_tables_and_constraints.skip_fn`
1816 and/or :paramref:`~.sort_tables_and_constraints.extra_dependencies`
1817 parameters.
1819 :param tables: a sequence of :class:`_schema.Table` objects.
1821 :param filter_fn: optional callable which will be passed a
1822 :class:`_schema.ForeignKeyConstraint` object,
1823 and returns a value based on
1824 whether this constraint should definitely be included or excluded as
1825 an inline constraint, or neither. If it returns False, the constraint
1826 will definitely be included as a dependency that cannot be subject
1827 to ALTER; if True, it will **only** be included as an ALTER result at
1828 the end. Returning None means the constraint is included in the
1829 table-based result unless it is detected as part of a dependency cycle.
1831 :param extra_dependencies: a sequence of 2-tuples of tables which will
1832 also be considered as dependent on each other.
1834 .. seealso::
1836 :func:`.sort_tables`
1839 """
1840 Table = util.preloaded.sql_schema.Table
1842 fixed_dependencies = set()
1843 mutable_dependencies = set()
1845 if extra_dependencies is not None:
1846 fixed_dependencies.update(extra_dependencies)
1848 remaining_fkcs = set()
1849 for table in tables:
1850 for fkc in table.foreign_key_constraints:
1851 if fkc.use_alter is True:
1852 remaining_fkcs.add(fkc)
1853 continue
1855 if filter_fn:
1856 filtered = filter_fn(fkc)
1858 if filtered is True:
1859 remaining_fkcs.add(fkc)
1860 continue
1862 dependent_on = fkc.referred_table
1863 if dependent_on is not table:
1864 mutable_dependencies.add((dependent_on, table))
1866 if isinstance(table._creator_ddl, _TableViaSelect):
1867 selectable = table._creator_ddl.selectable
1868 for selected_table in sql_util.find_tables(
1869 selectable,
1870 check_columns=True,
1871 include_aliases=True,
1872 include_joins=True,
1873 include_selects=True,
1874 include_crud=True,
1875 ):
1876 if (
1877 isinstance(selected_table, Table)
1878 and selected_table.metadata is table.metadata
1879 ):
1880 fixed_dependencies.add((selected_table, table))
1882 fixed_dependencies.update(
1883 (parent, table) for parent in table._extra_dependencies
1884 )
1886 try:
1887 candidate_sort = list(
1888 topological.sort(
1889 fixed_dependencies.union(mutable_dependencies),
1890 tables,
1891 )
1892 )
1893 except exc.CircularDependencyError as err:
1894 if _warn_for_cycles:
1895 util.warn(
1896 "Cannot correctly sort tables; there are unresolvable cycles "
1897 'between tables "%s", which is usually caused by mutually '
1898 "dependent foreign key constraints. Foreign key constraints "
1899 "involving these tables will not be considered; this warning "
1900 "may raise an error in a future release."
1901 % (", ".join(sorted(t.fullname for t in err.cycles)),)
1902 )
1903 for edge in err.edges:
1904 if edge in mutable_dependencies:
1905 table = edge[1]
1906 if table not in err.cycles:
1907 continue
1908 can_remove = [
1909 fkc
1910 for fkc in table.foreign_key_constraints
1911 if filter_fn is None or filter_fn(fkc) is not False
1912 ]
1913 remaining_fkcs.update(can_remove)
1914 for fkc in can_remove:
1915 dependent_on = fkc.referred_table
1916 if dependent_on is not table:
1917 mutable_dependencies.discard((dependent_on, table))
1918 candidate_sort = list(
1919 topological.sort(
1920 fixed_dependencies.union(mutable_dependencies),
1921 tables,
1922 )
1923 )
1925 return [
1926 (table, table.foreign_key_constraints.difference(remaining_fkcs))
1927 for table in candidate_sort
1928 ] + [(None, list(remaining_fkcs))]