我们从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 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 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 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 execute(self, bind=None, 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, connection, **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, bind, **kw): return bind.execute(self.against(target))
def __init__(self, element, on=None, 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, self).__init__(element, on=on, bind=bind) self.columns = [CreateColumn(column) for column in element.columns ]
