# sql/ddl.py

# Copyright (C) 2009-2026 the SQLAlchemy authors and contributors

# <see AUTHORS file>

#

# This module is part of SQLAlchemy and is released under

# the MIT License: https://www.opensource.org/licenses/mit-license.php

# mypy: allow-untyped-defs, allow-untyped-calls

"""

Provides the hierarchy of DDL-defining schema items as well as routines

to invoke them for a create/drop call.

"""

from __future__ import annotations

import contextlib

from enum import auto

from enum import Flag

import typing

from typing import Any

from typing import Callable

from typing import Generic

from typing import Iterable

from typing import List

from typing import Optional

from typing import Protocol

from typing import Sequence as typing_Sequence

from typing import Tuple

from typing import TypeVar

from typing import Union

from . import coercions

from . import roles

from . import util as sql_util

from .base import _generative

from .base import _NoArg

from .base import DialectKWArgs

from .base import Executable

from .base import NO_ARG

from .base import SchemaVisitor

from .elements import ClauseElement

from .selectable import SelectBase

from .selectable import TableClause

from .. import exc

from .. import util

from ..util import topological

from ..util.typing import Self

if typing.TYPE_CHECKING:

from .compiler import Compiled

from .compiler import DDLCompiler

from .elements import BindParameter

from .schema import Column

from .schema import Constraint

from .schema import ForeignKeyConstraint

from .schema import Index

from .schema import MetaData

from .schema import SchemaItem

from .schema import Sequence as Sequence # noqa: F401

from .schema import Table

from ..engine.base import Connection

from ..engine.interfaces import _CoreSingleExecuteParams

from ..engine.interfaces import CacheStats

from ..engine.interfaces import CompiledCacheType

from ..engine.interfaces import Dialect

from ..engine.interfaces import SchemaTranslateMapType

_SI = TypeVar("_SI", bound=Union["SchemaItem", str])

class BaseDDLElement(ClauseElement):

"""The root of DDL constructs, including those that are sub-elements

within the "create table" and other processes.

.. versionadded:: 2.0

"""

_hierarchy_supports_caching = False

"""disable cache warnings for all _DDLCompiles subclasses. """

def _compiler(self, dialect, **kw):

"""Return a compiler appropriate for this ClauseElement, given a

Dialect."""

return dialect.ddl_compiler(dialect, self, **kw)

def _compile_w_cache(

self,

dialect: Dialect,

*,

compiled_cache: Optional[CompiledCacheType],

column_keys: List[str],

for_executemany: bool = False,

schema_translate_map: Optional[SchemaTranslateMapType] = None,

**kw: Any,

) -> tuple[

Compiled,

typing_Sequence[BindParameter[Any]] | None,

_CoreSingleExecuteParams | None,

CacheStats,

]:

raise NotImplementedError()

class DDLIfCallable(Protocol):

def __call__(

self,

ddl: BaseDDLElement,

target: Union[SchemaItem, str],

bind: Optional[Connection],

tables: Optional[List[Table]] = None,

state: Optional[Any] = None,

*,

dialect: Dialect,

compiler: Optional[DDLCompiler] = ...,

checkfirst: bool,

) -> bool: ...

class DDLIf(typing.NamedTuple):

dialect: Optional[str]

callable_: Optional[DDLIfCallable]

state: Optional[Any]

def _should_execute(

self,

ddl: BaseDDLElement,

target: Union[SchemaItem, str],

bind: Optional[Connection],

compiler: Optional[DDLCompiler] = None,

**kw: Any,

) -> bool:

if bind is not None:

dialect = bind.dialect

elif compiler is not None:

dialect = compiler.dialect

else:

assert False, "compiler or dialect is required"

if isinstance(self.dialect, str):

if self.dialect != dialect.name:

return False

elif isinstance(self.dialect, (tuple, list, set)):

if dialect.name not in self.dialect:

return False

if self.callable_ is not None and not self.callable_(

ddl,

target,

bind,

state=self.state,

dialect=dialect,

compiler=compiler,

**kw,

):

return False

return True

class ExecutableDDLElement(roles.DDLRole, Executable, BaseDDLElement):

