Python sqlalchemy 模块,PickleType() 实例源码
我们从Python开源项目中,提取了以下14个代码示例,用于说明如何使用sqlalchemy.PickleType()。
def task_table(base):
class Task(base):
__tablename__ = 'Task'
id_ = Column(Integer, Sequence('task_id_seq'), primary_key=True)
run_id = Column(String, default='')
step_id = Column(String,)
sequence = Column(Integer,)
recovery = Column(Integer, nullable=False)
pid = Column(Integer, nullable=True)
status = Column(sqlEnum(TaskStatus), )
result = Column(PickleType() , nullable=True,)
created = Column(DateTime(), default=datetime.utcNow)
updated = Column(DateTime(), )
__table_args__ = (
UniqueConstraint('run_id', 'step_id', 'sequence', 'recovery'),
)
def __repr__(self):
return "<Task(id='%s',run_id='%s',step_id='%s',sequence='%s',recovery='%s',pid='%s',status='%s',created='%s',updated='%s')>" % (
self.id_, self.run_id, self.step_id, self.sequence, self.recovery, self.pid, self.status, self.created, self.updated)
return Task
def upgrade():
op.create_table('artists',
sa.Column('id', sqlalchemy_utils.types.uuid.UUIDType(), nullable=False),
sa.Column('name', sa.String(), nullable=True),
sa.Column('description',
sa.Column('search_vector', sqlalchemy_utils.types.ts_vector.TsvectorType(),
sa.PrimaryKeyConstraint('id')
)
op.create_table('torrents', sa.String(length=40),
sa.Column('info', sa.PickleType(),
sa.PrimaryKeyConstraint('id')
)
op.create_table('albums',
sa.Column('title',
sa.Column('tracks', sqlalchemy_utils.types.json.JSONType(),
sa.Column('torrent_id',
sa.Column('artist_id',
sa.ForeignKeyConstraint(['artist_id'], ['artists.id'], ),
sa.ForeignKeyConstraint(['torrent_id'], ['torrents.id'],
sa.PrimaryKeyConstraint('id')
)
def downgrade():
op.drop_column('session_data', 'user_id')
op.drop_column('session_data', 'role_id')
op.add_column('session_data', sa.Column('data', sa.PickleType, nullable=True))
def make_cache_table(Metadata, table_name='beaker_cache', schema_name=None):
"""Return a ``Table`` object suitable for storing cached values for the
namespace manager. Do not create the table."""
return sa.Table(table_name, Metadata,
sa.Column('namespace', sa.String(255), primary_key=True),
sa.Column('accessed', sa.DateTime,
sa.Column('created',
sa.Column('data',
schema=schema_name if schema_name else Metadata.schema)
def downgrade():
# use batch_alter_table to support sqlite workaround
with op.batch_alter_table("xcom") as batch_op:
batch_op.alter_column('value', type_=sa.PickleType(pickler=dill))
def upgrade():
op.add_column('dag_run', sa.Column('conf', nullable=True))
def upgrade():
# making ontology terms strings for Now
# leaving out externalId,diseases,pheno,etc. mappings for Now
op.create_table(
'individual', sa.BigInteger,
sa.Column('guid', sa.String(36), nullable=False, unique=True), sa.Text), sa.PickleType),
sa.Column('record_create_time',
sa.Column('record_update_time',
)
def upgrade():
op.execute(CreateSequence(Sequence('callset_id_seq', minvalue=0, start=0)))
op.create_table(
'callset', Sequence('callset_id_seq'),
sa.Column('individual_id', sa.ForeignKey('individual.id'),
sa.Column('dbrow_id', sa.ForeignKey('db_row.id'), sa.Text,
sa.Column('created',
sa.Column('updated', sa.PickleType)
)
def upgrade():
op.create_table('vpp_etcd_journal',
sa.Column('id', sa.Integer, primary_key=True,
autoincrement=True,
sa.Column('k',
sa.Column('v',
sa.Column('retry_count', default=0),
sa.Column('created_at',
default=sa.func.Now()),
sa.Column('last_retried', sa.TIMESTAMP,
server_default=sa.func.Now(),
onupdate=sa.func.Now()))
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('users', sa.Column('spd', nullable=True))
# ### end Alembic commands ###