""" Contains upgrade tasks that are executed when the application is being
upgraded on the server. See :class:`onegov.core.upgrade.upgrade_task`.
"""
from onegov.core.orm.types import HSTORE
from onegov.core.orm.types import JSON
from onegov.core.orm.types import UTCDateTime
from onegov.core.upgrade import upgrade_task
from onegov.core.upgrade import UpgradeContext
from sqlalchemy import Boolean
from sqlalchemy import Column
from sqlalchemy import Enum
from sqlalchemy import ForeignKey
from sqlalchemy import Text
@upgrade_task('Create archived results')
[docs]
def create_archived_results(context: UpgradeContext) -> None:
pass # obsolete data migration
@upgrade_task('Add ID to archived results')
[docs]
def add_id_to_archived_results(context: UpgradeContext) -> None:
pass # obsolete data migration
@upgrade_task('Update vote progress')
[docs]
def update_vote_progress(context: UpgradeContext) -> None:
pass # obsolete data migration
@upgrade_task('Add elected candidates to archived results')
[docs]
def add_elected_candidates(context: UpgradeContext) -> None:
pass # obsolete data migration
@upgrade_task('Add content columns to archived results')
[docs]
def add_content_columns_to_archived_results(context: UpgradeContext) -> None:
if not context.has_column('archived_results', 'content'):
context.operations.add_column(
'archived_results', Column('content', JSON)
)
@upgrade_task('Change last change columns')
[docs]
def change_last_change_columns(context: UpgradeContext) -> None:
if not context.has_column('archived_results', 'last_modified'):
context.operations.add_column(
'archived_results',
Column('last_modified', UTCDateTime, nullable=True)
)
if context.has_column('archived_results', 'last_result_change'):
context.operations.execute(
'ALTER TABLE {} ALTER COLUMN {} DROP NOT NULL;'.format(
'archived_results', 'last_result_change'
)
)
if (
context.has_column('notifications', 'last_change')
and not context.has_column('notifications', 'last_modified')
):
context.operations.execute(
'ALTER TABLE {} RENAME COLUMN {} TO {};'.format(
'notifications', 'last_change', 'last_modified'
)
)
if context.has_column('notifications', 'last_modified'):
context.operations.execute(
'ALTER TABLE {} ALTER COLUMN {} DROP NOT NULL;'.format(
'notifications', 'last_modified'
)
)
@upgrade_task('Make subscriber polymorphic')
[docs]
def make_subscriber_polymorphic(context: UpgradeContext) -> None:
if not context.has_column('subscribers', 'type'):
context.operations.add_column(
'subscribers',
Column('type', Text, nullable=True)
)
if (
context.has_column('subscribers', 'phone_number')
and not context.has_column('subscribers', 'address')
):
context.operations.execute(
'ALTER TABLE {} RENAME COLUMN {} TO {};'.format(
'subscribers', 'phone_number', 'address'
)
)
@upgrade_task('Make notifications polymorphic')
[docs]
def make_notifications_polymorphic(context: UpgradeContext) -> None:
if (
context.has_column('notifications', 'action')
and not context.has_column('notifications', 'type')
):
context.operations.execute(
'ALTER TABLE {} RENAME COLUMN {} TO {};'.format(
'notifications', 'action', 'type'
)
)
context.operations.execute(
'ALTER TABLE {} ALTER COLUMN {} DROP NOT NULL;'.format(
'notifications', 'type'
)
)
@upgrade_task('Apply static data')
[docs]
def apply_static_data(context: UpgradeContext) -> None:
pass # obsolete data migration
@upgrade_task('Add election compound to archive')
[docs]
def add_election_compound_to_archive(context: UpgradeContext) -> None:
old_type = Enum('election', 'vote', name='type_of_result')
new_type = Enum(
'election', 'election_compound', 'vote', name='type_of_result'
)
tmp_type = Enum(
'election', 'election_compound', 'vote', name='_type_of_result'
)
tmp_type.create(context.operations.get_bind(), checkfirst=False)
context.operations.execute(
'ALTER TABLE archived_results ALTER COLUMN type '
'TYPE _type_of_result USING type::text::_type_of_result'
)
old_type.drop(context.operations.get_bind(), checkfirst=False)
new_type.create(context.operations.get_bind(), checkfirst=False)
context.operations.execute(
'ALTER TABLE archived_results ALTER COLUMN type '
'TYPE type_of_result USING type::text::type_of_result'
)
tmp_type.drop(context.operations.get_bind(), checkfirst=False)
@upgrade_task('Add contraints to notifications and sources')
[docs]
def add_contraints_to_notifications_and_sources(
context: UpgradeContext
) -> None:
# We use SQL (rather than operations.xxx) so that we can drop and add
# the constraints in one statement
for ref in ('election', 'vote'):
for table in ('notifications', 'upload_data_source_item'):
context.operations.execute(
f'ALTER TABLE {table} '
f'DROP CONSTRAINT {table}_{ref}_id_fkey, '
f'ADD CONSTRAINT {table}_{ref}_id_fkey'
f' FOREIGN KEY ({ref}_id) REFERENCES {ref}s (id)'
f' ON UPDATE CASCADE'
)
@upgrade_task('Enable expats on votes and elections')
[docs]
def enable_expats(context: UpgradeContext) -> None:
pass # obsolete data migration
@upgrade_task('Adds active column to subscriber')
[docs]
def add_active_column_to_subscriver(context: UpgradeContext) -> None:
if not context.has_column('subscribers', 'active'):
context.operations.add_column(
'subscribers',
Column('active', Boolean, nullable=True)
)
@upgrade_task('Add election compound notification')
[docs]
def add_election_compound_notification(context: UpgradeContext) -> None:
if not context.has_column('notifications', 'election_compound_id'):
context.operations.add_column(
'notifications',
Column(
'election_compound_id',
Text,
ForeignKey('election_compounds.id', onupdate='CASCADE'),
nullable=True
)
)
@upgrade_task('Make election day models polymorphic type non-nullable')
[docs]
def make_election_day_models_polymorphic_type_non_nullable(
context: UpgradeContext
) -> None:
for table in ('notifications', 'subscribers'):
if context.has_table(table):
context.operations.execute(f"""
UPDATE {table} SET type = 'generic' WHERE type IS NULL;
""")
context.operations.alter_column(table, 'type', nullable=False)
@upgrade_task('Add domain and segment to screens')
[docs]
def add_domain_and_segment_to_screens(context: UpgradeContext) -> None:
for column in ('domain', 'domain_segment'):
if not context.has_column('election_day_screens', column):
context.operations.add_column(
'election_day_screens',
Column(column, Text, nullable=True)
)
@upgrade_task('Add has results to archived results')
[docs]
def add_has_results_to_archived_results(context: UpgradeContext) -> None:
if not context.has_column('archived_results', 'has_results'):
context.operations.add_column(
'archived_results',
Column('has_results', Boolean, nullable=True)
)
@upgrade_task('Delete websocket notifications')
[docs]
def delete_websocket_notifications(context: UpgradeContext) -> None:
context.operations.execute("""
DELETE FROM notifications WHERE type = 'websocket';
""")
@upgrade_task('Make upload token none-nullable')
[docs]
def make_upload_take_none_nullable(context: UpgradeContext) -> None:
if context.has_column('upload_tokens', 'token'):
context.operations.alter_column(
'upload_tokens', 'token', nullable=False
)
@upgrade_task('Add domain and segment to subscribers')
[docs]
def add_domain_and_segment_to_subscribers(context: UpgradeContext) -> None:
for column in ('domain', 'domain_segment'):
if not context.has_column('subscribers', column):
context.operations.add_column(
'subscribers',
Column(column, Text, nullable=True)
)
@upgrade_task('Add short title')
[docs]
def add_short_title(context: UpgradeContext) -> None:
tables = ('elections', 'election_compounds', 'votes')
for table in tables:
if not context.has_column(table, 'short_title_translations'):
context.operations.add_column(
table,
Column('short_title_translations', HSTORE, nullable=True)
)