"""Base class for standalone executable DDL expression constructs.

This class is the base for the general purpose :class:`.DDL` class,

as well as the various create/drop clause constructs such as

:class:`.CreateTable`, :class:`.DropTable`, :class:`.AddConstraint`,

etc.

.. versionchanged:: 2.0 :class:`.ExecutableDDLElement` is renamed from

:class:`.DDLElement`, which still exists for backwards compatibility.

:class:`.ExecutableDDLElement` integrates closely with SQLAlchemy events,

introduced in :ref:`event_toplevel`. An instance of one is

itself an event receiving callable::

event.listen(

users,

"after_create",

AddConstraint(constraint, isolate_from_table=True).execute_if(

dialect="postgresql"

),

)

.. seealso::

:class:`.DDL`

:class:`.DDLEvents`

:ref:`event_toplevel`

:ref:`schema_ddl_sequences`

"""

_ddl_if: Optional[DDLIf] = None

target: Union[SchemaItem, str, None] = None

def _execute_on_connection(

self, connection, distilled_params, execution_options

):

return connection._execute_ddl(

self, distilled_params, execution_options

)

@_generative

def against(self, target: SchemaItem) -> Self:

"""Return a copy of this :class:`_schema.ExecutableDDLElement` which

will include the given target.

This essentially applies the given item to the ``.target`` attribute of

the returned :class:`_schema.ExecutableDDLElement` object. This target

is then usable by event handlers and compilation routines in order to

provide services such as tokenization of a DDL string in terms of a

particular :class:`_schema.Table`.

When a :class:`_schema.ExecutableDDLElement` object is established as

an event handler for the :meth:`_events.DDLEvents.before_create` or

:meth:`_events.DDLEvents.after_create` events, and the event then

occurs for a given target such as a :class:`_schema.Constraint` or

:class:`_schema.Table`, that target is established with a copy of the

:class:`_schema.ExecutableDDLElement` object using this method, which

then proceeds to the :meth:`_schema.ExecutableDDLElement.execute`

method in order to invoke the actual DDL instruction.

:param target: a :class:`_schema.SchemaItem` that will be the subject

of a DDL operation.

:return: a copy of this :class:`_schema.ExecutableDDLElement` with the

``.target`` attribute assigned to the given

:class:`_schema.SchemaItem`.

.. seealso::

:class:`_schema.DDL` - uses tokenization against the "target" when

processing the DDL string.

"""

self.target = target

return self

@_generative

def execute_if(

self,

dialect: Optional[str] = None,

callable_: Optional[DDLIfCallable] = None,

state: Optional[Any] = None,

) -> Self:

r"""Return a callable that will execute this

:class:`_ddl.ExecutableDDLElement` conditionally within an event

handler.

Used to provide a wrapper for event listening::

event.listen(

metadata,

"before_create",

DDL("my_ddl").execute_if(dialect="postgresql"),

)

:param dialect: May be a string or tuple of strings.

If a string, it will be compared to the name of the

executing database dialect::

DDL("something").execute_if(dialect="postgresql")

If a tuple, specifies multiple dialect names::

DDL("something").execute_if(dialect=("postgresql", "mysql"))

:param callable\_: A callable, which will be invoked with

three positional arguments as well as optional keyword

arguments:

:ddl:

This DDL element.

:target:

The :class:`_schema.Table` or :class:`_schema.MetaData`

object which is the

target of this event. May be None if the DDL is executed

explicitly.

:bind:

The :class:`_engine.Connection` being used for DDL execution.

May be None if this construct is being created inline within

a table, in which case ``compiler`` will be present.

:tables:

Optional keyword argument - a list of Table objects which are to

be created/ dropped within a MetaData.create_all() or drop_all()

method call.

:dialect: keyword argument, but always present - the

:class:`.Dialect` involved in the operation.

:compiler: keyword argument. Will be ``None`` for an engine

level DDL invocation, but will refer to a :class:`.DDLCompiler`

if this DDL element is being created inline within a table.

:state:

Optional keyword argument - will be the ``state`` argument

passed to this function.

:checkfirst:

Keyword argument, will be True if the 'checkfirst' flag was

set during the call to ``create()``, ``create_all()``,

``drop()``, ``drop_all()``.

If the callable returns a True value, the DDL statement will be

executed.

:param state: any value which will be passed to the callable\_

as the ``state`` keyword argument.

.. seealso::

:meth:`.SchemaItem.ddl_if`

:class:`.DDLEvents`

:ref:`event_toplevel`

"""

self._ddl_if = DDLIf(dialect, callable_, state)

return self

def _should_execute(self, target, bind, **kw):

if self._ddl_if is None:

return True

else:

return self._ddl_if._should_execute(self, target, bind, **kw)

def _invoke_with(self, bind):

