Python sqlalchemy.dialects.postgresql 模块，HSTORE 实例源码

def test_dynamic_locale(self, Base):
        translation_hybrid = TranslationHybrid(
            lambda obj: obj.locale,
            'fi'
        )

        class Article(Base):
            __tablename__ = 'article'
            id = sa.Column(sa.Integer, primary_key=True)
            name_translations = sa.Column(HSTORE)
            name = translation_hybrid(name_translations)
            locale = sa.Column(sa.String)

        assert (
            'coalesce(article.name_translations -> article.locale'
            in str(Article.name.expression)
        )

def test_locales_casted_only_in_compilation_phase(self, Base):
        class LocaleGetter(object):
            def current_locale(self):
                return lambda obj: obj.locale

        flexmock(LocaleGetter).should_receive('current_locale').never()
        translation_hybrid = TranslationHybrid(
            LocaleGetter().current_locale,
            'fi'
        )

        class Article(Base):
            __tablename__ = 'article'
            id = sa.Column(sa.Integer, primary_key=True)
            name_translations = sa.Column(HSTORE)
            name = translation_hybrid(name_translations)
            locale = sa.Column(sa.String)

        Article.name

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('bit_performance_report',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('date', sa.DateTime(), nullable=True),
    sa.Column('year', sa.Integer(), nullable=True),
    sa.Column('month', sa.Integer(), nullable=True),
    sa.Column('day', sa.Integer(), nullable=True),
    sa.Column('name', sa.String(length=250), nullable=True),
    sa.Column('campaign_source', sa.String(length=250), nullable=True),
    sa.Column('campaign_name', sa.String(length=250), nullable=True),
    sa.Column('campaign_id', sa.String(length=250), nullable=True),
    sa.Column('clicks', sa.Integer(), nullable=True),
    sa.Column('clicks_unique', sa.Integer(), nullable=True),
    sa.Column('impressions', sa.Integer(), nullable=True),
    sa.Column('conversions', sa.Integer(), nullable=True),
    sa.Column('cost', sa.Numeric(precision=17, scale=5), nullable=True),
    sa.Column('breakdowns', postgresql.HSTORE(text_type=sa.Text()), nullable=True),
    sa.Column('measurements', postgresql.HSTORE(text_type=sa.Text()), nullable=True),
    sa.PrimaryKeyConstraint('id')
    )
    op.create_index(op.f('ix_bit_performance_report_breakdowns'), 'bit_performance_report', ['breakdowns'], unique=False)
    op.create_index(op.f('ix_bit_performance_report_campaign_id'), 'bit_performance_report', ['campaign_id'], unique=False)
    op.create_index(op.f('ix_bit_performance_report_campaign_name'), 'bit_performance_report', ['campaign_name'], unique=False)
    op.create_index(op.f('ix_bit_performance_report_campaign_source'), 'bit_performance_report', ['campaign_source'], unique=False)
    op.create_index(op.f('ix_bit_performance_report_date'), 'bit_performance_report', ['date'], unique=False)
    op.create_index(op.f('ix_bit_performance_report_day'), 'bit_performance_report', ['day'], unique=False)
    op.create_index(op.f('ix_bit_performance_report_month'), 'bit_performance_report', ['month'], unique=False)
    op.create_index(op.f('ix_bit_performance_report_name'), 'bit_performance_report', ['name'], unique=False)
    op.create_index(op.f('ix_bit_performance_report_year'), 'bit_performance_report', ['year'], unique=False)
    # ### end Alembic commands ###

def clear_column_type(cls, column_type=''):
        if column_type:
            column_type = column_type.split(') ')
            l = len(column_type)
            column_type = column_type[0]
            if l:
                if l > 1:
                    column_type = '{})'.format(column_type)

            column_type = 'sa.{}'.format(column_type)

            if 'INTEGER' in column_type or 'TINYINT' in column_type\
                    or 'BIGINT' in column_type:
                column_type = 'sa.Integer()'

            if 'string' in column_type:
                column_type = 'sa.Text()'

            if 'OBJECT' in column_type:
                column_type = 'sa.Text()'

            if 'DATETIME' in column_type or 'TIMESTAMP WITHOUT TIME ZONE' \
                    in column_type or 'TIMESTAMP WITH TIME ZONE'\
                    in column_type:
                column_type = 'sa.DateTime()'

            if 'HSTORE' in column_type:
                # column_type = 'postgresql.HSTORE(text_type=sa.Text())'
                column_type = 'postgresql.HSTORE'
        else:
            column_type = 'sa.Text()'

        return column_type

def is_text(self, col_name):
        try:
            return (
                isinstance(self.list_columns[col_name].type, sa.types.String)
                or isinstance(self.list_columns[col_name].type, HSTORE)
            )
        except Exception:
            return False

def test_get_field_type_can_map_pg_hstore_column(self):

        column = sqa.Column(postgresql.HSTORE())
        field = field_mapping.get_field_type(column)

        self.assertTrue(issubclass(field, CharMappingField))

def City(Base, translation_hybrid):
    class City(Base):
        __tablename__ = 'city'
        id = sa.Column(sa.Integer, primary_key=True)
        name_translations = sa.Column(HSTORE)
        name = translation_hybrid(name_translations)
        locale = 'en'
    return City

def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('user_media_links', sa.Column('attributes', postgresql.HSTORE(text_type=Text()), server_default=sa.text(u"''::hstore"), autoincrement=False, nullable=False))
    op.add_column('submission_media_links', sa.Column('attributes', postgresql.HSTORE(text_type=Text()), server_default=sa.text(u"''::hstore"), autoincrement=False, nullable=False))
    op.add_column('media_media_links', sa.Column('attributes', postgresql.HSTORE(text_type=Text()), server_default=sa.text(u"''::hstore"), autoincrement=False, nullable=False))
    # ### end Alembic commands ###