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

我们从Python开源项目中，提取了以下14个代码示例，用于说明如何使用sqlalchemy.Binary()。

项目：idealoom 作者：conversence | 项目源码 | 文件源码

def upgrade(pyramid_env):
    with context.begin_transaction():
        op.add_column('imported_post', sa.Column('imported_blob', sa.Binary))
        op.execute(
            """UPDATE imported_post SET imported_blob=(
                    SELECT email.full_message FROM email
                    WHERE email.id = imported_post.id
                    )
            """)
        op.drop_column('email', 'full_message')

    # Do stuff with the app's models here.
    from assembl import models as m
    db = m.get_session_maker()()
    with transaction.manager:
        pass

项目：idealoom 作者：conversence | 项目源码 | 文件源码

def upgrade(pyramid_env):
    with context.begin_transaction():
        op.drop_column('facebook_post', 'post_type')
        op.drop_column('facebook_post', 'link_name')
        # Could also alter column, but this operation has limitations
        # (which we won't hit)
        op.drop_column('facebook_post', 'attachment')
        op.add_column('facebook_post', sa.Column('attachment_blob', sa.Binary))
        op.add_column('facebook_source', sa.Column('lower_bound', sa.DateTime))
        op.add_column('facebook_source', sa.Column('upper_bound', sa.DateTime))

        # Do stuff with the app's models here.
        from assembl import models as m
        db = m.get_session_maker()()
        with transaction.manager:
            # Also correct the spelling mistake in attachment model.
            a = m.Attachment.__table__
            db.execute(
                a.update().where(a.c.attachmentPurpose == op.inline_literal(
                                 'EMBEEDED_ATTACHMENT')).
                values(attachmentPurpose=op.inline_literal("EMBED_ATTACHMENT"))
            )

项目：hostapd-mana 作者：adde88 | 项目源码 | 文件源码

def passwords_getitem(self, key):
        newkey, buf = PAW2_Buffer(self.storage.passwords[key]).pack()
        return xmlrpclib.Binary(buf)

项目：hostapd-mana 作者：adde88 | 项目源码 | 文件源码

def essids_getitem(self, essid, key):
        try:
            results = self.storage.essids[essid, key]
        except KeyError:
            return False
        else:
            buf = PYR2_Buffer(essid, results).pack()
            return xmlrpclib.Binary(buf)

项目：seedbox 作者：nailgun | 项目源码 | 文件源码

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('cluster', sa.Column('k8s_service_account_private_key', sa.Binary(), nullable=True))
    op.add_column('cluster', sa.Column('k8s_service_account_public_key', sa.Binary(), nullable=True))
    op.execute('UPDATE cluster SET k8s_service_account_private_key=(SELECT key FROM credentials_data WHERE credentials_data.id = cluster.service_account_keypair_id)')
    op.execute('UPDATE cluster SET k8s_service_account_public_key=(SELECT cert FROM credentials_data WHERE credentials_data.id = cluster.service_account_keypair_id)')
    op.alter_column('cluster', 'k8s_service_account_private_key', nullable=False)
    op.alter_column('cluster', 'k8s_service_account_public_key', nullable=False)
    op.drop_constraint('cluster_service_account_keypair_id_fkey', 'cluster', type_='foreignkey')
    op.drop_column('cluster', 'service_account_keypair_id')
    # ### end Alembic commands ###

项目：seedbox 作者：nailgun | 项目源码 | 文件源码

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('provision',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('node_id', sa.Integer(), nullable=False),
    sa.Column('applied_at', sa.DateTime(), nullable=True),
    sa.Column('config_version', sa.Integer(), nullable=False),
    sa.Column('ignition_config', sa.Binary(), nullable=True),
    sa.Column('ipxe_config', sa.Text(), nullable=True),
    sa.ForeignKeyConstraint(['node_id'], ['node.id'], ),
    sa.PrimaryKeyConstraint('id')
    )
    op.drop_column('node', 'active_ignition_config')
    # ### end Alembic commands ###

项目：sslstrip-hsts-openwrt 作者：adde88 | 项目源码 | 文件源码

def passwords_getitem(self, key):
        newkey, buf = PAW2_Buffer(self.storage.passwords[key]).pack()
        return xmlrpclib.Binary(buf)

项目：sslstrip-hsts-openwrt 作者：adde88 | 项目源码 | 文件源码

def essids_getitem(self, essid, key):
        try:
            results = self.storage.essids[essid, key]
        except KeyError:
            return False
        else:
            buf = PYR2_Buffer(essid, results).pack()
            return xmlrpclib.Binary(buf)

项目：gthnk 作者：iandennismiller | 项目源码 | 文件源码

