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"""
15from __future__ import annotations
17import contextlib
18import typing
19from typing import Any
20from typing import Callable
21from typing import Generic
22from typing import Iterable
23from typing import List
24from typing import Optional
25from typing import Sequence as typing_Sequence
26from typing import Tuple
27from typing import TypeVar
28from typing import Union
30from . import roles
31from .base import _generative
32from .base import Executable
33from .base import SchemaVisitor
34from .elements import ClauseElement
35from .. import exc
36from .. import util
37from ..util import topological
38from ..util.typing import Protocol
39from ..util.typing import Self
41if typing.TYPE_CHECKING:
42 from .compiler import Compiled
43 from .compiler import DDLCompiler
44 from .elements import BindParameter
45 from .schema import Column
46 from .schema import Constraint
47 from .schema import ForeignKeyConstraint
48 from .schema import Index
49 from .schema import SchemaItem
50 from .schema import Sequence as Sequence # noqa: F401
51 from .schema import Table
52 from .selectable import TableClause
53 from ..engine.base import Connection
54 from ..engine.interfaces import CacheStats
55 from ..engine.interfaces import CompiledCacheType
56 from ..engine.interfaces import Dialect
57 from ..engine.interfaces import SchemaTranslateMapType
59_SI = TypeVar("_SI", bound=Union["SchemaItem", str])
62class BaseDDLElement(ClauseElement):
63 """The root of DDL constructs, including those that are sub-elements
64 within the "create table" and other processes.
66 .. versionadded:: 2.0
68 """
70 _hierarchy_supports_caching = False
71 """disable cache warnings for all _DDLCompiles subclasses. """
73 def _compiler(self, dialect, **kw):
74 """Return a compiler appropriate for this ClauseElement, given a
75 Dialect."""
77 return dialect.ddl_compiler(dialect, self, **kw)
79 def _compile_w_cache(
80 self,
81 dialect: Dialect,
82 *,
83 compiled_cache: Optional[CompiledCacheType],
84 column_keys: List[str],
85 for_executemany: bool = False,
86 schema_translate_map: Optional[SchemaTranslateMapType] = None,
87 **kw: Any,
88 ) -> Tuple[
89 Compiled, Optional[typing_Sequence[BindParameter[Any]]], CacheStats
90 ]:
91 raise NotImplementedError()
94class DDLIfCallable(Protocol):
95 def __call__(
96 self,
97 ddl: BaseDDLElement,
98 target: Union[SchemaItem, str],
99 bind: Optional[Connection],
100 tables: Optional[List[Table]] = None,
101 state: Optional[Any] = None,
102 *,
103 dialect: Dialect,
104 compiler: Optional[DDLCompiler] = ...,
105 checkfirst: bool,
106 ) -> bool: ...
109class DDLIf(typing.NamedTuple):
110 dialect: Optional[str]
111 callable_: Optional[DDLIfCallable]
112 state: Optional[Any]
114 def _should_execute(
115 self,
116 ddl: BaseDDLElement,
117 target: Union[SchemaItem, str],
118 bind: Optional[Connection],
119 compiler: Optional[DDLCompiler] = None,
120 **kw: Any,
121 ) -> bool:
122 if bind is not None:
123 dialect = bind.dialect
124 elif compiler is not None:
125 dialect = compiler.dialect
126 else:
127 assert False, "compiler or dialect is required"
129 if isinstance(self.dialect, str):
130 if self.dialect != dialect.name:
131 return False
132 elif isinstance(self.dialect, (tuple, list, set)):
133 if dialect.name not in self.dialect:
134 return False
135 if self.callable_ is not None and not self.callable_(
136 ddl,
137 target,
138 bind,
139 state=self.state,
140 dialect=dialect,
141 compiler=compiler,
142 **kw,
143 ):
144 return False
146 return True
149class ExecutableDDLElement(roles.DDLRole, Executable, BaseDDLElement):
150 """Base class for standalone executable DDL expression constructs.
152 This class is the base for the general purpose :class:`.DDL` class,
153 as well as the various create/drop clause constructs such as
154 :class:`.CreateTable`, :class:`.DropTable`, :class:`.AddConstraint`,
155 etc.
157 .. versionchanged:: 2.0 :class:`.ExecutableDDLElement` is renamed from
158 :class:`.DDLElement`, which still exists for backwards compatibility.
160 :class:`.ExecutableDDLElement` integrates closely with SQLAlchemy events,
161 introduced in :ref:`event_toplevel`. An instance of one is
162 itself an event receiving callable::
164 event.listen(
165 users,
166 "after_create",
167 AddConstraint(constraint).execute_if(dialect="postgresql"),
168 )
170 .. seealso::
172 :class:`.DDL`
174 :class:`.DDLEvents`
176 :ref:`event_toplevel`
178 :ref:`schema_ddl_sequences`
180 """
182 _ddl_if: Optional[DDLIf] = None
183 target: Union[SchemaItem, str, None] = None
185 def _execute_on_connection(
186 self, connection, distilled_params, execution_options
187 ):
188 return connection._execute_ddl(
189 self, distilled_params, execution_options
190 )
192 @_generative
193 def against(self, target: SchemaItem) -> Self:
194 """Return a copy of this :class:`_schema.ExecutableDDLElement` which
195 will include the given target.
197 This essentially applies the given item to the ``.target`` attribute of
198 the returned :class:`_schema.ExecutableDDLElement` object. This target
199 is then usable by event handlers and compilation routines in order to
200 provide services such as tokenization of a DDL string in terms of a
201 particular :class:`_schema.Table`.
203 When a :class:`_schema.ExecutableDDLElement` object is established as
204 an event handler for the :meth:`_events.DDLEvents.before_create` or
205 :meth:`_events.DDLEvents.after_create` events, and the event then
206 occurs for a given target such as a :class:`_schema.Constraint` or
207 :class:`_schema.Table`, that target is established with a copy of the
208 :class:`_schema.ExecutableDDLElement` object using this method, which
209 then proceeds to the :meth:`_schema.ExecutableDDLElement.execute`
210 method in order to invoke the actual DDL instruction.
212 :param target: a :class:`_schema.SchemaItem` that will be the subject
213 of a DDL operation.
215 :return: a copy of this :class:`_schema.ExecutableDDLElement` with the
216 ``.target`` attribute assigned to the given
217 :class:`_schema.SchemaItem`.
219 .. seealso::
221 :class:`_schema.DDL` - uses tokenization against the "target" when
222 processing the DDL string.
224 """
225 self.target = target
226 return self
228 @_generative
229 def execute_if(
230 self,
231 dialect: Optional[str] = None,
232 callable_: Optional[DDLIfCallable] = None,
233 state: Optional[Any] = None,
234 ) -> Self:
235 r"""Return a callable that will execute this
236 :class:`_ddl.ExecutableDDLElement` conditionally within an event
237 handler.
239 Used to provide a wrapper for event listening::
241 event.listen(
242 metadata,
243 "before_create",
244 DDL("my_ddl").execute_if(dialect="postgresql"),
245 )
247 :param dialect: May be a string or tuple of strings.
248 If a string, it will be compared to the name of the
249 executing database dialect::
251 DDL("something").execute_if(dialect="postgresql")
253 If a tuple, specifies multiple dialect names::
255 DDL("something").execute_if(dialect=("postgresql", "mysql"))
257 :param callable\_: A callable, which will be invoked with
258 three positional arguments as well as optional keyword
259 arguments:
261 :ddl:
262 This DDL element.
264 :target:
265 The :class:`_schema.Table` or :class:`_schema.MetaData`
266 object which is the
267 target of this event. May be None if the DDL is executed
268 explicitly.
270 :bind:
271 The :class:`_engine.Connection` being used for DDL execution.
272 May be None if this construct is being created inline within
273 a table, in which case ``compiler`` will be present.
275 :tables:
276 Optional keyword argument - a list of Table objects which are to
277 be created/ dropped within a MetaData.create_all() or drop_all()
278 method call.
280 :dialect: keyword argument, but always present - the
281 :class:`.Dialect` involved in the operation.
283 :compiler: keyword argument. Will be ``None`` for an engine
284 level DDL invocation, but will refer to a :class:`.DDLCompiler`
285 if this DDL element is being created inline within a table.
287 :state:
288 Optional keyword argument - will be the ``state`` argument
289 passed to this function.
291 :checkfirst:
292 Keyword argument, will be True if the 'checkfirst' flag was
293 set during the call to ``create()``, ``create_all()``,
294 ``drop()``, ``drop_all()``.
296 If the callable returns a True value, the DDL statement will be
297 executed.
299 :param state: any value which will be passed to the callable\_
300 as the ``state`` keyword argument.
302 .. seealso::
304 :meth:`.SchemaItem.ddl_if`
306 :class:`.DDLEvents`
308 :ref:`event_toplevel`
310 """
311 self._ddl_if = DDLIf(dialect, callable_, state)
312 return self
314 def _should_execute(self, target, bind, **kw):
315 if self._ddl_if is None:
316 return True
317 else:
318 return self._ddl_if._should_execute(self, target, bind, **kw)
320 def _invoke_with(self, bind):
321 if self._should_execute(self.target, bind):
322 return bind.execute(self)
324 def __call__(self, target, bind, **kw):
325 """Execute the DDL as a ddl_listener."""
327 self.against(target)._invoke_with(bind)
329 def _generate(self):
330 s = self.__class__.__new__(self.__class__)
331 s.__dict__ = self.__dict__.copy()
332 return s
335DDLElement = ExecutableDDLElement
336""":class:`.DDLElement` is renamed to :class:`.ExecutableDDLElement`."""
339class DDL(ExecutableDDLElement):
340 """A literal DDL statement.
342 Specifies literal SQL DDL to be executed by the database. DDL objects
343 function as DDL event listeners, and can be subscribed to those events
344 listed in :class:`.DDLEvents`, using either :class:`_schema.Table` or
345 :class:`_schema.MetaData` objects as targets.
346 Basic templating support allows
347 a single DDL instance to handle repetitive tasks for multiple tables.
349 Examples::
351 from sqlalchemy import event, DDL
353 tbl = Table("users", metadata, Column("uid", Integer))
354 event.listen(tbl, "before_create", DDL("DROP TRIGGER users_trigger"))
356 spow = DDL("ALTER TABLE %(table)s SET secretpowers TRUE")
357 event.listen(tbl, "after_create", spow.execute_if(dialect="somedb"))
359 drop_spow = DDL("ALTER TABLE users SET secretpowers FALSE")
360 connection.execute(drop_spow)
362 When operating on Table events, the following ``statement``
363 string substitutions are available:
365 .. sourcecode:: text
367 %(table)s - the Table name, with any required quoting applied
368 %(schema)s - the schema name, with any required quoting applied
369 %(fullname)s - the Table name including schema, quoted if needed
371 The DDL's "context", if any, will be combined with the standard
372 substitutions noted above. Keys present in the context will override
373 the standard substitutions.
375 """
377 __visit_name__ = "ddl"
379 def __init__(self, statement, context=None):
380 """Create a DDL statement.
382 :param statement:
383 A string or unicode string to be executed. Statements will be
384 processed with Python's string formatting operator using
385 a fixed set of string substitutions, as well as additional
386 substitutions provided by the optional :paramref:`.DDL.context`
387 parameter.
389 A literal '%' in a statement must be escaped as '%%'.
391 SQL bind parameters are not available in DDL statements.
393 :param context:
394 Optional dictionary, defaults to None. These values will be
395 available for use in string substitutions on the DDL statement.
397 .. seealso::
399 :class:`.DDLEvents`
401 :ref:`event_toplevel`
403 """
405 if not isinstance(statement, str):
406 raise exc.ArgumentError(
407 "Expected a string or unicode SQL statement, got '%r'"
408 % statement
409 )
411 self.statement = statement
412 self.context = context or {}
414 def __repr__(self):
415 parts = [repr(self.statement)]
416 if self.context:
417 parts.append(f"context={self.context}")
419 return "<%s@%s; %s>" % (
420 type(self).__name__,
421 id(self),
422 ", ".join(parts),
423 )
426class _CreateDropBase(ExecutableDDLElement, Generic[_SI]):
427 """Base class for DDL constructs that represent CREATE and DROP or
428 equivalents.
430 The common theme of _CreateDropBase is a single
431 ``element`` attribute which refers to the element
432 to be created or dropped.
434 """
436 element: _SI
438 def __init__(self, element: _SI) -> None:
439 self.element = self.target = element
440 self._ddl_if = getattr(element, "_ddl_if", None)
442 @property
443 def stringify_dialect(self): # type: ignore[override]
444 assert not isinstance(self.element, str)
445 return self.element.create_drop_stringify_dialect
447 def _create_rule_disable(self, compiler):
448 """Allow disable of _create_rule using a callable.
450 Pass to _create_rule using
451 util.portable_instancemethod(self._create_rule_disable)
452 to retain serializability.
454 """
455 return False
458class _CreateBase(_CreateDropBase[_SI]):
459 def __init__(self, element: _SI, if_not_exists: bool = False) -> None:
460 super().__init__(element)
461 self.if_not_exists = if_not_exists
464class _DropBase(_CreateDropBase[_SI]):
465 def __init__(self, element: _SI, if_exists: bool = False) -> None:
466 super().__init__(element)
467 self.if_exists = if_exists
470class CreateSchema(_CreateBase[str]):
471 """Represent a CREATE SCHEMA statement.
473 The argument here is the string name of the schema.
475 """
477 __visit_name__ = "create_schema"
479 stringify_dialect = "default"
481 def __init__(
482 self,
483 name: str,
484 if_not_exists: bool = False,
485 ) -> None:
486 """Create a new :class:`.CreateSchema` construct."""
488 super().__init__(element=name, if_not_exists=if_not_exists)
491class DropSchema(_DropBase[str]):
492 """Represent a DROP SCHEMA statement.
494 The argument here is the string name of the schema.
496 """
498 __visit_name__ = "drop_schema"
500 stringify_dialect = "default"
502 def __init__(
503 self,
504 name: str,
505 cascade: bool = False,
506 if_exists: bool = False,
507 ) -> None:
508 """Create a new :class:`.DropSchema` construct."""
510 super().__init__(element=name, if_exists=if_exists)
511 self.cascade = cascade
514class CreateTable(_CreateBase["Table"]):
515 """Represent a CREATE TABLE statement."""
517 __visit_name__ = "create_table"
519 def __init__(
520 self,
521 element: Table,
522 include_foreign_key_constraints: Optional[
523 typing_Sequence[ForeignKeyConstraint]
524 ] = None,
525 if_not_exists: bool = False,
526 ) -> None:
527 """Create a :class:`.CreateTable` construct.
529 :param element: a :class:`_schema.Table` that's the subject
530 of the CREATE
531 :param on: See the description for 'on' in :class:`.DDL`.
532 :param include_foreign_key_constraints: optional sequence of
533 :class:`_schema.ForeignKeyConstraint` objects that will be included
534 inline within the CREATE construct; if omitted, all foreign key
535 constraints that do not specify use_alter=True are included.
537 :param if_not_exists: if True, an IF NOT EXISTS operator will be
538 applied to the construct.
540 .. versionadded:: 1.4.0b2
542 """
543 super().__init__(element, if_not_exists=if_not_exists)
544 self.columns = [CreateColumn(column) for column in element.columns]
545 self.include_foreign_key_constraints = include_foreign_key_constraints
548class _DropView(_DropBase["Table"]):
549 """Semi-public 'DROP VIEW' construct.
551 Used by the test suite for dialect-agnostic drops of views.
552 This object will eventually be part of a public "view" API.
554 """
556 __visit_name__ = "drop_view"
559class CreateConstraint(BaseDDLElement):
560 element: Constraint
562 def __init__(self, element: Constraint) -> None:
563 self.element = element
566class CreateColumn(BaseDDLElement):
567 """Represent a :class:`_schema.Column`
568 as rendered in a CREATE TABLE statement,
569 via the :class:`.CreateTable` construct.
571 This is provided to support custom column DDL within the generation
572 of CREATE TABLE statements, by using the
573 compiler extension documented in :ref:`sqlalchemy.ext.compiler_toplevel`
574 to extend :class:`.CreateColumn`.
576 Typical integration is to examine the incoming :class:`_schema.Column`
577 object, and to redirect compilation if a particular flag or condition
578 is found::
580 from sqlalchemy import schema
581 from sqlalchemy.ext.compiler import compiles
584 @compiles(schema.CreateColumn)
585 def compile(element, compiler, **kw):
586 column = element.element
588 if "special" not in column.info:
589 return compiler.visit_create_column(element, **kw)
591 text = "%s SPECIAL DIRECTIVE %s" % (
592 column.name,
593 compiler.type_compiler.process(column.type),
594 )
595 default = compiler.get_column_default_string(column)
596 if default is not None:
597 text += " DEFAULT " + default
599 if not column.nullable:
600 text += " NOT NULL"
602 if column.constraints:
603 text += " ".join(
604 compiler.process(const) for const in column.constraints
605 )
606 return text
608 The above construct can be applied to a :class:`_schema.Table`
609 as follows::
611 from sqlalchemy import Table, Metadata, Column, Integer, String
612 from sqlalchemy import schema
614 metadata = MetaData()
616 table = Table(
617 "mytable",
618 MetaData(),
619 Column("x", Integer, info={"special": True}, primary_key=True),
620 Column("y", String(50)),
621 Column("z", String(20), info={"special": True}),
622 )
624 metadata.create_all(conn)
626 Above, the directives we've added to the :attr:`_schema.Column.info`
627 collection
628 will be detected by our custom compilation scheme:
630 .. sourcecode:: sql
632 CREATE TABLE mytable (
633 x SPECIAL DIRECTIVE INTEGER NOT NULL,
634 y VARCHAR(50),
635 z SPECIAL DIRECTIVE VARCHAR(20),
636 PRIMARY KEY (x)
637 )
639 The :class:`.CreateColumn` construct can also be used to skip certain
640 columns when producing a ``CREATE TABLE``. This is accomplished by
641 creating a compilation rule that conditionally returns ``None``.
642 This is essentially how to produce the same effect as using the
643 ``system=True`` argument on :class:`_schema.Column`, which marks a column
644 as an implicitly-present "system" column.
646 For example, suppose we wish to produce a :class:`_schema.Table`
647 which skips
648 rendering of the PostgreSQL ``xmin`` column against the PostgreSQL
649 backend, but on other backends does render it, in anticipation of a
650 triggered rule. A conditional compilation rule could skip this name only
651 on PostgreSQL::
653 from sqlalchemy.schema import CreateColumn
656 @compiles(CreateColumn, "postgresql")
657 def skip_xmin(element, compiler, **kw):
658 if element.element.name == "xmin":
659 return None
660 else:
661 return compiler.visit_create_column(element, **kw)
664 my_table = Table(
665 "mytable",
666 metadata,
667 Column("id", Integer, primary_key=True),
668 Column("xmin", Integer),
669 )
671 Above, a :class:`.CreateTable` construct will generate a ``CREATE TABLE``
672 which only includes the ``id`` column in the string; the ``xmin`` column
673 will be omitted, but only against the PostgreSQL backend.
675 """
677 __visit_name__ = "create_column"
679 element: Column[Any]
681 def __init__(self, element: Column[Any]) -> None:
682 self.element = element
685class DropTable(_DropBase["Table"]):
686 """Represent a DROP TABLE statement."""
688 __visit_name__ = "drop_table"
690 def __init__(self, element: Table, if_exists: bool = False) -> None:
691 """Create a :class:`.DropTable` construct.
693 :param element: a :class:`_schema.Table` that's the subject
694 of the DROP.
695 :param on: See the description for 'on' in :class:`.DDL`.
696 :param if_exists: if True, an IF EXISTS operator will be applied to the
697 construct.
699 .. versionadded:: 1.4.0b2
701 """
702 super().__init__(element, if_exists=if_exists)
705class CreateSequence(_CreateBase["Sequence"]):
706 """Represent a CREATE SEQUENCE statement."""
708 __visit_name__ = "create_sequence"
711class DropSequence(_DropBase["Sequence"]):
712 """Represent a DROP SEQUENCE statement."""
714 __visit_name__ = "drop_sequence"
717class CreateIndex(_CreateBase["Index"]):
718 """Represent a CREATE INDEX statement."""
720 __visit_name__ = "create_index"
722 def __init__(self, element: Index, if_not_exists: bool = False) -> None:
723 """Create a :class:`.Createindex` construct.
725 :param element: a :class:`_schema.Index` that's the subject
726 of the CREATE.
727 :param if_not_exists: if True, an IF NOT EXISTS operator will be
728 applied to the construct.
730 .. versionadded:: 1.4.0b2
732 """
733 super().__init__(element, if_not_exists=if_not_exists)
736class DropIndex(_DropBase["Index"]):
737 """Represent a DROP INDEX statement."""
739 __visit_name__ = "drop_index"
741 def __init__(self, element: Index, if_exists: bool = False) -> None:
742 """Create a :class:`.DropIndex` construct.
744 :param element: a :class:`_schema.Index` that's the subject
745 of the DROP.
746 :param if_exists: if True, an IF EXISTS operator will be applied to the
747 construct.
749 .. versionadded:: 1.4.0b2
751 """
752 super().__init__(element, if_exists=if_exists)
755class AddConstraint(_CreateBase["Constraint"]):
756 """Represent an ALTER TABLE ADD CONSTRAINT statement."""
758 __visit_name__ = "add_constraint"
760 def __init__(
761 self,
762 element: Constraint,
763 *,
764 isolate_from_table: bool = True,
765 ) -> None:
766 """Construct a new :class:`.AddConstraint` construct.
768 :param element: a :class:`.Constraint` object
770 :param isolate_from_table: optional boolean, defaults to True. Has
771 the effect of the incoming constraint being isolated from being
772 included in a CREATE TABLE sequence when associated with a
773 :class:`.Table`.
775 .. versionadded:: 2.0.39 - added
776 :paramref:`.AddConstraint.isolate_from_table`, defaulting
777 to True. Previously, the behavior of this parameter was implicitly
778 turned on in all cases.
780 """
781 super().__init__(element)
783 if isolate_from_table:
784 element._create_rule = util.portable_instancemethod(
785 self._create_rule_disable
786 )
789class DropConstraint(_DropBase["Constraint"]):
790 """Represent an ALTER TABLE DROP CONSTRAINT statement."""
792 __visit_name__ = "drop_constraint"
794 def __init__(
795 self,
796 element: Constraint,
797 *,
798 cascade: bool = False,
799 if_exists: bool = False,
800 isolate_from_table: bool = True,
801 **kw: Any,
802 ) -> None:
803 """Construct a new :class:`.DropConstraint` construct.
805 :param element: a :class:`.Constraint` object
806 :param cascade: optional boolean, indicates backend-specific
807 "CASCADE CONSTRAINT" directive should be rendered if available
808 :param if_exists: optional boolean, indicates backend-specific
809 "IF EXISTS" directive should be rendered if available
810 :param isolate_from_table: optional boolean, defaults to True. Has
811 the effect of the incoming constraint being isolated from being
812 included in a CREATE TABLE sequence when associated with a
813 :class:`.Table`.
815 .. versionadded:: 2.0.39 - added
816 :paramref:`.DropConstraint.isolate_from_table`, defaulting
817 to True. Previously, the behavior of this parameter was implicitly
818 turned on in all cases.
820 """
821 self.cascade = cascade
822 super().__init__(element, if_exists=if_exists, **kw)
824 if isolate_from_table:
825 element._create_rule = util.portable_instancemethod(
826 self._create_rule_disable
827 )
830class SetTableComment(_CreateDropBase["Table"]):
831 """Represent a COMMENT ON TABLE IS statement."""
833 __visit_name__ = "set_table_comment"
836class DropTableComment(_CreateDropBase["Table"]):
837 """Represent a COMMENT ON TABLE '' statement.
839 Note this varies a lot across database backends.
841 """
843 __visit_name__ = "drop_table_comment"
846class SetColumnComment(_CreateDropBase["Column[Any]"]):
847 """Represent a COMMENT ON COLUMN IS statement."""
849 __visit_name__ = "set_column_comment"
852class DropColumnComment(_CreateDropBase["Column[Any]"]):
853 """Represent a COMMENT ON COLUMN IS NULL statement."""
855 __visit_name__ = "drop_column_comment"
858class SetConstraintComment(_CreateDropBase["Constraint"]):
859 """Represent a COMMENT ON CONSTRAINT IS statement."""
861 __visit_name__ = "set_constraint_comment"
864class DropConstraintComment(_CreateDropBase["Constraint"]):
865 """Represent a COMMENT ON CONSTRAINT IS NULL statement."""
867 __visit_name__ = "drop_constraint_comment"
870class InvokeDDLBase(SchemaVisitor):
871 def __init__(self, connection, **kw):
872 self.connection = connection
873 assert not kw, f"Unexpected keywords: {kw.keys()}"
875 @contextlib.contextmanager
876 def with_ddl_events(self, target, **kw):
877 """helper context manager that will apply appropriate DDL events
878 to a CREATE or DROP operation."""
880 raise NotImplementedError()
883class InvokeCreateDDLBase(InvokeDDLBase):
884 @contextlib.contextmanager
885 def with_ddl_events(self, target, **kw):
886 """helper context manager that will apply appropriate DDL events
887 to a CREATE or DROP operation."""
889 target.dispatch.before_create(
890 target, self.connection, _ddl_runner=self, **kw
891 )
892 yield
893 target.dispatch.after_create(
894 target, self.connection, _ddl_runner=self, **kw
895 )
898class InvokeDropDDLBase(InvokeDDLBase):
899 @contextlib.contextmanager
900 def with_ddl_events(self, target, **kw):
901 """helper context manager that will apply appropriate DDL events
902 to a CREATE or DROP operation."""
904 target.dispatch.before_drop(
905 target, self.connection, _ddl_runner=self, **kw
906 )
907 yield
908 target.dispatch.after_drop(
909 target, self.connection, _ddl_runner=self, **kw
910 )
913class SchemaGenerator(InvokeCreateDDLBase):
914 def __init__(
915 self, dialect, connection, checkfirst=False, tables=None, **kwargs
916 ):
917 super().__init__(connection, **kwargs)
918 self.checkfirst = checkfirst
919 self.tables = tables
920 self.preparer = dialect.identifier_preparer
921 self.dialect = dialect
922 self.memo = {}
924 def _can_create_table(self, table):
925 self.dialect.validate_identifier(table.name)
926 effective_schema = self.connection.schema_for_object(table)
927 if effective_schema:
928 self.dialect.validate_identifier(effective_schema)
929 return not self.checkfirst or not self.dialect.has_table(
930 self.connection, table.name, schema=effective_schema
931 )
933 def _can_create_index(self, index):
934 effective_schema = self.connection.schema_for_object(index.table)
935 if effective_schema:
936 self.dialect.validate_identifier(effective_schema)
937 return not self.checkfirst or not self.dialect.has_index(
938 self.connection,
939 index.table.name,
940 index.name,
941 schema=effective_schema,
942 )
944 def _can_create_sequence(self, sequence):
945 effective_schema = self.connection.schema_for_object(sequence)
947 return self.dialect.supports_sequences and (
948 (not self.dialect.sequences_optional or not sequence.optional)
949 and (
950 not self.checkfirst
951 or not self.dialect.has_sequence(
952 self.connection, sequence.name, schema=effective_schema
953 )
954 )
955 )
957 def visit_metadata(self, metadata):
958 if self.tables is not None:
959 tables = self.tables
960 else:
961 tables = list(metadata.tables.values())
963 collection = sort_tables_and_constraints(
964 [t for t in tables if self._can_create_table(t)]
965 )
967 seq_coll = [
968 s
969 for s in metadata._sequences.values()
970 if s.column is None and self._can_create_sequence(s)
971 ]
973 event_collection = [t for (t, fks) in collection if t is not None]
975 with self.with_ddl_events(
976 metadata,
977 tables=event_collection,
978 checkfirst=self.checkfirst,
979 ):
980 for seq in seq_coll:
981 self.traverse_single(seq, create_ok=True)
983 for table, fkcs in collection:
984 if table is not None:
985 self.traverse_single(
986 table,
987 create_ok=True,
988 include_foreign_key_constraints=fkcs,
989 _is_metadata_operation=True,
990 )
991 else:
992 for fkc in fkcs:
993 self.traverse_single(fkc)
995 def visit_table(
996 self,
997 table,
998 create_ok=False,
999 include_foreign_key_constraints=None,
1000 _is_metadata_operation=False,
1001 ):
1002 if not create_ok and not self._can_create_table(table):
1003 return
1005 with self.with_ddl_events(
1006 table,
1007 checkfirst=self.checkfirst,
1008 _is_metadata_operation=_is_metadata_operation,
1009 ):
1010 for column in table.columns:
1011 if column.default is not None:
1012 self.traverse_single(column.default)
1014 if not self.dialect.supports_alter:
1015 # e.g., don't omit any foreign key constraints
1016 include_foreign_key_constraints = None
1018 CreateTable(
1019 table,
1020 include_foreign_key_constraints=(
1021 include_foreign_key_constraints
1022 ),
1023 )._invoke_with(self.connection)
1025 if hasattr(table, "indexes"):
1026 for index in table.indexes:
1027 self.traverse_single(index, create_ok=True)
1029 if (
1030 self.dialect.supports_comments
1031 and not self.dialect.inline_comments
1032 ):
1033 if table.comment is not None:
1034 SetTableComment(table)._invoke_with(self.connection)
1036 for column in table.columns:
1037 if column.comment is not None:
1038 SetColumnComment(column)._invoke_with(self.connection)
1040 if self.dialect.supports_constraint_comments:
1041 for constraint in table.constraints:
1042 if constraint.comment is not None:
1043 self.connection.execute(
1044 SetConstraintComment(constraint)
1045 )
1047 def visit_foreign_key_constraint(self, constraint):
1048 if not self.dialect.supports_alter:
1049 return
1051 with self.with_ddl_events(constraint):
1052 AddConstraint(constraint)._invoke_with(self.connection)
1054 def visit_sequence(self, sequence, create_ok=False):
1055 if not create_ok and not self._can_create_sequence(sequence):
1056 return
1057 with self.with_ddl_events(sequence):
1058 CreateSequence(sequence)._invoke_with(self.connection)
1060 def visit_index(self, index, create_ok=False):
1061 if not create_ok and not self._can_create_index(index):
1062 return
1063 with self.with_ddl_events(index):
1064 CreateIndex(index)._invoke_with(self.connection)
1067class SchemaDropper(InvokeDropDDLBase):
1068 def __init__(
1069 self, dialect, connection, checkfirst=False, tables=None, **kwargs
1070 ):
1071 super().__init__(connection, **kwargs)
1072 self.checkfirst = checkfirst
1073 self.tables = tables
1074 self.preparer = dialect.identifier_preparer
1075 self.dialect = dialect
1076 self.memo = {}
1078 def visit_metadata(self, metadata):
1079 if self.tables is not None:
1080 tables = self.tables
1081 else:
1082 tables = list(metadata.tables.values())
1084 try:
1085 unsorted_tables = [t for t in tables if self._can_drop_table(t)]
1086 collection = list(
1087 reversed(
1088 sort_tables_and_constraints(
1089 unsorted_tables,
1090 filter_fn=lambda constraint: (
1091 False
1092 if not self.dialect.supports_alter
1093 or constraint.name is None
1094 else None
1095 ),
1096 )
1097 )
1098 )
1099 except exc.CircularDependencyError as err2:
1100 if not self.dialect.supports_alter:
1101 util.warn(
1102 "Can't sort tables for DROP; an "
1103 "unresolvable foreign key "
1104 "dependency exists between tables: %s; and backend does "
1105 "not support ALTER. To restore at least a partial sort, "
1106 "apply use_alter=True to ForeignKey and "
1107 "ForeignKeyConstraint "
1108 "objects involved in the cycle to mark these as known "
1109 "cycles that will be ignored."
1110 % (", ".join(sorted([t.fullname for t in err2.cycles])))
1111 )
1112 collection = [(t, ()) for t in unsorted_tables]
1113 else:
1114 raise exc.CircularDependencyError(
1115 err2.args[0],
1116 err2.cycles,
1117 err2.edges,
1118 msg="Can't sort tables for DROP; an "
1119 "unresolvable foreign key "
1120 "dependency exists between tables: %s. Please ensure "
1121 "that the ForeignKey and ForeignKeyConstraint objects "
1122 "involved in the cycle have "
1123 "names so that they can be dropped using "
1124 "DROP CONSTRAINT."
1125 % (", ".join(sorted([t.fullname for t in err2.cycles]))),
1126 ) from err2
1128 seq_coll = [
1129 s
1130 for s in metadata._sequences.values()
1131 if self._can_drop_sequence(s)
1132 ]
1134 event_collection = [t for (t, fks) in collection if t is not None]
1136 with self.with_ddl_events(
1137 metadata,
1138 tables=event_collection,
1139 checkfirst=self.checkfirst,
1140 ):
1141 for table, fkcs in collection:
1142 if table is not None:
1143 self.traverse_single(
1144 table,
1145 drop_ok=True,
1146 _is_metadata_operation=True,
1147 _ignore_sequences=seq_coll,
1148 )
1149 else:
1150 for fkc in fkcs:
1151 self.traverse_single(fkc)
1153 for seq in seq_coll:
1154 self.traverse_single(seq, drop_ok=seq.column is None)
1156 def _can_drop_table(self, table):
1157 self.dialect.validate_identifier(table.name)
1158 effective_schema = self.connection.schema_for_object(table)
1159 if effective_schema:
1160 self.dialect.validate_identifier(effective_schema)
1161 return not self.checkfirst or self.dialect.has_table(
1162 self.connection, table.name, schema=effective_schema
1163 )
1165 def _can_drop_index(self, index):
1166 effective_schema = self.connection.schema_for_object(index.table)
1167 if effective_schema:
1168 self.dialect.validate_identifier(effective_schema)
1169 return not self.checkfirst or self.dialect.has_index(
1170 self.connection,
1171 index.table.name,
1172 index.name,
1173 schema=effective_schema,
1174 )
1176 def _can_drop_sequence(self, sequence):
1177 effective_schema = self.connection.schema_for_object(sequence)
1178 return self.dialect.supports_sequences and (
1179 (not self.dialect.sequences_optional or not sequence.optional)
1180 and (
1181 not self.checkfirst
1182 or self.dialect.has_sequence(
1183 self.connection, sequence.name, schema=effective_schema
1184 )
1185 )
1186 )
1188 def visit_index(self, index, drop_ok=False):
1189 if not drop_ok and not self._can_drop_index(index):
1190 return
1192 with self.with_ddl_events(index):
1193 DropIndex(index)(index, self.connection)
1195 def visit_table(
1196 self,
1197 table,
1198 drop_ok=False,
1199 _is_metadata_operation=False,
1200 _ignore_sequences=(),
1201 ):
1202 if not drop_ok and not self._can_drop_table(table):
1203 return
1205 with self.with_ddl_events(
1206 table,
1207 checkfirst=self.checkfirst,
1208 _is_metadata_operation=_is_metadata_operation,
1209 ):
1210 DropTable(table)._invoke_with(self.connection)
1212 # traverse client side defaults which may refer to server-side
1213 # sequences. noting that some of these client side defaults may
1214 # also be set up as server side defaults
1215 # (see https://docs.sqlalchemy.org/en/
1216 # latest/core/defaults.html
1217 # #associating-a-sequence-as-the-server-side-
1218 # default), so have to be dropped after the table is dropped.
1219 for column in table.columns:
1220 if (
1221 column.default is not None
1222 and column.default not in _ignore_sequences
1223 ):
1224 self.traverse_single(column.default)
1226 def visit_foreign_key_constraint(self, constraint):
1227 if not self.dialect.supports_alter:
1228 return
1229 with self.with_ddl_events(constraint):
1230 DropConstraint(constraint)._invoke_with(self.connection)
1232 def visit_sequence(self, sequence, drop_ok=False):
1233 if not drop_ok and not self._can_drop_sequence(sequence):
1234 return
1235 with self.with_ddl_events(sequence):
1236 DropSequence(sequence)._invoke_with(self.connection)
1239def sort_tables(
1240 tables: Iterable[TableClause],
1241 skip_fn: Optional[Callable[[ForeignKeyConstraint], bool]] = None,
1242 extra_dependencies: Optional[
1243 typing_Sequence[Tuple[TableClause, TableClause]]
1244 ] = None,
1245) -> List[Table]:
1246 """Sort a collection of :class:`_schema.Table` objects based on
1247 dependency.
1249 This is a dependency-ordered sort which will emit :class:`_schema.Table`
1250 objects such that they will follow their dependent :class:`_schema.Table`
1251 objects.
1252 Tables are dependent on another based on the presence of
1253 :class:`_schema.ForeignKeyConstraint`
1254 objects as well as explicit dependencies
1255 added by :meth:`_schema.Table.add_is_dependent_on`.
1257 .. warning::
1259 The :func:`._schema.sort_tables` function cannot by itself
1260 accommodate automatic resolution of dependency cycles between
1261 tables, which are usually caused by mutually dependent foreign key
1262 constraints. When these cycles are detected, the foreign keys
1263 of these tables are omitted from consideration in the sort.
1264 A warning is emitted when this condition occurs, which will be an
1265 exception raise in a future release. Tables which are not part
1266 of the cycle will still be returned in dependency order.
1268 To resolve these cycles, the
1269 :paramref:`_schema.ForeignKeyConstraint.use_alter` parameter may be
1270 applied to those constraints which create a cycle. Alternatively,
1271 the :func:`_schema.sort_tables_and_constraints` function will
1272 automatically return foreign key constraints in a separate
1273 collection when cycles are detected so that they may be applied
1274 to a schema separately.
1276 .. versionchanged:: 1.3.17 - a warning is emitted when
1277 :func:`_schema.sort_tables` cannot perform a proper sort due to
1278 cyclical dependencies. This will be an exception in a future
1279 release. Additionally, the sort will continue to return
1280 other tables not involved in the cycle in dependency order
1281 which was not the case previously.
1283 :param tables: a sequence of :class:`_schema.Table` objects.
1285 :param skip_fn: optional callable which will be passed a
1286 :class:`_schema.ForeignKeyConstraint` object; if it returns True, this
1287 constraint will not be considered as a dependency. Note this is
1288 **different** from the same parameter in
1289 :func:`.sort_tables_and_constraints`, which is
1290 instead passed the owning :class:`_schema.ForeignKeyConstraint` object.
1292 :param extra_dependencies: a sequence of 2-tuples of tables which will
1293 also be considered as dependent on each other.
1295 .. seealso::
1297 :func:`.sort_tables_and_constraints`
1299 :attr:`_schema.MetaData.sorted_tables` - uses this function to sort
1302 """
1304 if skip_fn is not None:
1305 fixed_skip_fn = skip_fn
1307 def _skip_fn(fkc):
1308 for fk in fkc.elements:
1309 if fixed_skip_fn(fk):
1310 return True
1311 else:
1312 return None
1314 else:
1315 _skip_fn = None # type: ignore
1317 return [
1318 t
1319 for (t, fkcs) in sort_tables_and_constraints(
1320 tables,
1321 filter_fn=_skip_fn,
1322 extra_dependencies=extra_dependencies,
1323 _warn_for_cycles=True,
1324 )
1325 if t is not None
1326 ]
1329def sort_tables_and_constraints(
1330 tables, filter_fn=None, extra_dependencies=None, _warn_for_cycles=False
1331):
1332 """Sort a collection of :class:`_schema.Table` /
1333 :class:`_schema.ForeignKeyConstraint`
1334 objects.
1336 This is a dependency-ordered sort which will emit tuples of
1337 ``(Table, [ForeignKeyConstraint, ...])`` such that each
1338 :class:`_schema.Table` follows its dependent :class:`_schema.Table`
1339 objects.
1340 Remaining :class:`_schema.ForeignKeyConstraint`
1341 objects that are separate due to
1342 dependency rules not satisfied by the sort are emitted afterwards
1343 as ``(None, [ForeignKeyConstraint ...])``.
1345 Tables are dependent on another based on the presence of
1346 :class:`_schema.ForeignKeyConstraint` objects, explicit dependencies
1347 added by :meth:`_schema.Table.add_is_dependent_on`,
1348 as well as dependencies
1349 stated here using the :paramref:`~.sort_tables_and_constraints.skip_fn`
1350 and/or :paramref:`~.sort_tables_and_constraints.extra_dependencies`
1351 parameters.
1353 :param tables: a sequence of :class:`_schema.Table` objects.
1355 :param filter_fn: optional callable which will be passed a
1356 :class:`_schema.ForeignKeyConstraint` object,
1357 and returns a value based on
1358 whether this constraint should definitely be included or excluded as
1359 an inline constraint, or neither. If it returns False, the constraint
1360 will definitely be included as a dependency that cannot be subject
1361 to ALTER; if True, it will **only** be included as an ALTER result at
1362 the end. Returning None means the constraint is included in the
1363 table-based result unless it is detected as part of a dependency cycle.
1365 :param extra_dependencies: a sequence of 2-tuples of tables which will
1366 also be considered as dependent on each other.
1368 .. seealso::
1370 :func:`.sort_tables`
1373 """
1375 fixed_dependencies = set()
1376 mutable_dependencies = set()
1378 if extra_dependencies is not None:
1379 fixed_dependencies.update(extra_dependencies)
1381 remaining_fkcs = set()
1382 for table in tables:
1383 for fkc in table.foreign_key_constraints:
1384 if fkc.use_alter is True:
1385 remaining_fkcs.add(fkc)
1386 continue
1388 if filter_fn:
1389 filtered = filter_fn(fkc)
1391 if filtered is True:
1392 remaining_fkcs.add(fkc)
1393 continue
1395 dependent_on = fkc.referred_table
1396 if dependent_on is not table:
1397 mutable_dependencies.add((dependent_on, table))
1399 fixed_dependencies.update(
1400 (parent, table) for parent in table._extra_dependencies
1401 )
1403 try:
1404 candidate_sort = list(
1405 topological.sort(
1406 fixed_dependencies.union(mutable_dependencies),
1407 tables,
1408 )
1409 )
1410 except exc.CircularDependencyError as err:
1411 if _warn_for_cycles:
1412 util.warn(
1413 "Cannot correctly sort tables; there are unresolvable cycles "
1414 'between tables "%s", which is usually caused by mutually '
1415 "dependent foreign key constraints. Foreign key constraints "
1416 "involving these tables will not be considered; this warning "
1417 "may raise an error in a future release."
1418 % (", ".join(sorted(t.fullname for t in err.cycles)),)
1419 )
1420 for edge in err.edges:
1421 if edge in mutable_dependencies:
1422 table = edge[1]
1423 if table not in err.cycles:
1424 continue
1425 can_remove = [
1426 fkc
1427 for fkc in table.foreign_key_constraints
1428 if filter_fn is None or filter_fn(fkc) is not False
1429 ]
1430 remaining_fkcs.update(can_remove)
1431 for fkc in can_remove:
1432 dependent_on = fkc.referred_table
1433 if dependent_on is not table:
1434 mutable_dependencies.discard((dependent_on, table))
1435 candidate_sort = list(
1436 topological.sort(
1437 fixed_dependencies.union(mutable_dependencies),
1438 tables,
1439 )
1440 )
1442 return [
1443 (table, table.foreign_key_constraints.difference(remaining_fkcs))
1444 for table in candidate_sort
1445 ] + [(None, list(remaining_fkcs))]