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

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

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

def upgrade():
    # There can be data truncation here as LargeBinary can be smaller than the pickle 
    # type.

    # use batch_alter_table to support sqlite workaround
    with op.batch_alter_table("xcom") as batch_op:
        batch_op.alter_column('value', type_=sa.LargeBinary())

项目：kNowledge-repo    作者：airbnb    | 项目源码 | 文件源码

def init(self, auto_create=True):

        # Todo handle if user does not pass in table sqlite://path.db
        uri_splt = self.uri.split(":")
        engine_uri = u":".join(uri_splt[:-1])
        table_name = uri_splt[-1]

        Metadata = MetaData()
        postref_table = Table(table_name, Metadata,
                              Column('id', Integer, primary_key=True),
                              Column('created_at', DateTime, default=func.Now()),
                              Column('updated_at', default=func.Now(), onupdate=func.current_timestamp()),
                              Column('uuid', String(512)),
                              Column('path',
                              Column('revision', default=0),
                              Column('status', default=self.PostStatus.DRAFT.value),
                              Column('ref',
                              Column('data', LargeBinary))
        self.engine = create_engine(engine_uri, pool_recycle=3600)
        self.session = scoped_session(sessionmaker(bind=self.engine))
        if auto_create:
            postref_table.create(self.engine, checkfirst=True)

        class PostRef(object):
            pass
        mapper(PostRef, postref_table)
        self.PostRef = PostRef

    # ------------- Repository actions / state ------------------------------------

项目：vcf2db    作者：quinlan-lab    | 项目源码 | 文件源码

def variants_genotype_columns(self):
        return [sql.Column(name, sql.LargeBinary()) for name in self.gt_cols]

项目：glare    作者：openstack    | 项目源码 | 文件源码

def upgrade():
    op.create_table(
        'glare_blob_data',
        sa.Column('id', sa.String(255), primary_key=True, nullable=False),
        # Because of strange behavior of MysqL LargeBinary is converted to
        # BLOB instead of LONGBLOB. So we have to fix it explicitly with
        # 'with_variant' call.
        sa.Column(
            'data',
            sa.LargeBinary().with_variant(MysqL.LONGBLOB(), 'MysqL'),
            nullable=False),
        sa.PrimaryKeyConstraint('id'),
        MysqL_engine=MysqL_ENGINE,
        MysqL_charset=MysqL_CHARSET
    )

项目：Albireo    作者：lordfriend    | 项目源码 | 文件源码

def upgrade():
    ### commands auto generated by Alembic - please adjust! ###
    op.create_table('server_session',
    sa.Column('id', postgresql.UUID(as_uuid=True),
    sa.Column('session_id', sa.TEXT(), nullable=True),
    sa.Column('data', sa.LargeBinary(),
    sa.Column('expiry', sa.DateTime(),
    sa.PrimaryKeyConstraint('id'),
    sa.UniqueConstraint('session_id')
    )
    ### end Alembic commands ###

项目：forget    作者：codl    | 项目源码 | 文件源码

def upgrade():
    op.create_table('accounts',
    sa.Column('created_at', server_default=sa.text('Now()'),
    sa.Column('updated_at', sa.String(),
    sa.Column('policy_enabled', sa.Boolean(), server_default='FALSE',
    sa.Column('policy_keep_latest', sa.Integer(), server_default='0',
    sa.Column('policy_keep_favourites', server_default='TRUE',
    sa.Column('policy_delete_every', sa.Interval(),
    sa.Column('policy_keep_younger',
    sa.Column('display_name',
    sa.Column('screen_name',
    sa.Column('avatar_url',
    sa.Column('last_fetch', server_default='epoch',
    sa.Column('last_delete',
    sa.PrimaryKeyConstraint('id', name=op.f('pk_accounts'))
    )
    op.create_table('oauth_tokens',
    sa.Column('token',
    sa.Column('token_secret',
    sa.Column('account_id',
    sa.ForeignKeyConstraint(['account_id'], ['accounts.id'], name=op.f('fk_oauth_tokens_account_id_accounts'), onupdate='CASCADE', ondelete='CASCADE'),
    sa.PrimaryKeyConstraint('token', name=op.f('pk_oauth_tokens'))
    )
    op.create_table('posts',
    sa.Column('body',
    sa.Column('author_id',
    sa.Column('favourite',
    sa.ForeignKeyConstraint(['author_id'], name=op.f('fk_posts_author_id_accounts'), name=op.f('pk_posts'))
    )
    op.create_table('sessions', name=op.f('fk_sessions_account_id_accounts'), name=op.f('pk_sessions'))
    )
    op.create_table('twitter_archives',
    sa.Column('chunks',
    sa.Column('chunks_successful',
    sa.Column('chunks_Failed', name=op.f('fk_twitter_archives_account_id_accounts'), name=op.f('pk_twitter_archives'))
    )

项目：telemetrics-backend    作者：clearlinux    | 项目源码 | 文件源码

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('build',
    sa.Column('build',
    sa.PrimaryKeyConstraint('id')
    )
    op.create_table('classification',
    sa.Column('classification',
    sa.Column('domain',
    sa.Column('probe',
    sa.PrimaryKeyConstraint('id')
    )
    op.create_table('guilty',
    sa.Column('function',
    sa.Column('module',
    sa.Column('comment',
    sa.Column('hide',
    sa.PrimaryKeyConstraint('id')
    )
    op.create_table('guilty_blacklisted',
    sa.PrimaryKeyConstraint('id')
    )
    op.create_table('records',
    sa.Column('severity',
    sa.Column('machine',
    sa.Column('machine_id',
    sa.Column('architecture',
    sa.Column('kernel_version',
    sa.Column('os_name',
    sa.Column('record_format_version',
    sa.Column('payload_format_version',
    sa.Column('payload',
    sa.Column('tsp',
    sa.Column('tsp_server',
    sa.Column('buildstamp',
    sa.Column('backtrace',
    sa.Column('dupe_of',
    sa.Column('dupecount',
    sa.Column('dupemaster',
    sa.Column('security',
    sa.Column('processed',
    sa.Column('icon',
    sa.Column('classification_id',
    sa.Column('build_id',
    sa.Column('guilty_id',
    sa.Column('external',
    sa.ForeignKeyConstraint(['build_id'], ['build.id'], ),
    sa.ForeignKeyConstraint(['classification_id'], ['classification.id'],
    sa.ForeignKeyConstraint(['guilty_id'], ['guilty.id'],
    sa.PrimaryKeyConstraint('id')
    )
    # ### end Alembic commands ###

项目：websauna    作者：websauna    | 项目源码 | 文件源码

def map_column(self, mode: EditMode, request: Request, node: colander.SchemaNode, model: type, name: str, column: Column, column_type: TypeEngine) -> t.Tuple[colander.SchemaType, dict]:
        """Map non-relationship sqlAlchemy column to Colander SchemaNode.

        :return: Tuple(constructed colander.SchemaType,dict of addtional colander.SchemaNode construction arguments)
        """
        logger.debug("Mapping field %s,mode %s,node %s,column %s,column type %s", name, mode, node, column, column_type)

        # Check for autogenerated columns (updated_at)
        if column.onupdate:
            if mode in (EditMode.edit, EditMode.add):
                return TypeOverridesHandling.drop, {}

        # Don't fill default values when added,as they are automatically populated
        if column.default:
            if mode == EditMode.add:
                return TypeOverridesHandling.drop, {}

        # Never add primary keys
        # NOTE: Todo: We need to preserve ids because of nesting mechanism and groupedit widget wants it id
        if column.primary_key:
            # Todo: Looks like column.autoincrement is set True by default,so we cannot use it here
            if mode in (EditMode.edit, {}

        if column.foreign_keys:

            # Handled by relationship mapper
            return TypeOverridesHandling.drop, {}

        elif isinstance(column_type, (PostgresqlUUID, columns.UUID)):

            # UUID's cannot be22 edited
            if mode in (EditMode.add, EditMode.edit):
                return TypeOverridesHandling.drop, {}

            # But let's show them
            return fields.UUID(), dict(missing=colander.drop, widget=FriendlyUUIDWidget(readonly=True))

        elif isinstance(column_type, Text):
            return colander.String(), dict(widget=deform.widget.TextAreaWidget())
        elif isinstance(column_type, JSONB):
            return JSONValue(), dict(widget=JSONWidget())
        elif isinstance(column_type, (JSONB, columns.JSONB)):
            # Can't edit JSON
            if mode in (EditMode.add, {}
            return colander.String(), {}
        elif isinstance(column_type, LargeBinary):
            # Can't edit binary
            return TypeOverridesHandling.drop, Geometry):
            # Can't edit geometry
            return TypeOverridesHandling.drop, (INET, columns.INET)):
            return colander.String(), {}
        else:
            # Default mapping / unkNown,let the parent handle
            return TypeOverridesHandling.unkNown, {}