if self._should_execute(self.target, bind):

return bind.execute(self)

def __call__(self, target, bind, **kw):

"""Execute the DDL as a ddl_listener."""

self.against(target)._invoke_with(bind)

def _generate(self):

s = self.__class__.__new__(self.__class__)

s.__dict__ = self.__dict__.copy()

return s

DDLElement = ExecutableDDLElement

""":class:`.DDLElement` is renamed to :class:`.ExecutableDDLElement`."""

class DDL(ExecutableDDLElement):

"""A literal DDL statement.

Specifies literal SQL DDL to be executed by the database. DDL objects

function as DDL event listeners, and can be subscribed to those events

listed in :class:`.DDLEvents`, using either :class:`_schema.Table` or

:class:`_schema.MetaData` objects as targets.

Basic templating support allows

a single DDL instance to handle repetitive tasks for multiple tables.

Examples::

from sqlalchemy import event, DDL

tbl = Table("users", metadata, Column("uid", Integer))

event.listen(tbl, "before_create", DDL("DROP TRIGGER users_trigger"))

spow = DDL("ALTER TABLE %(table)s SET secretpowers TRUE")

event.listen(tbl, "after_create", spow.execute_if(dialect="somedb"))

drop_spow = DDL("ALTER TABLE users SET secretpowers FALSE")

connection.execute(drop_spow)

When operating on Table events, the following ``statement``

string substitutions are available:

.. sourcecode:: text

%(table)s - the Table name, with any required quoting applied

%(schema)s - the schema name, with any required quoting applied

%(fullname)s - the Table name including schema, quoted if needed

The DDL's "context", if any, will be combined with the standard

substitutions noted above. Keys present in the context will override

the standard substitutions.

"""

__visit_name__ = "ddl"

def __init__(self, statement, context=None):

"""Create a DDL statement.

:param statement:

A string or unicode string to be executed. Statements will be

processed with Python's string formatting operator using

a fixed set of string substitutions, as well as additional

substitutions provided by the optional :paramref:`.DDL.context`

parameter.

A literal '%' in a statement must be escaped as '%%'.

SQL bind parameters are not available in DDL statements.

:param context:

Optional dictionary, defaults to None. These values will be

available for use in string substitutions on the DDL statement.

.. seealso::

:class:`.DDLEvents`

:ref:`event_toplevel`

"""

if not isinstance(statement, str):

raise exc.ArgumentError(

"Expected a string or unicode SQL statement, got '%r'"

% statement

)

self.statement = statement

self.context = context or {}

def __repr__(self):

parts = [repr(self.statement)]

if self.context:

parts.append(f"context={self.context}")

return "<%s@%s; %s>" % (

type(self).__name__,

id(self),

", ".join(parts),

)

class _CreateDropBase(ExecutableDDLElement, Generic[_SI]):

"""Base class for DDL constructs that represent CREATE and DROP or

equivalents.

The common theme of _CreateDropBase is a single

``element`` attribute which refers to the element

to be created or dropped.

"""

element: _SI

def __init__(self, element: _SI) -> None:

self.element = self.target = element

self._ddl_if = getattr(element, "_ddl_if", None)

@property

def stringify_dialect(self): # type: ignore[override]

assert not isinstance(self.element, str)

return self.element.create_drop_stringify_dialect

def _create_rule_disable(self, compiler):

"""Allow disable of _create_rule using a callable.

Pass to _create_rule using

util.portable_instancemethod(self._create_rule_disable)

to retain serializability.

"""

return False

class _CreateBase(_CreateDropBase[_SI]):

def __init__(self, element: _SI, if_not_exists: bool = False) -> None:

super().__init__(element)

self.if_not_exists = if_not_exists

class TableCreateDDL(_CreateBase["Table"]):

def to_metadata(self, metadata: MetaData, table: Table) -> Self:

raise NotImplementedError()

class _DropBase(_CreateDropBase[_SI]):

def __init__(self, element: _SI, if_exists: bool = False) -> None:

super().__init__(element)

self.if_exists = if_exists

class TableDropDDL(_DropBase["Table"]):

def to_metadata(self, metadata: MetaData, table: Table) -> Self:

raise NotImplementedError()

class CreateSchema(_CreateBase[str]):

"""Represent a CREATE SCHEMA statement.

The argument here is the string name of the schema.

"""

__visit_name__ = "create_schema"

stringify_dialect = "default"

