Python sqlalchemy_utils 模块，TSVectorType() 实例源码

def upgrade():
    op.create_table(
        'pokemon',
        sa.Column('id', sa.Integer, primary_key=True),
        sa.Column('name', sa.Text, unique=True, nullable=False),
        sa.Column('flavor_text', sa.Text, nullable=False),
        sa.Column('habitat', sa.Text, nullable=True, default=None),
        sa.Column('color', sa.Text, nullable=False),
        sa.Column('shape', sa.Text, nullable=False),

        sa.Column('search_vector', TSVectorType(searchable)),

        sa.Column('inserted_at', sa.DateTime,
                  default=sa.func.current_timestamp(), nullable=False),
        sa.Column('updated_at', sa.DateTime,
                  default=sa.func.current_timestamp(),
                  onupdate=sa.func.current_timestamp(), nullable=False)
    )

    conn = op.get_bind()
    sync_trigger(conn, 'pokemon', 'search_vector', searchable)

def upgrade():
    op.create_table(
        'pokedexes',
        sa.Column('id', sa.Integer, primary_key=True),
        sa.Column('name', sa.Text, unique=True, nullable=False),
        sa.Column('official_name', sa.Text, unique=True, nullable=False),
        sa.Column('region', sa.Text, nullable=True, default=None),
        sa.Column('description', sa.Text, nullable=True, default=None),

        sa.Column('search_vector', TSVectorType(searchable)),

        sa.Column('inserted_at', sa.DateTime,
                  default=sa.func.current_timestamp(), nullable=False),
        sa.Column('updated_at', sa.DateTime,
                  default=sa.func.current_timestamp(),
                  onupdate=sa.func.current_timestamp(), nullable=False)
    )

    conn = op.get_bind()
    sync_trigger(conn, 'pokedexes', 'search_vector', searchable)

def upgrade():
    op.create_table(
        'moves',
        sa.Column('id', sa.Integer, primary_key=True),
        sa.Column('name', sa.Text, unique=True, nullable=False),
        sa.Column('flavor_text', sa.Text, nullable=True, default=None),
        sa.Column('short_effect', sa.Text, nullable=False),
        sa.Column('effect', sa.Text, nullable=False),
        sa.Column('damage_class', sa.Text, nullable=True, default=None),
        sa.Column('power_points', sa.Integer, nullable=True, default=None),
        sa.Column('power', sa.Integer, nullable=True, default=None),
        sa.Column('accuracy', sa.Integer, nullable=True, default=None),

        sa.Column('search_vector', TSVectorType(searchable)),

        sa.Column('inserted_at', sa.DateTime,
                  default=sa.func.current_timestamp(), nullable=False),
        sa.Column('updated_at', sa.DateTime,
                  default=sa.func.current_timestamp(),
                  onupdate=sa.func.current_timestamp(), nullable=False)
    )

    conn = op.get_bind()
    sync_trigger(conn, 'moves', 'search_vector', searchable)

def __init__(self, *args, **kwargs):
        """
        Initializes new TSVectorType

        :param *args: list of column names
        :param **kwargs: various other options for this TSVectorType
        """
        self.columns = args
        self.options = kwargs
        super(TSVectorType, self).__init__()

def User(Base):
    class User(Base):
        __tablename__ = 'user'
        id = sa.Column(sa.Integer, primary_key=True)
        name = sa.Column(sa.Unicode(255))
        search_index = sa.Column(
            TSVectorType(name, regconfig='pg_catalog.finnish')
        )

        def __repr__(self):
            return 'User(%r)' % self.id
    return User

def test_catalog_and_columns_as_args(self):
        type_ = TSVectorType('name', 'age', regconfig='pg_catalog.simple')
        assert type_.columns == ('name', 'age')
        assert type_.options['regconfig'] == 'pg_catalog.simple'

def tsvector_reduce_concat(vectors):
    return sa.sql.expression.cast(
        sa.func.coalesce(
            sa.func.array_to_string(sa.func.array_agg(vectors), ' ')
        ),
        TSVectorType
    )

def Catalog(Base, Product):
    class Catalog(Base):
        __tablename__ = 'catalog'
        id = sa.Column(sa.Integer, primary_key=True)
        name = sa.Column(sa.Unicode(255))

        @aggregated('products', sa.Column(TSVectorType))
        def product_search_vector(self):
            return tsvector_reduce_concat(
                sa.func.to_tsvector(Product.name)
            )

        products = sa.orm.relationship('Product', backref='catalog')
    return Catalog