MonScoDocEssai/migrations/versions/829683efddc4_change_apcvalidationannee.py

"""Change ApcValidationAnnee

Revision ID: 829683efddc4
Revises: c701224fa255
Create Date: 2023-06-28 09:47:16.591028

"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.orm import sessionmaker  # added by ev

# revision identifiers, used by Alembic.
revision = "829683efddc4"
down_revision = "c701224fa255"
branch_labels = None
depends_on = None

Session = sessionmaker()


# Voir https://stackoverflow.com/questions/24082542/check-if-a-table-column-exists-in-the-database-using-sqlalchemy-and-alembic
from sqlalchemy import inspect


def column_exists(table_name, column_name):
    bind = op.get_context().bind
    insp = inspect(bind)
    columns = insp.get_columns(table_name)
    return any(c["name"] == column_name for c in columns)


def upgrade():
    if column_exists("apc_validation_annee", "referentiel_competence_id"):
        return  # utile durant developpement
    # Enleve la contrainte erronée
    with op.batch_alter_table("apc_validation_annee", schema=None) as batch_op:
        batch_op.drop_constraint(
            "apc_validation_annee_etudid_annee_scolaire_ordre_key", type_="unique"
        )
        # Ajoute colonne referentiel, nullable pour l'instant
        batch_op.add_column(
            sa.Column("referentiel_competence_id", sa.Integer(), nullable=True)
        )

    # Affecte le referentiel des anciennes validations
    bind = op.get_bind()
    session = Session(bind=bind)
    session.execute(
        sa.text(
            """
        UPDATE apc_validation_annee AS a
        SET referentiel_competence_id = (
            SELECT f.referentiel_competence_id
            FROM notes_formations f
            WHERE f.id = a.formation_id
        )
        """
        )
    )
    # En principe, on n'a pas pu entrer de validation sur des formations sans referentiel
    # par prudence, on les supprime avant d'ajouter la contrainte
    session.execute(
        sa.text(
            "DELETE FROM apc_validation_annee WHERE referentiel_competence_id is NULL"
        )
    )
    op.alter_column(
        "apc_validation_annee",
        "referentiel_competence_id",
        nullable=False,
    )
    op.create_foreign_key(
        "apc_validation_annee_refcomp_fkey",
        "apc_validation_annee",
        "apc_referentiel_competences",
        ["referentiel_competence_id"],
        ["id"],
    )
    # Efface les validations d'année dupliquées
    # (garde la validation la plus récente)
    session.execute(
        sa.text(
            """
        DELETE FROM apc_validation_annee t1
        WHERE t1.id <> (SELECT max(t2.id)
            FROM apc_validation_annee t2
            WHERE t1.etudid = t2.etudid
            AND t1.referentiel_competence_id = t2.referentiel_competence_id
            AND t1.ordre = t2.ordre
        )
        """
        )
    )
    # Et ajoute la contrainte unicité de décision année par étudiant/ref. comp.:
    op.create_unique_constraint(
        "apc_validation_annee_etudid_ordre_refcomp_key",
        "apc_validation_annee",
        ["etudid", "ordre", "referentiel_competence_id"],
    )
    op.drop_column("apc_validation_annee", "formation_id")


def downgrade():
    # Se contente de ré-ajouter la colonne formation_id sans re-générer son contenu
    with op.batch_alter_table("apc_validation_annee", schema=None) as batch_op:
        # batch_op.drop_constraint(
        #     "apc_validation_annee_etudid_ordre_refcomp_key", type_="unique"
        # )
        # batch_op.drop_column("referentiel_competence_id")
        batch_op.add_column(sa.Column("formation_id", sa.Integer(), nullable=True))