Python sqlalchemy 模块，sql() 实例源码

def constrained(self):
        from zeus.auth import get_current_tenant

        if not self.current_constrained:
            return self

        mzero = self._mapper_zero()
        if mzero is not None:
            tenant = get_current_tenant()
            if tenant.repository_ids:
                return self.enable_assertions(False).filter(
                    mzero.class_.repository_id.in_(tenant.repository_ids)
                )
            else:
                return self.enable_assertions(False).filter(sqlalchemy.sql.false())
        return self

def run(self, connection):
        print "Upgrading to schema version {0}".format(self.schema_version)
        with connection.begin():
            if self.schema_version != 1:
                old_version = connection.execute("""SELECT value FROM config
                  WHERE key = 'schema_version'""").scalar()
                if int(old_version) + 1 != self.schema_version:
                    raise DatabaseError("""Tried to update schema from
                      {0} to {1}""".format(old_version, self.schema_version))
            if self.statements:
                for s in self.statements:
                    connection.execute(s)
            if self.callable:
                self.callable(connection)
            connection.execute(
                sqlalchemy.sql.text(
                    """UPDATE config SET value = :version
                    WHERE key = 'schema_version'"""),
                version=self.schema_version)

def get_column_default_string(self, column):
        if (isinstance(column.server_default, schema.DefaultClause) and
            isinstance(column.server_default.arg, basestring)):
                if isinstance(column.type, (sqltypes.Integer, sqltypes.Numeric)):
                    return self.sql_compiler.process(text(column.server_default.arg))

        return super(InfoDDLCompiler, self).get_column_default_string(column)

    ### Informix wants the constraint name at the end, hence this ist c&p from sql/compiler.py

def execute_statement(table):
    cmd = (sa.sql
        .select([attributes])
        .limit(5)
    )

    cmd_string = 'select([{}]).limit(5)'.format(table.name)
    cd.display.code_block(cmd_string, language_id='py')

    results = engine.execute(cmd)
    for r in results:
        print(r)

def _upsert_run(self, session, info):
        sql = pg.insert(db.Run).values(
            id=info['id'], info=info,
            hostname=info['host']['nodename'],
            created_at=info['created_at']
        ).on_conflict_do_nothing(index_elements=['id'])

        session.execute(sql)

        return db.Run(info)