def __init__(

self,

name: str,

if_not_exists: bool = False,

) -> None:

"""Create a new :class:`.CreateSchema` construct."""

super().__init__(element=name, if_not_exists=if_not_exists)

class DropSchema(_DropBase[str]):

"""Represent a DROP SCHEMA statement.

The argument here is the string name of the schema.

"""

__visit_name__ = "drop_schema"

stringify_dialect = "default"

def __init__(

self,

name: str,

cascade: bool = False,

if_exists: bool = False,

) -> None:

"""Create a new :class:`.DropSchema` construct."""

super().__init__(element=name, if_exists=if_exists)

self.cascade = cascade

class CreateTable(TableCreateDDL):

"""Represent a CREATE TABLE statement."""

__visit_name__ = "create_table"

def __init__(

self,

element: Table,

include_foreign_key_constraints: Optional[

typing_Sequence[ForeignKeyConstraint]

] = None,

if_not_exists: bool = False,

) -> None:

"""Create a :class:`.CreateTable` construct.

:param element: a :class:`_schema.Table` that's the subject

of the CREATE

:param on: See the description for 'on' in :class:`.DDL`.

:param include_foreign_key_constraints: optional sequence of

:class:`_schema.ForeignKeyConstraint` objects that will be included

inline within the CREATE construct; if omitted, all foreign key

constraints that do not specify use_alter=True are included.

:param if_not_exists: if True, an IF NOT EXISTS operator will be

applied to the construct.

.. versionadded:: 1.4.0b2

"""

super().__init__(element, if_not_exists=if_not_exists)

self.columns = [CreateColumn(column) for column in element.columns]

self.include_foreign_key_constraints = include_foreign_key_constraints

def to_metadata(self, metadata: MetaData, table: Table) -> Self:

return self.__class__(table, if_not_exists=self.if_not_exists)

class _TableViaSelect(TableCreateDDL, ExecutableDDLElement):

"""Common base class for DDL constructs that generate and render for a

:class:`.Table` given a :class:`.Select`

.. versionadded:: 2.1

"""

table: Table

""":class:`.Table` object representing the table that this

:class:`.CreateTableAs` would generate when executed."""

def __init__(

self,

selectable: SelectBase,

name: str,

*,

metadata: Optional["MetaData"] = None,

schema: Optional[str] = None,

temporary: bool = False,

if_not_exists: bool = False,

):

# Coerce selectable to a Select statement

selectable = coercions.expect(roles.DMLSelectRole, selectable)

self.schema = schema

self.selectable = selectable

self.temporary = bool(temporary)

self.if_not_exists = bool(if_not_exists)

self.metadata = metadata

self.table_name = name

self._gen_table()

@property

def element(self): # type: ignore

return self.table

def to_metadata(self, metadata: MetaData, table: Table) -> Self:

new = self.__class__.__new__(self.__class__)

new.__dict__.update(self.__dict__)

new.metadata = metadata

new.table = table

return new

@util.preload_module("sqlalchemy.sql.schema")

def _gen_table(self) -> None:

MetaData = util.preloaded.sql_schema.MetaData

Column = util.preloaded.sql_schema.Column

Table = util.preloaded.sql_schema.Table

MetaData = util.preloaded.sql_schema.MetaData

column_name_type_pairs = (

(name, col_element.type)

for _, name, _, col_element, _ in (

self.selectable._generate_columns_plus_names(

anon_for_dupe_key=False

)

)

)

if self.metadata is None:

self.metadata = metadata = MetaData()

else:

metadata = self.metadata

self.table = Table(

self.table_name,

metadata,

*(Column(name, typ) for name, typ in column_name_type_pairs),

schema=self.schema,

_creator_ddl=self,

)

class CreateTableAs(DialectKWArgs, _TableViaSelect):

