Mercurial > libervia-backend
view sat/memory/migration/versions/602caf848068_drop_message_types_table_fix_nullable.py @ 3893:045af0eeda3f
core, CLI (base), tools (common/data_format): typing/core readability
author | Goffi <goffi@goffi.org> |
---|---|
date | Wed, 21 Sep 2022 22:20:43 +0200 |
parents | 16ade4ad63f3 |
children |
line wrap: on
line source
"""drop message_types table + fix nullable + rename constraints Revision ID: 602caf848068 Revises: Create Date: 2021-06-26 12:42:54.148313 """ from alembic import op from sqlalchemy import ( Table, Column, MetaData, TEXT, INTEGER, Text, Integer, Float, Enum, ForeignKey, Index, PrimaryKeyConstraint, ) from sqlalchemy.sql import table, column # revision identifiers, used by Alembic. revision = "602caf848068" down_revision = None branch_labels = None depends_on = None def upgrade(): # we have to recreate former tables for batch_alter_table's reflexion, otherwise the # database will be used, and this will keep unammed UNIQUE constraints in addition # to the named ones that we create metadata = MetaData( naming_convention={ "ix": "ix_%(column_0_label)s", "uq": "uq_%(table_name)s_%(column_0_name)s", "ck": "ck_%(table_name)s_%(constraint_name)s", "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s", "pk": "pk_%(table_name)s", }, ) old_profiles_table = Table( "profiles", metadata, Column("id", Integer, primary_key=True, nullable=True, autoincrement=False), Column("name", Text, unique=True), ) old_components_table = Table( "components", metadata, Column( "profile_id", ForeignKey("profiles.id", ondelete="CASCADE"), nullable=True, primary_key=True, ), Column("entry_point", Text, nullable=False), ) old_message_table = Table( "message", metadata, Column("id", Integer, primary_key=True, nullable=True, autoincrement=False), Column("history_uid", ForeignKey("history.uid", ondelete="CASCADE")), Column("message", Text), Column("language", Text), Index("message__history_uid", "history_uid"), ) old_subject_table = Table( "subject", metadata, Column("id", Integer, primary_key=True, nullable=True, autoincrement=False), Column("history_uid", ForeignKey("history.uid", ondelete="CASCADE")), Column("subject", Text), Column("language", Text), Index("subject__history_uid", "history_uid"), ) old_thread_table = Table( "thread", metadata, Column("id", Integer, primary_key=True, nullable=True, autoincrement=False), Column("history_uid", ForeignKey("history.uid", ondelete="CASCADE")), Column("thread_id", Text), Column("parent_id", Text), Index("thread__history_uid", "history_uid"), ) old_history_table = Table( "history", metadata, Column("uid", Text, primary_key=True, nullable=True), Column("stanza_id", Text), Column("update_uid", Text), Column("profile_id", Integer, ForeignKey("profiles.id", ondelete="CASCADE")), Column("source", Text), Column("dest", Text), Column("source_res", Text), Column("dest_res", Text), Column("timestamp", Float, nullable=False), Column("received_timestamp", Float), Column("type", Text, ForeignKey("message_types.type")), Column("extra", Text), Index("history__profile_id_timestamp", "profile_id", "timestamp"), Index( "history__profile_id_received_timestamp", "profile_id", "received_timestamp" ), ) old_param_gen_table = Table( "param_gen", metadata, Column("category", Text, primary_key=True), Column("name", Text, primary_key=True), Column("value", Text), ) old_param_ind_table = Table( "param_ind", metadata, Column("category", Text, primary_key=True), Column("name", Text, primary_key=True), Column( "profile_id", ForeignKey("profiles.id", ondelete="CASCADE"), primary_key=True ), Column("value", Text), ) old_private_gen_table = Table( "private_gen", metadata, Column("namespace", Text, primary_key=True), Column("key", Text, primary_key=True), Column("value", Text), ) old_private_ind_table = Table( "private_ind", metadata, Column("namespace", Text, primary_key=True), Column("key", Text, primary_key=True), Column( "profile_id", ForeignKey("profiles.id", ondelete="CASCADE"), primary_key=True ), Column("value", Text), ) old_private_gen_bin_table = Table( "private_gen_bin", metadata, Column("namespace", Text, primary_key=True), Column("key", Text, primary_key=True), Column("value", Text), ) old_private_ind_bin_table = Table( "private_ind_bin", metadata, Column("namespace", Text, primary_key=True), Column("key", Text, primary_key=True), Column( "profile_id", ForeignKey("profiles.id", ondelete="CASCADE"), primary_key=True ), Column("value", Text), ) old_files_table = Table( "files", metadata, Column("id", Text, primary_key=True), Column("public_id", Text, unique=True), Column("version", Text, primary_key=True), Column("parent", Text, nullable=False), Column( "type", Enum("file", "directory", name="file_type", create_constraint=True), nullable=False, server_default="file", ), Column("file_hash", Text), Column("hash_algo", Text), Column("name", Text, nullable=False), Column("size", Integer), Column("namespace", Text), Column("media_type", Text), Column("media_subtype", Text), Column("created", Float, nullable=False), Column("modified", Float), Column("owner", Text), Column("access", Text), Column("extra", Text), Column("profile_id", ForeignKey("profiles.id", ondelete="CASCADE")), Index("files__profile_id_owner_parent", "profile_id", "owner", "parent"), Index( "files__profile_id_owner_media_type_media_subtype", "profile_id", "owner", "media_type", "media_subtype", ), ) op.drop_table("message_types") with op.batch_alter_table( "profiles", copy_from=old_profiles_table, schema=None ) as batch_op: batch_op.create_unique_constraint(batch_op.f("uq_profiles_name"), ["name"]) with op.batch_alter_table( "components", copy_from=old_components_table, naming_convention={ "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s", }, schema=None, ) as batch_op: batch_op.create_unique_constraint(batch_op.f("uq_profiles_name"), ["name"]) with op.batch_alter_table( "history", copy_from=old_history_table, naming_convention={ "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s", }, schema=None, ) as batch_op: batch_op.alter_column("uid", existing_type=TEXT(), nullable=False) batch_op.alter_column( "type", type_=Enum( "chat", "error", "groupchat", "headline", "normal", "info", name="message_type", create_constraint=True, ), existing_type=TEXT(), nullable=False, ) batch_op.create_unique_constraint( batch_op.f("uq_history_profile_id"), ["profile_id", "stanza_id", "source", "dest"], ) batch_op.drop_constraint("fk_history_type_message_types", type_="foreignkey") with op.batch_alter_table( "message", copy_from=old_message_table, schema=None ) as batch_op: batch_op.alter_column( "id", existing_type=INTEGER(), nullable=False, autoincrement=False ) with op.batch_alter_table( "subject", copy_from=old_subject_table, schema=None ) as batch_op: batch_op.alter_column( "id", existing_type=INTEGER(), nullable=False, autoincrement=False ) with op.batch_alter_table( "thread", copy_from=old_thread_table, schema=None ) as batch_op: batch_op.alter_column( "id", existing_type=INTEGER(), nullable=False, autoincrement=False ) with op.batch_alter_table( "param_gen", copy_from=old_param_gen_table, schema=None ) as batch_op: batch_op.alter_column("category", existing_type=TEXT(), nullable=False) batch_op.alter_column("name", existing_type=TEXT(), nullable=False) with op.batch_alter_table( "param_ind", copy_from=old_param_ind_table, schema=None ) as batch_op: batch_op.alter_column("category", existing_type=TEXT(), nullable=False) batch_op.alter_column("name", existing_type=TEXT(), nullable=False) batch_op.alter_column("profile_id", existing_type=INTEGER(), nullable=False) with op.batch_alter_table( "private_gen", copy_from=old_private_gen_table, schema=None ) as batch_op: batch_op.alter_column("namespace", existing_type=TEXT(), nullable=False) batch_op.alter_column("key", existing_type=TEXT(), nullable=False) with op.batch_alter_table( "private_ind", copy_from=old_private_ind_table, schema=None ) as batch_op: batch_op.alter_column("namespace", existing_type=TEXT(), nullable=False) batch_op.alter_column("key", existing_type=TEXT(), nullable=False) batch_op.alter_column("profile_id", existing_type=INTEGER(), nullable=False) with op.batch_alter_table( "private_gen_bin", copy_from=old_private_gen_bin_table, schema=None ) as batch_op: batch_op.alter_column("namespace", existing_type=TEXT(), nullable=False) batch_op.alter_column("key", existing_type=TEXT(), nullable=False) # found some invalid rows in local database, maybe old values made during development, # but in doubt we have to delete them op.execute("DELETE FROM private_ind_bin WHERE namespace IS NULL") with op.batch_alter_table( "private_ind_bin", copy_from=old_private_ind_bin_table, schema=None ) as batch_op: batch_op.alter_column("namespace", existing_type=TEXT(), nullable=False) batch_op.alter_column("key", existing_type=TEXT(), nullable=False) batch_op.alter_column("profile_id", existing_type=INTEGER(), nullable=False) with op.batch_alter_table( "files", copy_from=old_files_table, schema=None ) as batch_op: batch_op.create_unique_constraint(batch_op.f("uq_files_public_id"), ["public_id"]) batch_op.alter_column( "type", type_=Enum("file", "directory", name="file_type", create_constraint=True), existing_type=Text(), nullable=False, ) def downgrade(): # downgrade doesn't restore the exact same state as before upgrade, as it # would be useless and waste of resource to restore broken things such as # anonymous constraints with op.batch_alter_table("thread", schema=None) as batch_op: batch_op.alter_column( "id", existing_type=INTEGER(), nullable=True, autoincrement=False ) with op.batch_alter_table("subject", schema=None) as batch_op: batch_op.alter_column( "id", existing_type=INTEGER(), nullable=True, autoincrement=False ) with op.batch_alter_table("private_ind_bin", schema=None) as batch_op: batch_op.alter_column("profile_id", existing_type=INTEGER(), nullable=True) batch_op.alter_column("key", existing_type=TEXT(), nullable=True) batch_op.alter_column("namespace", existing_type=TEXT(), nullable=True) with op.batch_alter_table("private_ind", schema=None) as batch_op: batch_op.alter_column("profile_id", existing_type=INTEGER(), nullable=True) batch_op.alter_column("key", existing_type=TEXT(), nullable=True) batch_op.alter_column("namespace", existing_type=TEXT(), nullable=True) with op.batch_alter_table("private_gen_bin", schema=None) as batch_op: batch_op.alter_column("key", existing_type=TEXT(), nullable=True) batch_op.alter_column("namespace", existing_type=TEXT(), nullable=True) with op.batch_alter_table("private_gen", schema=None) as batch_op: batch_op.alter_column("key", existing_type=TEXT(), nullable=True) batch_op.alter_column("namespace", existing_type=TEXT(), nullable=True) with op.batch_alter_table("param_ind", schema=None) as batch_op: batch_op.alter_column("profile_id", existing_type=INTEGER(), nullable=True) batch_op.alter_column("name", existing_type=TEXT(), nullable=True) batch_op.alter_column("category", existing_type=TEXT(), nullable=True) with op.batch_alter_table("param_gen", schema=None) as batch_op: batch_op.alter_column("name", existing_type=TEXT(), nullable=True) batch_op.alter_column("category", existing_type=TEXT(), nullable=True) with op.batch_alter_table("message", schema=None) as batch_op: batch_op.alter_column( "id", existing_type=INTEGER(), nullable=True, autoincrement=False ) op.create_table( "message_types", Column("type", TEXT(), nullable=True), PrimaryKeyConstraint("type"), ) message_types_table = table("message_types", column("type", TEXT())) op.bulk_insert( message_types_table, [ {"type": "chat"}, {"type": "error"}, {"type": "groupchat"}, {"type": "headline"}, {"type": "normal"}, {"type": "info"}, ], ) with op.batch_alter_table("history", schema=None) as batch_op: batch_op.alter_column( "type", type_=TEXT(), existing_type=TEXT(), nullable=True, ) batch_op.create_foreign_key( batch_op.f("fk_history_type_message_types"), "message_types", ["type"], ["type"], ) batch_op.alter_column("uid", existing_type=TEXT(), nullable=True)