def upgrade():
    ### commands auto generated by Alembic - please adjust! ###
    op.create_table('page',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('day_id', sa.Integer(), nullable=True),
    sa.Column('sequence', sa.Integer(), nullable=True),
    sa.Column('binary', sa.Binary(), nullable=True),
    sa.Column('title', sa.Unicode(length=1024), nullable=True),
    sa.ForeignKeyConstraint(['day_id'], ['day.id'], ),
    sa.PrimaryKeyConstraint('id')
    )
    ### end Alembic commands ###

项目：gthnk 作者：iandennismiller | 项目源码 | 文件源码

def upgrade():
    ### commands auto generated by Alembic - please adjust! ###
    op.add_column('page', sa.Column('preview', sa.Binary(), nullable=True))
    op.add_column('page', sa.Column('thumbnail', sa.Binary(), nullable=True))
    ### end Alembic commands ###

项目：nyaa 作者：nyaadevs | 项目源码 | 文件源码

def upgrade():

    for prefix in TABLE_PREFIXES:
        op.add_column(prefix + '_torrents', sa.Column('uploader_ip', sa.Binary(), nullable=True))
        # ### end Alembic commands ###

项目：nyaa 作者：nyaadevs | 项目源码 | 文件源码

def upgrade():
    op.create_table('bans',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('created_time', sa.DateTime(), nullable=True),
    sa.Column('admin_id', sa.Integer(), nullable=False),
    sa.Column('user_id', sa.Integer(), nullable=True),
    sa.Column('user_ip', sa.Binary(length=16), nullable=True),
    sa.Column('reason', sa.String(length=2048), nullable=False),
    sa.ForeignKeyConstraint(['admin_id'], ['users.id'], ),
    sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
    sa.PrimaryKeyConstraint('id'),
    )
    op.create_index('user_ip_16', 'bans', ['user_ip'], unique=True, mysql_length=16)
    op.create_index('user_ip_4', 'bans', ['user_ip'], unique=True, mysql_length=4)

项目：idealoom 作者：conversence | 项目源码 | 文件源码

def downgrade(pyramid_env):
    with context.begin_transaction():
        op.add_column('email', sa.Column('full_message', sa.Binary))
        op.execute(
            """UPDATE email SET full_message=(
                SELECT imported_post.imported_blob from imported_post
                WHERE imported_post.id = email.id
            )"""
        )
        op.drop_column('imported_post', 'imported_blob')

项目：league 作者：massgo | 项目源码 | 文件源码

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('games',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('winner', sa.Enum('white', 'black', name='color'), nullable=True),
    sa.Column('handicap', sa.SmallInteger(), nullable=True),
    sa.Column('komi', sa.SmallInteger(), nullable=True),
    sa.Column('season', sa.Integer(), nullable=True),
    sa.Column('episode', sa.Integer(), nullable=True),
    sa.PrimaryKeyConstraint('id')
    )
    op.create_table('players',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('first_name', sa.String(length=30), nullable=True),
    sa.Column('last_name', sa.String(length=30), nullable=True),
    sa.Column('aga_id', sa.Integer(), nullable=True),
    sa.Column('aga_rank', sa.Integer(), nullable=True),
    sa.PrimaryKeyConstraint('id')
    )
    op.create_index(op.f('ix_players_aga_id'), 'players', ['aga_id'], unique=True)
    op.create_table('users',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('username', sa.String(length=80), nullable=False),
    sa.Column('email', sa.String(length=80), nullable=False),
    sa.Column('password', sa.Binary(), nullable=True),
    sa.Column('created_at', sa.DateTime(), nullable=False),
    sa.Column('first_name', sa.String(length=30), nullable=True),
    sa.Column('last_name', sa.String(length=30), nullable=True),
    sa.Column('active', sa.Boolean(), nullable=True),
    sa.Column('is_admin', sa.Boolean(), nullable=True),
    sa.PrimaryKeyConstraint('id'),
    sa.UniqueConstraint('email'),
    sa.UniqueConstraint('username')
    )
    op.create_table('black_player_games',
    sa.Column('player_id', sa.Integer(), nullable=False),
    sa.Column('game_id', sa.Integer(), nullable=False),
    sa.ForeignKeyConstraint(['game_id'], ['games.id'], ),
    sa.ForeignKeyConstraint(['player_id'], ['players.id'], ),
    sa.PrimaryKeyConstraint('player_id', 'game_id')
    )
    op.create_table('roles',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('name', sa.String(length=80), nullable=False),
    sa.Column('user_id', sa.Integer(), nullable=True),
    sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
    sa.PrimaryKeyConstraint('id'),
    sa.UniqueConstraint('name')
    )
    op.create_table('white_player_games',
    sa.Column('player_id', sa.Integer(), nullable=False),
    sa.Column('game_id', sa.Integer(), nullable=False),
    sa.ForeignKeyConstraint(['game_id'], ['games.id'], ),
    sa.ForeignKeyConstraint(['player_id'], ['players.id'], ),
    sa.PrimaryKeyConstraint('player_id', 'game_id')
    )
    # ### end Alembic commands ###