Source code for search.upgrade

""" Contains upgrade tasks that are executed when the application is being
upgraded on the server. See :class:`onegov.core.upgrade.upgrade_task`.

"""
from __future__ import annotations

from onegov.core.upgrade import upgrade_task, UpgradeContext
from onegov.search.utils import searchable_sqlalchemy_models
from sqlalchemy import inspect, String
from sqlalchemy.dialects.postgresql import ARRAY


@upgrade_task('Remove fts index column from all searchable models')
[docs] def remove_fts_index_columns(context: UpgradeContext) -> None: tables = { model.__tablename__ for base in context.app.session_manager.bases for model in searchable_sqlalchemy_models(base) } for table in tables: if context.has_column(table, 'fts_idx'): context.operations.drop_column(table, 'fts_idx')
@upgrade_task('Make last_change nullable')
[docs] def make_last_change_nullable(context: UpgradeContext) -> None: if context.has_table('search_index'): context.operations.alter_column( 'search_index', 'last_change', nullable=True )
@upgrade_task('Change tags from HSTORE to ARRAY')
[docs] def change_tags_from_hstore_to_array(context: UpgradeContext) -> None: if context.has_table('search_index'): ins = inspect(context.engine) for meta in ins.get_columns('search_index', context.schema): if meta['name'] == 'tags': current_type = meta['type'] break else: # the column doesn't exist, although this shouldn't happen # let's just do nothing for now, if it ever does return if isinstance(current_type, ARRAY): # the column is already the correct type return context.operations.alter_column( 'search_index', 'tags', type_=ARRAY(String), postgresql_using='akeys(tags)' )