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

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

项目：eventor    作者：Acrisel    | 项目源码 | 文件源码

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

项目：tracker    作者：Songbee    | 项目源码 | 文件源码

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')
    )

项目：kuberdock-platform    作者：cloudlinux    | 项目源码 | 文件源码

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))

项目：MCSManager-fsmodule    作者：Suwings    | 项目源码 | 文件源码

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)

项目：incubator-airflow-old    作者：apache    | 项目源码 | 文件源码

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))

项目：incubator-airflow-old    作者：apache    | 项目源码 | 文件源码

def upgrade():
    op.add_column('dag_run', sa.Column('conf', nullable=True))

项目：download-manager    作者：thispc    | 项目源码 | 文件源码

def make_cache_table(Metadata, table_name='beaker_cache'):
    """Return a ``Table`` object suitable for storing cached values for the
    namespace manager.  Do not create the table."""
    return sa.Table(table_name, nullable=False))

项目：python-group-proj    作者：Sharcee    | 项目源码 | 文件源码

def make_cache_table(Metadata,
                    schema=schema_name if schema_name else Metadata.schema)

项目：GenomicsSampleAPIs    作者：Intel-HLS    | 项目源码 | 文件源码

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',
    )

项目：GenomicsSampleAPIs    作者：Intel-HLS    | 项目源码 | 文件源码

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)
    )

项目：networking-vpp    作者：openstack    | 项目源码 | 文件源码

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()))

项目：airflow    作者：apache-airflow    | 项目源码 | 文件源码

def upgrade():
    op.add_column('dag_run', nullable=True))

项目：NJU_helper    作者：llllnvidia    | 项目源码 | 文件源码

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

项目：networking-vpp    作者：iawells    | 项目源码 | 文件源码

def upgrade():
    op.create_table('vpp_etcd_journal',
    sa.Column('id', autoincrement=True,
    sa.Column('k',
    sa.Column('v',
    sa.Column('retry_count',
    sa.Column('created_at', server_default=sa.func.Now()),
    sa.Column('last_retried', server_default=sa.func.Now(),
          onupdate=sa.func.Now()))