ScoDoc-Lille/migrations/versions/829683efddc4_change_apcvalidationannee.py

111 lines
3.5 KiB
Python

"""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))