ScoDoc/migrations/versions/ae9bb0feea7a_contraintes_identite.py

"""contraintes identite

Revision ID: ae9bb0feea7a
Revises: 5731e904baac
Create Date: 2023-03-12 19:00:58.544873

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


# revision identifiers, used by Alembic.
revision = "ae9bb0feea7a"
down_revision = "5731e904baac"
branch_labels = None
depends_on = None

Session = sessionmaker()


def upgrade():
    # On répare une erreur (typo) dans la déclaration de la table Identite
    # qui faisait que les contraintes d'unicité des couples (ine, dept)
    # et (nip, dept) n'était pas prises en compte.
    # On commence par chercher les éventuels (rares) doublons, changer leurs codes
    # avant de créer les contraintes.
    bind = op.get_bind()
    session = Session(bind=bind)
    # Corrige NIP
    dups = session.execute(
        sa.text(
            """SELECT dept_id, code_nip
            FROM identite
            WHERE code_nip IS NOT NULL
            GROUP BY dept_id, code_nip
            HAVING COUNT(*) > 1;"""
        )
    ).all()
    for dept_id, code_nip in dups:
        etuds_dups = session.execute(
            sa.text(
                """SELECT id, nom, prenom FROM identite
            WHERE dept_id=:dept_id AND code_nip=:code_nip"""
            ),
            {"dept_id": dept_id, "code_nip": code_nip},
        ).all()
        for i, (etudid, nom, prenom) in enumerate(etuds_dups[1:], start=1):
            session.execute(
                sa.text("""UPDATE identite SET code_nip=:code_nip WHERE id=:etudid"""),
                {
                    "code_nip": f"{code_nip}-{i}",
                    "etudid": etudid,
                },
            )
            print(
                f"Warning: duplication de code NIP détectée: vérifier {nom} {prenom} NIP={code_nip}"
            )
    session.commit()
    # Corrige INE
    dups = session.execute(
        sa.text(
            """SELECT dept_id, code_ine
            FROM identite
            WHERE code_ine IS NOT NULL
            GROUP BY dept_id, code_ine
            HAVING COUNT(*) > 1;"""
        )
    ).all()
    for dept_id, code_ine in dups:
        etuds_dups = session.execute(
            """SELECT id, nom, prenom FROM identite
            WHERE dept_id=:dept_id AND code_ine=:code_ine""",
            {"dept_id": dept_id, "code_ine": code_ine},
        ).all()
        for i, (etudid, nom, prenom) in enumerate(etuds_dups[1:], start=1):
            session.execute(
                sa.text("""UPDATE identite SET code_ine=:code_ine WHERE id=:etudid"""),
                {
                    "code_ine": f"{code_ine}-{i}",
                    "etudid": etudid,
                },
            )
            print(
                f"Warning: duplication de code INE détectée: vérifier {nom} {prenom} NIP={code_ine}"
            )
    session.commit()

    # CREATION DES CONTRAINTES
    op.create_unique_constraint(
        "identite_dept_id_code_nip_key", "identite", ["dept_id", "code_nip"]
    )
    op.create_unique_constraint(
        "identite_dept_id_code_ine_key", "identite", ["dept_id", "code_ine"]
    )
    # ### end Alembic commands ###


def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_constraint("identite_dept_id_code_nip_key", "identite", type_="unique")
    op.drop_constraint("identite_dept_id_code_ine_key", "identite", type_="unique")
    # ### end Alembic commands ###