"""Represent a CREATE TABLE ... AS statement.

This creates a new table directly from the output of a SELECT, including

its schema and its initial set of data. Unlike a view, the

new table is fixed and does not synchronize further with the originating

SELECT statement.

The example below illustrates basic use of :class:`.CreateTableAs`; given a

:class:`.Select` and optional :class:`.MetaData`, the

:class:`.CreateTableAs` may be invoked directly via

:meth:`.Connection.execute` or indirectly via :meth:`.MetaData.create_all`;

the :attr:`.CreateTableAs.table` attribute provides a :class:`.Table`

object with which to generate new queries::

from sqlalchemy import CreateTableAs

from sqlalchemy import select

# instantiate CreateTableAs given a select() and optional MetaData

cas = CreateTableAs(

select(users.c.id, users.c.name).where(users.c.status == "active"),

"active_users",

metadata=some_metadata,

)

# a Table object is available immediately via the .table attribute

new_statement = select(cas.table)

# to emit CREATE TABLE AS, either invoke CreateTableAs directly...

with engine.begin() as conn:

conn.execute(cas)

# or alternatively, invoke metadata.create_all()

some_metdata.create_all(engine)

# drop is performed in the usual way, via drop_all

# or table.drop()

some_metdata.drop_all(engine)

For detailed background on :class:`.CreateTableAs` see

:ref:`metadata_create_table_as`.

.. versionadded:: 2.1

:param selectable: :class:`_sql.Select`

The SELECT statement providing the columns and rows.

:param table_name: table name as a string. Combine with the optional

:paramref:`.CreateTableAs.schema` parameter to indicate a

schema-qualified table name.

:param metadata: :class:`_schema.MetaData`, optional

If provided, the :class:`_schema.Table` object available via the

:attr:`.table` attribute will be associated with this

:class:`.MetaData`. Otherwise, a new, empty :class:`.MetaData`

is created.

:param schema: str, optional schema or owner name.

:param temporary: bool, default False.

If True, render ``TEMPORARY``

:param if_not_exists: bool, default False.

If True, render ``IF NOT EXISTS``

.. seealso::

:ref:`metadata_create_table_as` - in :ref:`metadata_toplevel`

:meth:`_sql.SelectBase.into` - convenience method to create a

:class:`_schema.CreateTableAs` from a SELECT statement

:class:`.CreateView`

"""

__visit_name__ = "create_table_as"

inherit_cache = False

table: Table

""":class:`.Table` object representing the table that this

:class:`.CreateTableAs` would generate when executed."""

def __init__(

self,

selectable: SelectBase,

table_name: str,

*,

metadata: Optional["MetaData"] = None,

schema: Optional[str] = None,

temporary: bool = False,

if_not_exists: bool = False,

**dialect_kwargs: Any,

):

self._validate_dialect_kwargs(dialect_kwargs)

super().__init__(

selectable=selectable,

name=table_name,

metadata=metadata,

schema=schema,

temporary=temporary,

if_not_exists=if_not_exists,

)

class CreateView(DialectKWArgs, _TableViaSelect):

"""Represent a CREATE VIEW statement.

This creates a new view based on a particular SELECT statement. The schema

of the view is based on the columns of the SELECT statement, and the data

present in the view is derived from the rows represented by the

SELECT. A non-materialized view will evaluate the SELECT statement

dynamically as it is queried, whereas a materialized view represents a

snapshot of the SELECT statement at a particular point in time and

typically needs to be refreshed manually using database-specific commands.

The example below illustrates basic use of :class:`.CreateView`; given a

:class:`.Select` and optional :class:`.MetaData`, the

:class:`.CreateView` may be invoked directly via

:meth:`.Connection.execute` or indirectly via :meth:`.MetaData.create_all`;

the :attr:`.CreateView.table` attribute provides a :class:`.Table`

object with which to generate new queries::

from sqlalchemy import select

from sqlalchemy.sql.ddl import CreateView

# instantiate CreateView given a select() and optional MetaData

create_view = CreateView(

select(users.c.id, users.c.name).where(users.c.status == "active"),

"active_users_view",

metadata=some_metadata,

)

# a Table object is available immediately via the .table attribute

new_statement = select(create_view.table)

# to emit CREATE VIEW, either invoke CreateView directly...

with engine.begin() as conn:

conn.execute(create_view)

# or alternatively, invoke metadata.create_all()

some_metdata.create_all(engine)

# drop is performed in the usual way, via drop_all

# or table.drop() (will emit DROP VIEW)

some_metdata.drop_all(engine)

For detailed background on :class:`.CreateView` see

:ref:`metadata_create_view`.

.. versionadded:: 2.1

:param selectable: :class:`_sql.Select`

The SELECT statement defining the view.

:param view_name: table name as a string. Combine with the optional

:paramref:`.CreateView.schema` parameter to indicate a

schema-qualified table name.

:param metadata: :class:`_schema.MetaData`, optional

If provided, the :class:`_schema.Table` object available via the

:attr:`.table` attribute will be associated with this

:class:`.MetaData`. Otherwise, a new, empty :class:`.MetaData`

is created.

:param schema: str, optional schema or owner name.

:param temporary: bool, default False.

If True, render ``TEMPORARY``

:param or_replace: bool, default False.

If True, render ``OR REPLACE`` to replace an existing view if it

exists. Supported by PostgreSQL, MySQL, MariaDB, and Oracle.

Not supported by SQLite or SQL Server.

.. versionadded:: 2.1

:param materialized: bool, default False.

If True, render ``MATERIALIZED`` to create a materialized view.

Materialized views store the query results physically and can be

refreshed periodically. Not supported by all database backends.

.. versionadded:: 2.1

:param dialect_kw: Additional keyword arguments are dialect-specific and

are passed as keyword arguments to the dialect's compiler.

.. note::

For SQLite, the ``sqlite_if_not_exists`` boolean parameter

is supported to render ``CREATE VIEW IF NOT EXISTS``.

.. versionadded:: 2.1

.. seealso::

:ref:`metadata_create_view` - in :ref:`metadata_toplevel`

:class:`.CreateTableAs` - for creating a table from a SELECT statement

"""

