1
0
forked from ScoDoc/ScoDoc
ScoDoc/migrations/versions/497ba81343f7_identite_admission.py

220 lines
7.4 KiB
Python

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