Python sqlalchemy 模块,DDL 实例源码
我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用sqlalchemy.DDL。
def __init__(
self, element, on=None, bind=None,
include_foreign_key_constraints=None):
"""Create a :class:`.CreateTable` construct.
:param element: a :class:`.Table` that's the subject
of the CREATE
:param on: See the description for 'on' in :class:`.DDL`.
:param bind: See the description for 'bind' in :class:`.DDL`.
:param include_foreign_key_constraints: optional sequence of
:class:`.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.
.. versionadded:: 1.0.0
"""
super(CreateTable, self).__init__(element, on=on, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def append_constraint(self, constraint):
"""Append a :class:`~.schema.Constraint` to this
:class:`~.schema.Table`.
This has the effect of the constraint being included in any
future CREATE TABLE statement,assuming specific DDL creation
events have not been associated with the given
:class:`~.schema.Constraint` object.
Note that this does **not** produce the constraint within the
relational database automatically,for a table that already exists
in the database. To add a constraint to an
existing relational database table,the sql ALTER command must
be used. sqlAlchemy also provides the
:class:`.AddConstraint` construct which can produce this sql when
invoked as an executable clause.
"""
constraint._set_parent_with_dispatch(self)
def __init__(self, *columns, **kw):
"""
:param \*columns:
A sequence of column names or Column objects.
:param name:
Optional,the in-database name of this constraint.
:param deferrable:
Optional bool. If set,emit DEFERRABLE or NOT DEFERRABLE when
issuing DDL for this constraint.
:param initially:
Optional string. If set,emit INITIALLY <value> when issuing DDL
for this constraint.
"""
ColumnCollectionMixin.__init__(self, *columns)
Constraint.__init__(self, **kw)
def _bind_or_error(schemaitem, msg=None):
bind = schemaitem.bind
if not bind:
name = schemaitem.__class__.__name__
label = getattr(schemaitem, 'fullname',
getattr(schemaitem, 'name', None))
if label:
item = '%s %r' % (name, label)
else:
item = name
if isinstance(schemaitem, (MetaData, DDL)):
bindable = "the %s's .bind" % name
else:
bindable = "this %s's .Metadata.bind" % name
if msg is None:
msg = "The %s is not bound to an Engine or Connection. "\
"Execution can not proceed without a database to execute "\
"against. Either execute with an explicit connection or "\
"assign %s to enable implicit execution." % \
(item, bindable)
raise exc.UnboundExecutionError(msg)
return bind
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def upgrade():
conn = op.get_bind()
func = sa.DDL("""CREATE FUNCTION set_Meta_updated()
RETURNS TRIGGER
LANGUAGE plpgsql
AS $$
BEGIN
NEW.Meta_updated := Now();
RETURN NEW;
END;
$$;""")
conn.execute(func)
for table in updatable_tables:
trigger_params = {'trigger': ('%s_set_Meta_updated' % table), 'table': table}
trigger = ("""CREATE TRIGGER %(trigger)s
BEFORE UPDATE ON %(table)s
FOR EACH ROW EXECUTE PROCEDURE set_Meta_updated();""" % trigger_params)
conn.execute(trigger)
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def main():
database = 'postgres'
user = 'postgres'
url = sa.engine.url.URL(
'postgresql', host=os.environ['PGHOST'], database=database,
username=user, password=os.environ.get('PGPASSWORD', None)
)
ddl_text = sa.DDL('CREATE DATABASE {};'.format(os.environ['PGDATABASE']))
engine = sa.create_engine(url)
engine.raw_connection().set_isolation_level(
ISOLATION_LEVEL_AUTOCOMMIT
)
try:
engine.execute(ddl_text)
sys.stdout.write('Creating environment successfully.\n')
except psycopg2.Error:
raise SystemExit('Could not connect to Postgresql.\n{0}'.format(sys.exc_info()))
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __init__(
self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
self.include_foreign_key_constraints = include_foreign_key_constraints
def execute(self, target=None):
"""Execute this DDL immediately.
Executes the DDL statement in isolation using the supplied
:class:`.Connectable` or
:class:`.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.
:param bind:
Optional,an ``Engine`` or ``Connection``. If not supplied,a valid
:class:`.Connectable` must be present in the
``.bind`` property.
:param target:
Optional,defaults to None. The target SchemaItem for the
execute call. Will be passed to the ``on`` callable if any,
and may also provide string expansion data for the
statement. See ``execute_at`` for more information.
"""
if bind is None:
bind = _bind_or_error(self)
if self._should_execute(target, bind):
return bind.execute(self.against(target))
else:
bind.engine.logger.info(
"DDL execution skipped,criteria not met.")
def execute_at(self, event_name, target):
"""Link execution of this DDL to the DDL lifecycle of a SchemaItem.
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.
:param event:
One of the events defined in the schema item's ``.ddl_events``;
e.g. 'before-create','after-create','before-drop' or 'after-drop'
:param target:
The Table or MetaData instance for which this DDLElement will
be associated with.
A DDLElement instance can be linked to any number of schema items.
``execute_at`` builds on the ``append_ddl_listener`` interface of
:class:`.MetaData` and :class:`.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.
"""
def call_event(target, connection, **kw):
if self._should_execute_deprecated(event_name,
target, **kw):
return connection.execute(self.against(target))
event.listen(target, "" + event_name.replace('-', '_'), call_event)
def against(self, target):
"""Return a copy of this DDL against a specific schema item."""
self.target = target
def __call__(self, target, bind, **kw):
"""Execute the DDL as a ddl_listener."""
if self._should_execute(target, **kw):
return bind.execute(self.against(target))
def against(self, target):
"""Return a copy of this DDL against a specific schema item."""
self.target = target
def __call__(self, **kw):
return bind.execute(self.against(target))
def against(self, target):
"""Return a copy of this DDL against a specific schema item."""
self.target = target
def __call__(self, **kw):
return bind.execute(self.against(target))
def __init__(self, bind=None):
"""Create a :class:`.CreateTable` construct.
:param element: a :class:`.Table` that's the subject
of the CREATE
:param on: See the description for 'on' in :class:`.DDL`.
:param bind: See the description for 'bind' in :class:`.DDL`.
"""
super(CreateTable, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
def against(self, target):
"""Return a copy of this DDL against a specific schema item."""
self.target = target
def __call__(self, **kw):
return bind.execute(self.against(target))
def __init__(self, bind=bind)
self.columns = [CreateColumn(column)
for column in element.columns
]
def against(self, target):
"""Return a copy of this DDL against a specific schema item."""
self.target = target