In the preceding sections we’ve discussed a variety of schema constructs
including Table
,
ForeignKeyConstraint
,
CheckConstraint
, and
Sequence
. Throughout, we’ve relied upon the
create()
and create_all()
methods of
Table
and MetaData
in
order to issue data definition language (DDL) for all constructs. When issued,
a pre-determined order of operations is invoked, and DDL to create each table
is created unconditionally including all constraints and other objects
associated with it. For more complex scenarios where database-specific DDL is
required, SQLAlchemy offers two techniques which can be used to add any DDL
based on any condition, either accompanying the standard generation of tables
or by itself.
The sqlalchemy.schema
package contains SQL expression constructs that
provide DDL expressions. For example, to produce a CREATE TABLE
statement:
from sqlalchemy.schema import CreateTable
sqlengine.execute(CreateTable(mytable))
CREATE TABLE mytable (
col1 INTEGER,
col2 INTEGER,
col3 INTEGER,
col4 INTEGER,
col5 INTEGER,
col6 INTEGER
)
Above, the CreateTable
construct works like any
other expression construct (such as select()
, table.insert()
, etc.). A
full reference of available constructs is in DDL Expression Constructs API.
The DDL constructs all extend a common base class which provides the
capability to be associated with an individual
Table
or MetaData
object, to be invoked upon create/drop events. Consider the example of a table
which contains a CHECK constraint:
users = Table('users', metadata,
Column('user_id', Integer, primary_key=True),
Column('user_name', String(40), nullable=False),
CheckConstraint('length(user_name) >= 8',name="cst_user_name_length")
)
sqlusers.create(engine)
CREATE TABLE users (
user_id SERIAL NOT NULL,
user_name VARCHAR(40) NOT NULL,
PRIMARY KEY (user_id),
CONSTRAINT cst_user_name_length CHECK (length(user_name) >= 8)
)
The above table contains a column “user_name” which is subject to a CHECK
constraint that validates that the length of the string is at least eight
characters. When a create()
is issued for this table, DDL for the
CheckConstraint
will also be issued inline within
the table definition.
The CheckConstraint
construct can also be
constructed externally and associated with the
Table
afterwards:
constraint = CheckConstraint('length(user_name) >= 8',name="cst_user_name_length")
users.append_constraint(constraint)
So far, the effect is the same. However, if we create DDL elements
corresponding to the creation and removal of this constraint, and associate
them with the Table
as events, these new events
will take over the job of issuing DDL for the constraint. Additionally, the
constraint will be added via ALTER:
from sqlalchemy import event
event.listen(
users,
"after_create",
AddConstraint(constraint)
)
event.listen(
users,
"before_drop",
DropConstraint(constraint)
)
sqlusers.create(engine)
CREATE TABLE users (
user_id SERIAL NOT NULL,
user_name VARCHAR(40) NOT NULL,
PRIMARY KEY (user_id)
)
ALTER TABLE users ADD CONSTRAINT cst_user_name_length CHECK (length(user_name) >= 8)
sqlusers.drop(engine)
ALTER TABLE users DROP CONSTRAINT cst_user_name_length
DROP TABLE users
The real usefulness of the above becomes clearer once we illustrate the
DDLElement.execute_if()
method. This method returns a modified form of
the DDL callable which will filter on criteria before responding to a
received event. It accepts a parameter dialect
, which is the string
name of a dialect or a tuple of such, which will limit the execution of the
item to just those dialects. It also accepts a callable_
parameter which
may reference a Python callable which will be invoked upon event reception,
returning True
or False
indicating if the event should proceed.
If our CheckConstraint
was only supported by
Postgresql and not other databases, we could limit its usage to just that dialect:
event.listen(
users,
'after_create',
AddConstraint(constraint).execute_if(dialect='postgresql')
)
event.listen(
users,
'before_drop',
DropConstraint(constraint).execute_if(dialect='postgresql')
)
Or to any set of dialects:
event.listen(
users,
"after_create",
AddConstraint(constraint).execute_if(dialect=('postgresql', 'mysql'))
)
event.listen(
users,
"before_drop",
DropConstraint(constraint).execute_if(dialect=('postgresql', 'mysql'))
)
When using a callable, the callable is passed the ddl element, the
Table
or MetaData
object whose “create” or “drop” event is in progress, and the
Connection
object being used for the
operation, as well as additional information as keyword arguments. The
callable can perform checks, such as whether or not a given item already
exists. Below we define should_create()
and should_drop()
callables
that check for the presence of our named constraint:
def should_create(ddl, target, connection, **kw):
row = connection.execute("select conname from pg_constraint where conname='%s'" % ddl.element.name).scalar()
return not bool(row)
def should_drop(ddl, target, connection, **kw):
return not should_create(ddl, target, connection, **kw)
event.listen(
users,
"after_create",
AddConstraint(constraint).execute_if(callable_=should_create)
)
event.listen(
users,
"before_drop",
DropConstraint(constraint).execute_if(callable_=should_drop)
)
sqlusers.create(engine)
CREATE TABLE users (
user_id SERIAL NOT NULL,
user_name VARCHAR(40) NOT NULL,
PRIMARY KEY (user_id)
)
select conname from pg_constraint where conname='cst_user_name_length'
ALTER TABLE users ADD CONSTRAINT cst_user_name_length CHECK (length(user_name) >= 8)
sqlusers.drop(engine)
select conname from pg_constraint where conname='cst_user_name_length'
ALTER TABLE users DROP CONSTRAINT cst_user_name_length
DROP TABLE users
Custom DDL phrases are most easily achieved using the
DDL
construct. This construct works like all the
other DDL elements except it accepts a string which is the text to be emitted:
event.listen(
metadata,
"after_create",
DDL("ALTER TABLE users ADD CONSTRAINT "
"cst_user_name_length "
" CHECK (length(user_name) >= 8)")
)
A more comprehensive method of creating libraries of DDL constructs is to use custom compilation - see Custom SQL Constructs and Compilation Extension for details.
sqlalchemy.schema.
sort_tables
(tables, skip_fn=None, extra_dependencies=None)¶sort a collection of Table
objects based on dependency.
This is a dependency-ordered sort which will emit Table
objects such that they will follow their dependent Table
objects.
Tables are dependent on another based on the presence of
ForeignKeyConstraint
objects as well as explicit dependencies
added by Table.add_is_dependent_on()
.
Warning
The sort_tables()
function cannot by itself accommodate
automatic resolution of dependency cycles between tables, which
are usually caused by mutually dependent foreign key constraints.
To resolve these cycles, either the
ForeignKeyConstraint.use_alter
parameter may be appled
to those constraints, or use the
sql.sort_tables_and_constraints()
function which will break
out foreign key constraints involved in cycles separately.
Parameters: |
|
---|
sqlalchemy.schema.
sort_tables_and_constraints
(tables, filter_fn=None, extra_dependencies=None)¶sort a collection of Table
/ ForeignKeyConstraint
objects.
This is a dependency-ordered sort which will emit tuples of
(Table, [ForeignKeyConstraint, ...])
such that each
Table
follows its dependent Table
objects.
Remaining ForeignKeyConstraint
objects that are separate due to
dependency rules not satisifed by the sort are emitted afterwards
as (None, [ForeignKeyConstraint ...])
.
Tables are dependent on another based on the presence of
ForeignKeyConstraint
objects, explicit dependencies
added by Table.add_is_dependent_on()
, as well as dependencies
stated here using the skip_fn
and/or extra_dependencies
parameters.
Parameters: |
|
---|
New in version 1.0.0.
See also
sqlalchemy.schema.
DDLElement
¶Bases: sqlalchemy.sql.expression.Executable
, sqlalchemy.schema._DDLCompiles
Base class for DDL expression constructs.
This class is the base for the general purpose DDL
class,
as well as the various create/drop clause constructs such as
CreateTable
, DropTable
, AddConstraint
,
etc.
DDLElement
integrates closely with SQLAlchemy events,
introduced in Events. An instance of one is
itself an event receiving callable:
event.listen(
users,
'after_create',
AddConstraint(constraint).execute_if(dialect='postgresql')
)
__call__
(target, bind, **kw)¶Execute the DDL as a ddl_listener.
against
(target)¶Return a copy of this DDL against a specific schema item.
bind
¶callable_
= None¶dialect
= None¶execute
(bind=None, target=None)¶Execute this DDL immediately.
Executes the DDL statement in isolation using the supplied
Connectable
or
Connectable
assigned to the .bind
property, if not supplied. If the DDL has a conditional on
criteria, it will be invoked with None as the event.
Parameters: |
|
---|
execute_at
(event_name, target)¶Link execution of this DDL to the DDL lifecycle of a SchemaItem.
Deprecated since version 0.7: See DDLEvents
, as well as DDLElement.execute_if()
.
Links this DDLElement
to a Table
or MetaData
instance,
executing it when that schema item is created or dropped. The DDL
statement will be executed using the same Connection and transactional
context as the Table create/drop itself. The .bind
property of
this statement is ignored.
Parameters: |
---|
A DDLElement instance can be linked to any number of schema items.
execute_at
builds on the append_ddl_listener
interface of
MetaData
and Table
objects.
Caveat: Creating or dropping a Table in isolation will also trigger
any DDL set to execute_at
that Table’s MetaData. This may change
in a future release.
execute_if
(dialect=None, callable_=None, state=None)¶Return a callable that will execute this DDLElement conditionally.
Used to provide a wrapper for event listening:
event.listen(
metadata,
'before_create',
DDL("my_ddl").execute_if(dialect='postgresql')
)
Parameters: |
|
---|
on
= None¶target
= None¶sqlalchemy.schema.
DDL
(statement, on=None, context=None, bind=None)¶Bases: sqlalchemy.schema.DDLElement
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 DDLEvents
, using either Table
or
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 substitions are available:
%(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.
__init__
(statement, on=None, context=None, bind=None)¶Create a DDL statement.
Parameters: |
|
---|
sqlalchemy.schema.
CreateTable
(element, on=None, bind=None, include_foreign_key_constraints=None)¶Bases: sqlalchemy.schema._CreateDropBase
Represent a CREATE TABLE statement.
__init__
(element, on=None, bind=None, include_foreign_key_constraints=None)¶Create a CreateTable
construct.
Parameters: |
|
---|
sqlalchemy.schema.
DropTable
(element, on=None, bind=None)¶Bases: sqlalchemy.schema._CreateDropBase
Represent a DROP TABLE statement.
sqlalchemy.schema.
CreateColumn
(element)¶Bases: sqlalchemy.schema._DDLCompiles
Represent a Column
as rendered in a CREATE TABLE statement,
via the CreateTable
construct.
This is provided to support custom column DDL within the generation
of CREATE TABLE statements, by using the
compiler extension documented in Custom SQL Constructs and Compilation Extension
to extend CreateColumn
.
Typical integration is to examine the incoming 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 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 Column.info
collection
will be detected by our custom compilation scheme:
CREATE TABLE mytable (
x SPECIAL DIRECTIVE INTEGER NOT NULL,
y VARCHAR(50),
z SPECIAL DIRECTIVE VARCHAR(20),
PRIMARY KEY (x)
)
The CreateColumn
construct can also be used to skip certain
columns when producing a CREATE TABLE
. This is accomplished by
creating a compilation rule that conditionally returns None
.
This is essentially how to produce the same effect as using the
system=True
argument on Column
, which marks a column
as an implicitly-present “system” column.
For example, suppose we wish to produce a Table
which skips
rendering of the Postgresql xmin
column against the Postgresql
backend, but on other backends does render it, in anticipation of a
triggered rule. A conditional compilation rule could skip this name only
on Postgresql:
from sqlalchemy.schema import CreateColumn
@compiles(CreateColumn, "postgresql")
def skip_xmin(element, compiler, **kw):
if element.element.name == 'xmin':
return None
else:
return compiler.visit_create_column(element, **kw)
my_table = Table('mytable', metadata,
Column('id', Integer, primary_key=True),
Column('xmin', Integer)
)
Above, a CreateTable
construct will generate a CREATE TABLE
which only includes the id
column in the string; the xmin
column
will be omitted, but only against the Postgresql backend.
New in version 0.8.3: The CreateColumn
construct supports
skipping of columns by returning None
from a custom compilation
rule.
New in version 0.8: The CreateColumn
construct was added
to support custom column creation styles.
sqlalchemy.schema.
CreateSequence
(element, on=None, bind=None)¶Bases: sqlalchemy.schema._CreateDropBase
Represent a CREATE SEQUENCE statement.
sqlalchemy.schema.
DropSequence
(element, on=None, bind=None)¶Bases: sqlalchemy.schema._CreateDropBase
Represent a DROP SEQUENCE statement.
sqlalchemy.schema.
CreateIndex
(element, on=None, bind=None)¶Bases: sqlalchemy.schema._CreateDropBase
Represent a CREATE INDEX statement.
sqlalchemy.schema.
DropIndex
(element, on=None, bind=None)¶Bases: sqlalchemy.schema._CreateDropBase
Represent a DROP INDEX statement.
sqlalchemy.schema.
AddConstraint
(element, *args, **kw)¶Bases: sqlalchemy.schema._CreateDropBase
Represent an ALTER TABLE ADD CONSTRAINT statement.
sqlalchemy.schema.
DropConstraint
(element, cascade=False, **kw)¶Bases: sqlalchemy.schema._CreateDropBase
Represent an ALTER TABLE DROP CONSTRAINT statement.
sqlalchemy.schema.
CreateSchema
(name, quote=None, **kw)¶Bases: sqlalchemy.schema._CreateDropBase
Represent a CREATE SCHEMA statement.
New in version 0.7.4.
The argument here is the string name of the schema.
__init__
(name, quote=None, **kw)¶Create a new CreateSchema
construct.
sqlalchemy.schema.
DropSchema
(name, quote=None, cascade=False, **kw)¶Bases: sqlalchemy.schema._CreateDropBase
Represent a DROP SCHEMA statement.
The argument here is the string name of the schema.
New in version 0.7.4.
__init__
(name, quote=None, cascade=False, **kw)¶Create a new DropSchema
construct.