python类UUID的实例源码

def auto_uuid():
    uuid_gen_expr = sa.text('uuid_generate_v4()')
    return sa.Column(sap.UUID(as_uuid=True),
                     primary_key=True, server_default=uuid_gen_expr)

def load_dialect_impl(self, dialect):
        if dialect.name == 'postgresql':
            return dialect.type_descriptor(UUID())
        elif dialect.name == 'oracle':
            return dialect.type_descriptor(RAW(16))
        elif dialect.name == 'mysql':
            return dialect.type_descriptor(BINARY(16))
        else:
            return dialect.type_descriptor(CHAR(32))

def process_bind_param(self, value, dialect):
        if value is None:
            return value
        elif dialect.name == 'postgresql':
            return str(value).lower()
        elif dialect.name == 'oracle':
            return uuid.UUID(value).bytes
        elif dialect.name == 'mysql':
            return uuid.UUID(value).bytes
        else:
            if not isinstance(value, uuid.UUID):
                return "%.32x" % uuid.UUID(value)
            else:
                # hexstring
                return "%.32x" % value

def process_result_value(self, value, dialect):
        if value is None:
            return value
        elif dialect.name == 'oracle':
            return str(uuid.UUID(bytes=value)).replace('-', '').lower()
        elif dialect.name == 'mysql':
            return str(uuid.UUID(bytes=value)).replace('-', '').lower()
        else:
            return str(uuid.UUID(value)).replace('-', '').lower()

def load_dialect_impl(self, dialect):
        if dialect.name == 'postgresql' and self.native:
            # Use the native UUID type.
            return dialect.type_descriptor(postgresql.UUID())

        else:
            # Fallback to either a BINARY or a CHAR.
            kind = self.impl if self.binary else types.CHAR(32)
            return dialect.type_descriptor(kind)

def _coerce(value):
        if value and not isinstance(value, uuid.UUID):
            try:
                value = uuid.UUID(value)

            except (TypeError, ValueError):
                value = uuid.UUID(bytes=value)

        return value

def process_bind_param(self, value, dialect):
        if value is None:
            return value

        if not isinstance(value, uuid.UUID):
            value = self._coerce(value)

        if self.native and dialect.name == 'postgresql':
            return str(value)

        return value.bytes if self.binary else value.hex

def process_result_value(self, value, dialect):
        if value is None:
            return value

        if self.native and dialect.name == 'postgresql':
            if isinstance(value, uuid.UUID):
                # Some drivers convert PostgreSQL's uuid values to
                # Python's uuid.UUID objects by themselves
                return value
            return uuid.UUID(value)

        return uuid.UUID(bytes=value) if self.binary else uuid.UUID(value)

def upgrade():
    if current_env in ['test', 'dev']:
        users_table = table(
            'users',
            sa.Column('id', postgresql.UUID, server_default=sa.text('uuid_generate_v1()'), primary_key=True),
            sa.Column('name', sa.Text, nullable=False, unique=True),
            sa.Column('password', sa.Text, nullable=False),
            sa.Column('modified', sa.DateTime, server_default=sa.text('clock_timestamp()')),
            sa.Column('created', sa.DateTime, server_default=sa.text('now()'))
        )

        op.bulk_insert(users_table, [{'name': "test_user", 'password': hash_password('test123')}])

def upgrade():

    connection = op.get_bind()
    connection.execute('CREATE EXTENSION IF NOT EXISTS "uuid-ossp";')

    op.create_table(
        'users',
        sa.Column('id', postgresql.UUID, server_default=sa.text('uuid_generate_v1()'), primary_key=True),
        sa.Column('name', sa.Text, nullable=False, unique=True),
        sa.Column('password', sa.Text, nullable=False),
        sa.Column('modified', sa.DateTime, server_default=sa.text('clock_timestamp()')),
        sa.Column('created', sa.DateTime, server_default=sa.text('now()')),
    )

def load_dialect_impl(self, dialect):
        if dialect.name == 'postgresql':
            return dialect.type_descriptor(UUID())
        else:
            return dialect.type_descriptor(CHAR(32))

