MonScoDocEssai/migrations/versions/497ba81343f7_identite_admission.py

"""identite_admission

Revision ID: 497ba81343f7
Revises: 5c44d0d215ca
Create Date: 2023-10-14 10:09:02.330634

Diverses amlioration du modèle Identite:
- boursier non null
- departement non null
- admission : 1 seule admission (one-to-one)
- adresse: etudid non null.
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.orm import sessionmaker  # added by ev


# revision identifiers, used by Alembic.
revision = "497ba81343f7"
down_revision = "5c44d0d215ca"
branch_labels = None
depends_on = None

Session = sessionmaker()


def upgrade():
    bind = op.get_bind()
    session = Session(bind=bind)
    # Enleve les éventuels nulls de boursier
    session.execute(
        sa.text(
            """
            UPDATE identite SET boursier = false WHERE boursier IS NULL;
            """
        )
    )
    # Enleve les éventuelles adresses orphelines:
    session.execute(
        sa.text(
            """
            DELETE FROM adresse WHERE etudid IS NULL;
            """
        )
    )
    # Affecte arbitrairement les éventuels étudiants sans département au 1er
    # (il ne devrait pas y en avoir, sauf essais manuels ou bugs)
    # nb_etuds_sans_dept = session.execute(
    #     sa.text("""select count(*) from identite where dept_id is NULL;""")
    # )
    # if nb_etuds_sans_dept:
    #     session.execute(
    #         sa.text(
    #             """
    #             INSERT INTO departement (acronym, description, visible)
    #             VALUES ('SANS_DEPT', 'interne: rattache objets sans departement', false);
    #             UPDATE identite SET code_ine = NULL WHERE dept_id IS NULL;
    #             UPDATE identite SET code_nip = NULL WHERE dept_id IS NULL;
    #             UPDATE identite SET dept_id = (
    #                 SELECT id from departement where acronym  = 'SANS_DEPT'
    #             ) WHERE dept_id IS NULL;
    #             """
    #         )
    #     )
    session.execute(
        sa.text(
            """
            DELETE FROM identite WHERE dept_id IS NULL;
            """
        )
    )

    with op.batch_alter_table("identite", schema=None) as batch_op:
        batch_op.add_column(sa.Column("admission_id", sa.Integer(), nullable=True))
        batch_op.alter_column("boursier", existing_type=sa.BOOLEAN(), nullable=False)
        batch_op.alter_column("dept_id", existing_type=sa.Integer(), nullable=False)
        batch_op.create_foreign_key(
            "admissions_etudid_fkey", "admissions", ["admission_id"], ["id"]
        )
        batch_op.drop_constraint("identite_dept_id_code_ine_key", type_="unique")
        batch_op.drop_constraint("identite_dept_id_code_nip_key", type_="unique")
        batch_op.create_index(
            "unique_dept_ine_except_null",
            ["dept_id", "code_ine"],
            unique=True,
            postgresql_where=sa.text("code_ine IS NOT NULL"),
        )
        batch_op.create_index(
            "unique_dept_nip_except_null",
            ["dept_id", "code_nip"],
            unique=True,
            postgresql_where=sa.text("code_nip IS NOT NULL"),
        )
        batch_op.alter_column(
            "civilite_etat_civil",
            existing_type=sa.VARCHAR(length=1),
            nullable=True,
            existing_server_default=sa.text("'X'::character varying"),
        )
        batch_op.alter_column(
            "prenom_etat_civil",
            existing_type=sa.TEXT(),
            nullable=True,
            existing_server_default=sa.text("''::text"),
        )

    with op.batch_alter_table("adresse", schema=None) as batch_op:
        batch_op.alter_column("etudid", existing_type=sa.Integer(), nullable=False)
        batch_op.drop_constraint("adresse_etudid_fkey", type_="foreignkey")
        batch_op.create_foreign_key(
            "adresse_etudid_fkey", "identite", ["etudid"], ["id"]
        )

    # Elimine eventuels duplicats dans Admission
    session.execute(
        sa.text(
            """
            DELETE FROM admissions
            WHERE id NOT IN (
                SELECT MIN(id)
                FROM admissions
                GROUP BY etudid
            );
            """
        )
    )
    # Copie id
    session.execute(
        sa.text(
            """
            UPDATE identite SET admission_id = admissions.id
            FROM admissions WHERE admissions.etudid = identite.id;
            """
        )
    )

    with op.batch_alter_table("admissions", schema=None) as batch_op:
        batch_op.drop_constraint("admissions_etudid_fkey", type_="foreignkey")
        batch_op.alter_column(
            "physique",
            existing_type=sa.DOUBLE_PRECISION(precision=53),
            type_=sa.Text(),
            existing_nullable=True,
        )
        batch_op.alter_column(
            "anglais",
            existing_type=sa.DOUBLE_PRECISION(precision=53),
            type_=sa.Text(),
            existing_nullable=True,
        )
        batch_op.alter_column(
            "francais",
            existing_type=sa.DOUBLE_PRECISION(precision=53),
            type_=sa.Text(),
            existing_nullable=True,
        )
        # laisse l'ancienne colonne pour downgrade (tests)
        # batch_op.drop_column('etudid')


def downgrade():
    with op.batch_alter_table("identite", schema=None) as batch_op:
        batch_op.drop_constraint("admissions_etudid_fkey", type_="foreignkey")
        batch_op.alter_column("boursier", existing_type=sa.BOOLEAN(), nullable=True)
        batch_op.alter_column("dept_id", existing_type=sa.Integer(), nullable=True)
        batch_op.drop_column("admission_id")
        batch_op.drop_index(
            "unique_dept_nip_except_null",
            postgresql_where=sa.text("code_nip IS NOT NULL"),
        )
        batch_op.drop_index(
            "unique_dept_ine_except_null",
            postgresql_where=sa.text("code_ine IS NOT NULL"),
        )
        batch_op.alter_column(
            "prenom_etat_civil",
            existing_type=sa.TEXT(),
            nullable=False,
            existing_server_default=sa.text("''::text"),
        )
        batch_op.alter_column(
            "civilite_etat_civil",
            existing_type=sa.VARCHAR(length=1),
            nullable=False,
            existing_server_default=sa.text("'X'::character varying"),
        )

    with op.batch_alter_table("admissions", schema=None) as batch_op:
        # batch_op.add_column(
        #    sa.Column("etudid", sa.INTEGER(), autoincrement=False, nullable=True)
        # )
        batch_op.create_foreign_key(
            "admissions_etudid_fkey", "identite", ["etudid"], ["id"], ondelete="CASCADE"
        )
        batch_op.alter_column(
            "francais",
            existing_type=sa.Text(),
            type_=sa.DOUBLE_PRECISION(precision=53),
            existing_nullable=True,
        )
        batch_op.alter_column(
            "anglais",
            existing_type=sa.Text(),
            type_=sa.DOUBLE_PRECISION(precision=53),
            existing_nullable=True,
        )
        batch_op.alter_column(
            "physique",
            existing_type=sa.Text(),
            type_=sa.DOUBLE_PRECISION(precision=53),
            existing_nullable=True,
        )

    with op.batch_alter_table("adresse", schema=None) as batch_op:
        batch_op.drop_constraint("adresse_etudid_fkey", type_="foreignkey")
        batch_op.create_foreign_key(
            "adresse_etudid_fkey", "identite", ["etudid"], ["id"], ondelete="CASCADE"
        )
        batch_op.alter_column("etudid", existing_type=sa.INTEGER(), nullable=True)