我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用sqlalchemy.Unicode()。
def upgrade(): op.create_table('sprints', sa.Column('id', sa.Integer(), nullable=False), sa.Column('user_name', sa.Unicode(), nullable=True), sa.Column('creation_date', sa.DateTime(), nullable=True), sa.PrimaryKeyConstraint('id')) op.create_table('retrospective_items', sa.Column('id', sa.Integer(), nullable=False), sa.Column('sprint_id', sa.Integer(), nullable=False), sa.Column('category', sa.Unicode(), nullable=True), sa.Column('text', sa.Unicode(), nullable=True), sa.Column('user_name', sa.Unicode(), nullable=True), sa.Column('creation_date', sa.DateTime(), nullable=True), sa.PrimaryKeyConstraint('id')) op.create_index(op.f('ix_retrospective_items'), 'retrospective_items', ['sprint_id', 'category'], unique=False)
def upgrade(): # ### commands auto generated by Alembic - please adjust! ### op.create_table('user', sa.Column('id', sqlalchemy_utils.types.uuid.UUIDType(), server_default=sa.text('uuid_generate_v4()'), nullable=False), sa.Column('username', sa.Unicode(length=255), nullable=False), sa.Column('password', sqlalchemy_utils.types.password.PasswordType(), nullable=False), sa.Column('created_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False), sa.Column('is_superuser', sa.Boolean(), server_default='FALSE', nullable=False), sa.PrimaryKeyConstraint('id') ) op.create_index(op.f('ix_user_username'), 'user', ['username'], unique=True) op.create_table('user_problem', sa.Column('id', sqlalchemy_utils.types.uuid.UUIDType(), server_default=sa.text('uuid_generate_v4()'), nullable=False), sa.Column('problem_id', sqlalchemy_utils.types.uuid.UUIDType(), nullable=False), sa.Column('user_id', sqlalchemy_utils.types.uuid.UUIDType(), nullable=False), sa.ForeignKeyConstraint(['problem_id'], ['problem.id'], ), sa.ForeignKeyConstraint(['user_id'], ['user.id'], ), sa.PrimaryKeyConstraint('id') ) op.create_index(op.f('ix_user_problem_user_id'), 'user_problem', ['user_id'], unique=False) # ### end Alembic commands ###
def upgrade(): conn = op.get_bind() # ### commands auto generated by Alembic - please adjust! ### op.add_column('projects', sa.Column('task_creation_mode', sa.Integer(), nullable=True)) op.create_index('idx_geometry', 'projects', ['geometry'], unique=False, postgresql_using='gist') op.add_column('tasks', sa.Column('extra_properties', sa.Unicode(), nullable=True)) for project in conn.execute(projects.select()): zooms = conn.execute( sa.sql.expression.select([tasks.c.zoom]).distinct(tasks.c.zoom) .where(tasks.c.project_id == project.id)) zooms = zooms.fetchall() if len(zooms) == 1 and zooms[0] == (None,): op.execute( projects.update().where(projects.c.id == project.id) .values(task_creation_mode=1)) # ### end Alembic commands ###
def downgrade(pyramid_env): # with context.begin_transaction(): # op.add_column("content", sa.Column( # "subject", sa.Unicode, server_default="")) # op.add_column("content", sa.Column( # "body", sa.UnicodeText, server_default="")) from assembl import models as m db = m.get_session_maker()() with transaction.manager: for target in ("subject", "body"): r = db.execute( """select content.id, langstring_entry.value from content join langstring_entry on content.{0}_id = langstring_entry.langstring_id join locale on langstring_entry.locale_id = locale.id where locale.code not like '%-x-mtfrom-%'""".format(target)) for id, text in r: if len(text): db.execute("UPDATE content set %s = :txt WHERE id= :id" % ( (target,)), dict(txt=text, id=id)) mark_changed()
def upgrade(pyramid_env): with context.begin_transaction(): op.create_table( 'timeline_event', sa.Column('id', sa.Integer, primary_key=True), sa.Column('discussion_id', sa.Integer, sa.ForeignKey( 'discussion.id', ondelete='CASCADE', onupdate='CASCADE')), sa.Column('type', sa.String(60), nullable=False), sa.Column('title', sa.Unicode(), nullable=False), sa.Column('description', sa.UnicodeText), sa.Column('start', sa.DateTime), sa.Column('end', sa.DateTime), sa.Column('previous_event_id', sa.Integer, sa.ForeignKey('timeline_event.id')), )
def downgrade(pyramid_env): with context.begin_transaction(): from assembl import models as m db = m.get_session_maker()() with transaction.manager: # Undo correcting of the spelling mistake in attachment model. a = m.Attachment.__table__ db.execute( a.update().where(a.c.attachmentPurpose == op.inline_literal( 'EMBED_ATTACHMENT')). values(attachmentPurpose=op.inline_literal( "EMBEEDED_ATTACHMENT")) ) op.drop_column('facebook_source', 'upper_bound') op.drop_column('facebook_source', 'lower_bound') op.drop_column('facebook_post', 'attachment_blob') op.add_column('facebook_post', sa.Column('post_type', sa.String(20))) op.add_column('facebook_post', sa.Column('link_name', sa.Unicode(1024))) op.add_column('facebook_post', sa.Column('attachment', sa.String(1024)))
def upgrade(pyramid_env): # Do stuff with the app's models here. from assembl import models as m db = m.get_session_maker()() with transaction.manager: titles = list(db.execute( """SELECT id, title FROM document""")) with context.begin_transaction(): op.drop_column('document', 'title') op.add_column('document', sa.Column('title', sa.Unicode(1024), server_default="")) with transaction.manager: docs = m.Document.__table__ for (id, title) in titles: db.execute(docs.update().where(docs.c.id == id). values(title=sa.sql.expression.cast( UnicodeDammit(title).unicode_markup, sa.Unicode))) mark_changed()
def upgrade(): table_prefix = context.config.get_main_option('table_prefix') op.create_table( table_prefix + 'repos', sa.Column('id', sa.Integer(), nullable=False, primary_key=True), sa.Column('repo_name', sa.Unicode(100), nullable=False), sa.Column('env_id', sa.Integer(), nullable=False), sa.Column('git_url', sa.String(255), server_default='', nullable=False), sa.Column('ref', sa.String(255), server_default='', nullable=False), sa.Column('user_key', sa.String(255), server_default='', nullable=False), sa.UniqueConstraint('env_id', name='_env_id_unique'))
def on_connect(self): if self.cx_oracle_ver < (5,): # no output type handlers before version 5 return cx_Oracle = self.dbapi def output_type_handler(cursor, name, defaultType, size, precision, scale): # convert all NUMBER with precision + positive scale to Decimal # this almost allows "native decimal" mode. if self.supports_native_decimal and \ defaultType == cx_Oracle.NUMBER and \ precision and scale > 0: return cursor.var( cx_Oracle.STRING, 255, outconverter=self._to_decimal, arraysize=cursor.arraysize) # if NUMBER with zero precision and 0 or neg scale, this appears # to indicate "ambiguous". Use a slower converter that will # make a decision based on each value received - the type # may change from row to row (!). This kills # off "native decimal" mode, handlers still needed. elif self.supports_native_decimal and \ defaultType == cx_Oracle.NUMBER \ and not precision and scale <= 0: return cursor.var( cx_Oracle.STRING, 255, outconverter=self._detect_decimal, arraysize=cursor.arraysize) # allow all strings to come back natively as Unicode elif self.coerce_to_unicode and \ defaultType in (cx_Oracle.STRING, cx_Oracle.FIXED_CHAR): return cursor.var(util.text_type, size, cursor.arraysize) def on_connect(conn): conn.outputtypehandler = output_type_handler return on_connect
def upgrade(): op.add_column('app', sa.Column('long_description', sa.Unicode))
def upgrade(): op.create_table( 'announcement', sa.Column('id', sa.Integer, primary_key=True), sa.Column('title', sa.Unicode(length=255), nullable=False), sa.Column('body', sa.UnicodeText, nullable=False), sa.Column('user_id', sa.Integer, sa.ForeignKey('user.id')), sa.Column('created', sa.Text, default=make_timestamp), )
def downgrade(): op.alter_column('app', 'long_description', type_=sa.Unicode) pass
def upgrade(): op.create_table( 'blogpost', sa.Column('id', sa.Integer, primary_key=True), sa.Column('title', sa.Unicode(length=255), nullable=False), sa.Column('body', sa.UnicodeText, nullable=False), sa.Column('app_id', sa.Integer, sa.ForeignKey('app.id', ondelete='CASCADE'), nullable=False), sa.Column('user_id', sa.Integer, sa.ForeignKey('user.id')), sa.Column('created', sa.Text, default=make_timestamp), )
def upgrade(): ### commands auto generated by Alembic - please adjust! ### op.add_column('upload', sa.Column('created', sa.DateTime(), nullable=True)) op.add_column('upload', sa.Column('original_name', sa.Unicode(length=512), nullable=True)) ### end Alembic commands ###
def upgrade(): # ### commands auto generated by Alembic - please adjust! ### op.add_column('RubricItem', sa.Column('header', sa.Unicode(), nullable=True)) # ### end Alembic commands ###
def upgrade(): # ### commands auto generated by Alembic - please adjust! ### op.add_column('User', sa.Column('username', sa.Unicode(), nullable=False)) op.alter_column('User', 'email', existing_type=sa.VARCHAR(), nullable=False) op.create_index(op.f('ix_User_username'), 'User', ['username'], unique=True) op.drop_constraint('User_email_key', 'User', type_='unique') # ### end Alembic commands ###
def upgrade(): # ### commands auto generated by Alembic - please adjust! ### op.add_column('Assignment', sa.Column('cgignore', sa.Unicode(), nullable=True)) # ### end Alembic commands ###
def downgrade(): # ### commands auto generated by Alembic - please adjust! ### op.add_column('label_event', sa.Column('label', sa.VARCHAR(length=255), autoincrement=False, nullable=True)) op.add_column('dataset', sa.Column('probability', postgresql.DOUBLE_PRECISION(precision=53), autoincrement=False, nullable=True)) op.add_column('problem', sa.Column('label', sa.Unicode(length=255), nullable=True)) op.execute(''' UPDATE label_event SET label = (SELECT label FROM problem_label WHERE problem_label.id = label_event.label_id) ''') op.execute(''' UPDATE dataset SET probability = (SELECT probability FROM dataset_label_probability WHERE dataset_label_probability.data_id = dataset.id) ''') op.execute(''' UPDATE problem SET label = (SELECT label FROM problem_label WHERE problem_label.problem_id = problem.id) ''') op.alter_column('label_event', 'label', nullable=False) op.alter_column('problem', 'label', nullable=False) op.drop_column('problem', 'name') op.drop_index(op.f('ix_label_event_label_id'), table_name='label_event') op.drop_column('label_event', 'label_id') op.drop_index(op.f('ix_dataset_label_probability_label_id'), table_name='dataset_label_probability') op.drop_index(op.f('ix_dataset_label_probability_data_id'), table_name='dataset_label_probability') op.drop_table('dataset_label_probability') op.drop_table('problem_label') # ### end Alembic commands ###
def upgrade(): # ### commands auto generated by Alembic - please adjust! ### op.add_column('tool', sa.Column('link', sa.Unicode(), nullable=True)) # ### end Alembic commands ###
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 ###
def upgrade(): ### commands auto generated by Alembic - please adjust! ### op.create_table('item_list', sa.Column('id', sa.Integer(), nullable=False), sa.Column('name', sa.Unicode(length=256), nullable=True), sa.Column('content', sa.Unicode(length=4294967296), nullable=True), sa.PrimaryKeyConstraint('id') ) ### end Alembic commands ###
def Team(self, Base, encryption_engine): self._team_key = None class Team(Base): __tablename__ = 'team' id = sa.Column(sa.Integer, primary_key=True) key = sa.Column(sa.String(50)) name = sa.Column(EncryptedType( sa.Unicode, lambda: self._team_key, encryption_engine) ) return Team
def Building(Base): class Building(Base): __tablename__ = 'building' id = sa.Column('_id', sa.Integer, primary_key=True) name = sa.Column('_name', sa.Unicode(255)) return Building
def Building(Base): class Building(Base): __tablename__ = 'building' id = sa.Column(sa.Integer, primary_key=True) name = sa.Column('_name', sa.Unicode(255)) return Building
def upgrade(migrate_engine): meta = sa.MetaData(bind=migrate_engine) tags = sa.Table('tags', meta, sa.Column('resource_id', sa.String(36), primary_key=True, nullable=False), sa.Column('tag', sa.Unicode(80), primary_key=True, nullable=False), sa.Index('tags_tag_idx', 'tag'), mysql_engine='InnoDB', mysql_charset='utf8') tags.create()
def upgrade(pyramid_env): with context.begin_transaction(): op.create_table( 'preferences', sa.Column('id', sa.Integer, primary_key=True), sa.Column('name', sa.Unicode, nullable=False), sa.Column('cascade_id', sa.Integer, sa.ForeignKey('preferences.id'), nullable=True), sa.Column('values', sa.Text())) op.add_column( 'discussion', sa.Column('preferences_id', sa.Integer, sa.ForeignKey("preferences.id"))) # The contents are not worth saving yet. op.drop_column('discussion', 'settings')