def process_bind_param(self, value, dialect):
        if value is None:
            return value
        elif dialect.name == 'postgresql':
            return str(value)
        else:
            if not isinstance(value, uuid.UUID):
                return "%.32x" % uuid.UUID(value).int
            else:
                # hexstring
                return "%.32x" % value.int

def process_result_value(self, value, dialect):
        if value is None:
            return value
        else:
            return uuid.UUID(value)

def get_species_birth_generation(self, scenario_id, species_id, session=None):
        if session is None:
            session = self.Session()

        return self.get_individuals(scenario_id=scenario_id, session=session) \
            .with_entities(Individual.generation) \
            .filter(Individual.species == uuid.UUID(int=species_id)) \
            .order_by(Individual.generation) \
            .first() \
            .generation

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.execute('CREATE EXTENSION IF NOT EXISTS "uuid-ossp";')
    op.create_table('translated_string',
    sa.Column('id', postgresql.UUID(as_uuid=True), server_default=sa.text('uuid_generate_v4()'), nullable=False),
    sa.Column('base_string', sa.Text(), nullable=True),
    sa.Column('translation', sa.Text(), nullable=True),
    sa.Column('comment', sa.Text(), nullable=True),
    sa.Column('translator_comment', sa.Text(), nullable=True),
    sa.Column('context', sa.Text(), nullable=True),
    sa.PrimaryKeyConstraint('id')
    )
    # ### end Alembic commands ###

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('po_metadata',
    sa.Column('id', postgresql.UUID(as_uuid=True), server_default=sa.text('uuid_generate_v4()'), nullable=False),
    sa.Column('resource_pk', postgresql.UUID(as_uuid=True), nullable=False),
    sa.Column('language_pk', sa.Integer(), nullable=False),
    sa.Column('key', sa.Text(), nullable=False),
    sa.Column('value', sa.Text(), nullable=False),
    sa.ForeignKeyConstraint(['language_pk'], ['language.id'], ),
    sa.ForeignKeyConstraint(['resource_pk'], ['resource.id'], ),
    sa.PrimaryKeyConstraint('id'),
    sa.UniqueConstraint('resource_pk', 'language_pk', 'key', name='resource_language_key_uc')
    )
    # ### end Alembic commands ###

def upgrade():
    op.create_table('resource',
    sa.Column('id', postgresql.UUID(as_uuid=True), server_default=sa.text('uuid_generate_v4()'), nullable=False),
    sa.PrimaryKeyConstraint('id')
    )
    op.add_column('translated_string', sa.Column('resource_pk', postgresql.UUID(as_uuid=True), nullable=False))
    op.create_foreign_key(None, 'translated_string', 'resource', ['resource_pk'], ['id'])

def downgrade():
    op.drop_table('translated_string')
    op.create_table('translated_string',
    sa.Column('id', postgresql.UUID(), server_default=sa.text('uuid_generate_v4()'), autoincrement=False, nullable=False),
    sa.Column('base_string', sa.TEXT(), autoincrement=False, nullable=False),
    sa.Column('translation', sa.TEXT(), autoincrement=False, nullable=False),
    sa.Column('comment', sa.TEXT(), autoincrement=False, nullable=False),
    sa.Column('translator_comment', sa.TEXT(), autoincrement=False, nullable=False),
    sa.Column('context', sa.TEXT(), autoincrement=False, nullable=False),
    sa.Column('resource_pk', postgresql.UUID(), autoincrement=False, nullable=False),
    sa.ForeignKeyConstraint(['resource_pk'], ['resource.id'], name='translated_string_resource_pk_fkey'),
    sa.PrimaryKeyConstraint('id', name='translated_string_pkey')
    )
    op.drop_table('base_string')

def load_dialect_impl(self, dialect):
        if dialect.name == 'postgresql':
            return dialect.type_descriptor(UUID())
        else:
            return dialect.type_descriptor(CHAR(32))

def process_bind_param(self, value, dialect):
        if value is None:
            return value
        elif dialect.name == 'postgresql':
            return str(value)
        else:
            if not isinstance(value, uuid.UUID):
                return '%.32x' % uuid.UUID(value).int
            else:
                # hexstring
                return '%.32x' % value.int