ScoDoc/migrations/versions/ae9bb0feea7a_contraintes_identite.py

98 lines
3.2 KiB
Python
Raw Normal View History

"""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(
"""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(
"""SELECT id, nom, prenom FROM identite
WHERE code_nip=:code_nip""",
{"code_nip": code_nip},
).all()
for i, (etudid, nom, prenom) in enumerate(etuds_dups[1:], start=1):
session.execute(
"""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(
"""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 code_ine=:code_ine""",
{"code_ine": code_ine},
).all()
for i, (etudid, nom, prenom) in enumerate(etuds_dups[1:], start=1):
session.execute(
"""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 ###