__visit_name__ = "create_view"

inherit_cache = False

table: Table

""":class:`.Table` object representing the view that this

:class:`.CreateView` would generate when executed."""

materialized: bool

"""Boolean flag indicating if this is a materialized view."""

or_replace: bool

"""Boolean flag indicating if OR REPLACE should be used."""

def __init__(

self,

selectable: SelectBase,

view_name: str,

*,

metadata: Optional["MetaData"] = None,

schema: Optional[str] = None,

temporary: bool = False,

or_replace: bool = False,

materialized: bool = False,

**dialect_kwargs: Any,

):

self._validate_dialect_kwargs(dialect_kwargs)

super().__init__(

selectable=selectable,

name=view_name,

metadata=metadata,

schema=schema,

temporary=temporary,

if_not_exists=False,

)

self.materialized = materialized

self.or_replace = or_replace

self.table._dropper_ddl = DropView(

self.table, materialized=materialized

)

class DropView(TableDropDDL):

"""'DROP VIEW' construct.

.. versionadded:: 2.1 the :class:`.DropView` construct became public

and was renamed from ``_DropView``.

"""

__visit_name__ = "drop_view"

materialized: bool

"""Boolean flag indicating if this is a materialized view."""

def __init__(

self,

element: Table,

*,

if_exists: bool = False,

materialized: bool = False,

) -> None:

super().__init__(element, if_exists=if_exists)

self.materialized = materialized

def to_metadata(self, metadata: MetaData, table: Table) -> Self:

new = self.__class__.__new__(self.__class__)

new.__dict__.update(self.__dict__)

new.element = table

return new

class CreateConstraint(BaseDDLElement):

element: Constraint

def __init__(self, element: Constraint) -> None:

self.element = element

class CreateColumn(BaseDDLElement):

"""Represent a :class:`_schema.Column`

as rendered in a CREATE TABLE statement,

via the :class:`.CreateTable` construct.

This is provided to support custom column DDL within the generation

of CREATE TABLE statements, by using the

compiler extension documented in :ref:`sqlalchemy.ext.compiler_toplevel`

to extend :class:`.CreateColumn`.

Typical integration is to examine the incoming :class:`_schema.Column`

object, and to redirect compilation if a particular flag or condition

is found::

from sqlalchemy import schema

from sqlalchemy.ext.compiler import compiles

@compiles(schema.CreateColumn)

def compile(element, compiler, **kw):

column = element.element

if "special" not in column.info:

return compiler.visit_create_column(element, **kw)

text = "%s SPECIAL DIRECTIVE %s" % (

column.name,

compiler.type_compiler.process(column.type),

)

default = compiler.get_column_default_string(column)

if default is not None:

text += " DEFAULT " + default

if not column.nullable:

text += " NOT NULL"

if column.constraints:

text += " ".join(

compiler.process(const) for const in column.constraints

)

return text

The above construct can be applied to a :class:`_schema.Table`

as follows::

from sqlalchemy import Table, Metadata, Column, Integer, String

from sqlalchemy import schema

metadata = MetaData()

table = Table(

"mytable",

MetaData(),

Column("x", Integer, info={"special": True}, primary_key=True),

Column("y", String(50)),

Column("z", String(20), info={"special": True}),

)

metadata.create_all(conn)

Above, the directives we've added to the :attr:`_schema.Column.info`

collection

will be detected by our custom compilation scheme:

.. sourcecode:: sql