Python sqlalchemy 模块，Binary() 实例源码
我们从Python开源项目中，提取了以下14个代码示例，用于说明如何使用sqlalchemy.Binary()。
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
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"))
)
def passwords_getitem(self, key):
newkey, buf = PAW2_Buffer(self.storage.passwords[key]).pack()
return xmlrpclib.Binary(buf)
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)
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', 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 ###
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.Column('applied_at', sa.DateTime(), nullable=True),
sa.Column('config_version',
sa.Column('ignition_config',
sa.Column('ipxe_config', sa.Text(),
sa.ForeignKeyConstraint(['node_id'], ['node.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.drop_column('node', 'active_ignition_config')
# ### end Alembic commands ###
def passwords_getitem(self, buf = PAW2_Buffer(self.storage.passwords[key]).pack()
return xmlrpclib.Binary(buf)
def essids_getitem(self, results).pack()
return xmlrpclib.Binary(buf)
def upgrade():
### commands auto generated by Alembic - please adjust! ###
op.create_table('page',
sa.Column('day_id',
sa.Column('sequence',
sa.Column('binary',
sa.Column('title', sa.Unicode(length=1024),
sa.ForeignKeyConstraint(['day_id'], ['day.id'],
sa.PrimaryKeyConstraint('id')
)
### end Alembic commands ###
def upgrade():
### commands auto generated by Alembic - please adjust! ###
op.add_column('page', sa.Column('preview', nullable=True))
op.add_column('page', sa.Column('thumbnail', nullable=True))
### end Alembic commands ###
def upgrade():
for prefix in TABLE_PREFIXES:
op.add_column(prefix + '_torrents', sa.Column('uploader_ip', nullable=True))
# ### end Alembic commands ###
def upgrade():
op.create_table('bans',
sa.Column('created_time',
sa.Column('admin_id',
sa.Column('user_id',
sa.Column('user_ip', sa.Binary(length=16),
sa.Column('reason', sa.String(length=2048),
sa.ForeignKeyConstraint(['admin_id'], ['users.id'],
sa.ForeignKeyConstraint(['user_id'],
sa.PrimaryKeyConstraint('id'),
)
op.create_index('user_ip_16', 'bans', ['user_ip'], unique=True, MysqL_length=16)
op.create_index('user_ip_4', MysqL_length=4)
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')
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('games',
sa.Column('winner', sa.Enum('white', 'black', name='color'),
sa.Column('handicap', sa.SmallInteger(),
sa.Column('komi',
sa.Column('season',
sa.Column('episode',
sa.PrimaryKeyConstraint('id')
)
op.create_table('players',
sa.Column('first_name', sa.String(length=30),
sa.Column('last_name',
sa.Column('aga_id',
sa.Column('aga_rank',
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_players_aga_id'), 'players', ['aga_id'], unique=True)
op.create_table('users',
sa.Column('username', sa.String(length=80),
sa.Column('email',
sa.Column('password',
sa.Column('created_at',
sa.Column('active', sa.Boolean(),
sa.Column('is_admin',
sa.UniqueConstraint('email'),
sa.UniqueConstraint('username')
)
op.create_table('black_player_games',
sa.Column('player_id',
sa.Column('game_id',
sa.ForeignKeyConstraint(['game_id'], ['games.id'],
sa.ForeignKeyConstraint(['player_id'], ['players.id'],
sa.PrimaryKeyConstraint('player_id', 'game_id')
)
op.create_table('roles',
sa.Column('name',
sa.UniqueConstraint('name')
)
op.create_table('white_player_games', 'game_id')
)
# ### end Alembic commands ###
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。
