forked from ScoDoc/ScoDoc
383 lines
15 KiB
Python
383 lines
15 KiB
Python
# -*- coding: UTF-8 -*
|
|
##############################################################################
|
|
# ScoDoc
|
|
# Copyright (c) 1999 - 2024 Emmanuel Viennet. All rights reserved.
|
|
# See LICENSE
|
|
##############################################################################
|
|
|
|
"""ScoDoc models: Groups & partitions
|
|
"""
|
|
from operator import attrgetter
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from app import db, log
|
|
from app.models import ScoDocModel, GROUPNAME_STR_LEN, SHORT_STR_LEN
|
|
from app.models.etudiants import Identite
|
|
from app.models.events import Scolog
|
|
from app.scodoc import sco_cache
|
|
from app.scodoc import sco_utils as scu
|
|
from app.scodoc.sco_exceptions import AccessDenied, ScoValueError
|
|
|
|
|
|
class Partition(ScoDocModel):
|
|
"""Partition: découpage d'une promotion en groupes"""
|
|
|
|
__table_args__ = (db.UniqueConstraint("formsemestre_id", "partition_name"),)
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
partition_id = db.synonym("id")
|
|
formsemestre_id = db.Column(
|
|
db.Integer,
|
|
db.ForeignKey("notes_formsemestre.id"),
|
|
index=True,
|
|
)
|
|
# "TD", "TP", ... (NULL for 'all')
|
|
partition_name = db.Column(db.String(SHORT_STR_LEN))
|
|
# Numero = ordre de presentation)
|
|
numero = db.Column(db.Integer, nullable=False, default=0)
|
|
# Calculer le rang ?
|
|
bul_show_rank = db.Column(
|
|
db.Boolean(), nullable=False, default=False, server_default="false"
|
|
)
|
|
# Montrer quand on indique les groupes de l'étudiant ?
|
|
show_in_lists = db.Column(
|
|
db.Boolean(), nullable=False, default=True, server_default="true"
|
|
)
|
|
# Editable (créer/renommer groupes) ? (faux pour les groupes de parcours)
|
|
groups_editable = db.Column(
|
|
db.Boolean(), nullable=False, default=True, server_default="true"
|
|
)
|
|
groups = db.relationship(
|
|
"GroupDescr",
|
|
backref=db.backref("partition", lazy=True),
|
|
lazy="dynamic",
|
|
cascade="all, delete-orphan",
|
|
order_by="GroupDescr.numero, GroupDescr.group_name",
|
|
)
|
|
_sco_dept_relations = ("FormSemestre",)
|
|
|
|
def __init__(self, **kwargs):
|
|
super(Partition, self).__init__(**kwargs)
|
|
if self.numero is None:
|
|
# génère numero à la création
|
|
last_partition = Partition.query.order_by(Partition.numero.desc()).first()
|
|
if last_partition:
|
|
self.numero = last_partition.numero + 1
|
|
else:
|
|
self.numero = 1
|
|
|
|
def __repr__(self):
|
|
return f"""<{self.__class__.__name__} {self.id} "{self.partition_name or '(default)'}">"""
|
|
|
|
@classmethod
|
|
def check_name(
|
|
cls, formsemestre: "FormSemestre", partition_name: str, existing=False
|
|
) -> bool:
|
|
"""check if a partition named 'partition_name' can be created in the given formsemestre.
|
|
If existing is True, allow a partition_name already existing in the formsemestre.
|
|
"""
|
|
if not isinstance(partition_name, str):
|
|
return False
|
|
if not (0 < len(partition_name.strip()) < SHORT_STR_LEN):
|
|
return False
|
|
if (not existing) and (
|
|
partition_name in [p.partition_name for p in formsemestre.partitions]
|
|
):
|
|
return False
|
|
return True
|
|
|
|
@classmethod
|
|
def formsemestre_remove_etud(cls, formsemestre_id: int, etud: "Identite"):
|
|
"retire l'étudiant de toutes les partitions de ce semestre"
|
|
for group in GroupDescr.query.join(Partition).filter_by(
|
|
formsemestre_id=formsemestre_id
|
|
):
|
|
group.remove_etud(etud)
|
|
|
|
def is_default(self) -> bool:
|
|
"vrai si partition par défault (tous les étudiants)"
|
|
return not self.partition_name
|
|
|
|
def is_parcours(self) -> bool:
|
|
"Vrai s'il s'agit de la partition de parcours"
|
|
return self.partition_name == scu.PARTITION_PARCOURS
|
|
|
|
def to_dict(self, with_groups=False, str_keys: bool = False) -> dict:
|
|
"""as a dict, with or without groups.
|
|
If str_keys, convert integer dict keys to strings (useful for JSON)
|
|
"""
|
|
d = dict(self.__dict__)
|
|
d["partition_id"] = self.id
|
|
d.pop("_sa_instance_state", None)
|
|
d.pop("formsemestre", None)
|
|
|
|
if with_groups:
|
|
groups = sorted(self.groups, key=attrgetter("numero", "group_name"))
|
|
# un dict et non plus une liste, pour JSON
|
|
if str_keys:
|
|
d["groups"] = {
|
|
str(group.id): group.to_dict(with_partition=False)
|
|
for group in groups
|
|
}
|
|
else:
|
|
d["groups"] = {
|
|
group.id: group.to_dict(with_partition=False) for group in groups
|
|
}
|
|
return d
|
|
|
|
def get_etud_group(self, etudid: int) -> "GroupDescr":
|
|
"Le groupe de l'étudiant dans cette partition, ou None si pas présent"
|
|
return (
|
|
GroupDescr.query.filter_by(partition_id=self.id)
|
|
.join(group_membership)
|
|
.filter_by(etudid=etudid)
|
|
.first()
|
|
)
|
|
|
|
def set_etud_group(self, etud: "Identite", group: "GroupDescr") -> bool:
|
|
"""Affect etudid to group_id in given partition.
|
|
Raises IntegrityError si conflit,
|
|
or ValueError si ce group_id n'est pas dans cette partition
|
|
ou que l'étudiant n'est pas inscrit au semestre.
|
|
Return True si changement, False s'il était déjà dans ce groupe.
|
|
"""
|
|
if not group.id in (g.id for g in self.groups):
|
|
raise ScoValueError(
|
|
f"""Le groupe {group.id} n'est pas dans la partition {
|
|
self.partition_name or "tous"}"""
|
|
)
|
|
if etud.id not in (e.id for e in self.formsemestre.etuds):
|
|
raise ScoValueError(
|
|
f"""étudiant {etud.nomprenom} non inscrit au formsemestre du groupe {
|
|
group.group_name}"""
|
|
)
|
|
try:
|
|
existing_row = (
|
|
db.session.query(group_membership)
|
|
.filter_by(etudid=etud.id)
|
|
.join(GroupDescr)
|
|
.filter_by(partition_id=self.id)
|
|
.first()
|
|
)
|
|
if existing_row:
|
|
existing_group_id = existing_row[1]
|
|
if group.id == existing_group_id:
|
|
return False
|
|
# Fait le changement avec l'ORM sinon risque élevé de blocage
|
|
existing_group = db.session.get(GroupDescr, existing_group_id)
|
|
db.session.commit()
|
|
group.etuds.append(etud)
|
|
existing_group.etuds.remove(etud)
|
|
db.session.add(etud)
|
|
db.session.add(existing_group)
|
|
db.session.add(group)
|
|
else:
|
|
new_row = group_membership.insert().values(
|
|
etudid=etud.id, group_id=group.id
|
|
)
|
|
db.session.execute(new_row)
|
|
db.session.commit()
|
|
except IntegrityError:
|
|
db.session.rollback()
|
|
raise
|
|
return True
|
|
|
|
def create_group(self, group_name="", default=False) -> "GroupDescr":
|
|
"Crée un groupe dans cette partition"
|
|
if not self.formsemestre.can_change_groups():
|
|
raise AccessDenied(
|
|
"""Vous n'avez pas le droit d'effectuer cette opération,
|
|
ou bien le semestre est verrouillé !"""
|
|
)
|
|
if group_name:
|
|
group_name = group_name.strip()
|
|
if not group_name and not default:
|
|
raise ValueError("invalid group name: ()")
|
|
if not GroupDescr.check_name(self, group_name, default=default):
|
|
raise ScoValueError(
|
|
f"Le groupe {group_name} existe déjà dans cette partition"
|
|
)
|
|
numeros = [g.numero if g.numero is not None else 0 for g in self.groups]
|
|
if len(numeros) > 0:
|
|
new_numero = max(numeros) + 1
|
|
else:
|
|
new_numero = 0
|
|
group = GroupDescr(partition=self, group_name=group_name, numero=new_numero)
|
|
db.session.add(group)
|
|
db.session.commit()
|
|
log(f"create_group: created group_id={group.id}")
|
|
#
|
|
return group
|
|
|
|
|
|
class GroupDescr(ScoDocModel):
|
|
"""Description d'un groupe d'une partition"""
|
|
|
|
__tablename__ = "group_descr"
|
|
__table_args__ = (db.UniqueConstraint("partition_id", "group_name"),)
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
group_id = db.synonym("id")
|
|
partition_id = db.Column(db.Integer, db.ForeignKey("partition.id"))
|
|
group_name = db.Column(db.String(GROUPNAME_STR_LEN))
|
|
"""nom du groupe: "A", "C2", ... (NULL for 'all')"""
|
|
edt_id: str | None = db.Column(db.Text(), index=True, nullable=True)
|
|
"identifiant emplois du temps (unicité non imposée)"
|
|
numero = db.Column(db.Integer, nullable=False, default=0)
|
|
"Numero = ordre de presentation"
|
|
|
|
_sco_dept_relations = (
|
|
"Partition",
|
|
"FormSemestre",
|
|
)
|
|
|
|
etuds = db.relationship(
|
|
"Identite",
|
|
secondary="group_membership",
|
|
lazy="dynamic",
|
|
)
|
|
|
|
def __repr__(self):
|
|
return (
|
|
f"""<{self.__class__.__name__} {self.id} "{self.group_name or '(tous)'}">"""
|
|
)
|
|
|
|
@classmethod
|
|
def filter_model_attributes(cls, data: dict, excluded: set[str] = None) -> dict:
|
|
"""Returns a copy of dict with only the keys belonging to the Model and not in excluded.
|
|
Exclude `partition_id` : a group cannot be moved from a partition to another.
|
|
"""
|
|
return super().filter_model_attributes(
|
|
data,
|
|
excluded=(excluded or set()) | {"partition_id"},
|
|
)
|
|
|
|
def get_nom_with_part(self, default="-") -> str:
|
|
"""Nom avec partition: 'TD A'
|
|
Si groupe par défaut (tous), utilise default ou "-"
|
|
"""
|
|
if self.partition.partition_name is None:
|
|
return default
|
|
return f"{self.partition.partition_name or ''} {self.group_name or '-'}"
|
|
|
|
def to_dict(self, with_partition=True) -> dict:
|
|
"""as a dict, with or without partition"""
|
|
if with_partition:
|
|
partition_dict = self.partition.to_dict(with_groups=False)
|
|
d = dict(self.__dict__)
|
|
d.pop("_sa_instance_state", None)
|
|
if with_partition:
|
|
d["partition"] = partition_dict
|
|
return d
|
|
|
|
def get_edt_ids(self) -> list[str]:
|
|
"les ids normalisés pour l'emploi du temps: à défaut, le nom scodoc du groupe"
|
|
return [
|
|
scu.normalize_edt_id(x)
|
|
for x in scu.split_id(self.edt_id) or [self.group_name] or []
|
|
]
|
|
|
|
def get_nb_inscrits(self) -> int:
|
|
"""Nombre inscrits à ce group et au formsemestre.
|
|
C'est nécessaire car lors d'une désinscription, on conserve l'appartenance
|
|
aux groupes pour faciliter une éventuelle ré-inscription.
|
|
"""
|
|
from app.models.formsemestre import FormSemestreInscription
|
|
|
|
return (
|
|
Identite.query.join(group_membership)
|
|
.filter_by(group_id=self.id)
|
|
.join(FormSemestreInscription)
|
|
.filter_by(formsemestre_id=self.partition.formsemestre.id)
|
|
.count()
|
|
)
|
|
|
|
@classmethod
|
|
def check_name(
|
|
cls, partition: "Partition", group_name: str, existing=False, default=False
|
|
) -> bool:
|
|
"""check if a group named 'group_name' can be created in the given partition.
|
|
If existing is True, allow a group_name already existing in the partition.
|
|
If default, name must be empty and default group must not yet exists.
|
|
"""
|
|
if not isinstance(group_name, str):
|
|
return False
|
|
if not default and not (0 < len(group_name.strip()) < GROUPNAME_STR_LEN):
|
|
return False
|
|
if (not existing) and (group_name in [g.group_name for g in partition.groups]):
|
|
return False
|
|
return True
|
|
|
|
def set_name(self, group_name: str, dest_url: str = None):
|
|
"""Set group name, and optionally edt_id.
|
|
Check permission (partition must be groups_editable)
|
|
and invalidate caches. Commit session.
|
|
dest_url is used for error messages.
|
|
"""
|
|
if not self.partition.formsemestre.can_change_groups():
|
|
raise AccessDenied("Vous n'avez pas le droit d'effectuer cette opération !")
|
|
if self.group_name is None:
|
|
raise ValueError("can't set a name to default group")
|
|
if not self.partition.groups_editable:
|
|
raise AccessDenied("Partition non éditable")
|
|
if group_name:
|
|
group_name = group_name.strip()
|
|
if not group_name:
|
|
raise ScoValueError("nom de groupe vide !", dest_url=dest_url)
|
|
if group_name != self.group_name and not GroupDescr.check_name(
|
|
self.partition, group_name
|
|
):
|
|
raise ScoValueError(
|
|
"Le nom de groupe existe déjà dans la partition", dest_url=dest_url
|
|
)
|
|
|
|
self.group_name = group_name
|
|
db.session.add(self)
|
|
db.session.commit()
|
|
sco_cache.invalidate_formsemestre(
|
|
formsemestre_id=self.partition.formsemestre_id
|
|
)
|
|
|
|
def set_edt_id(self, edt_id: str):
|
|
"Set edt_id. Check permission. Commit session."
|
|
if not self.partition.formsemestre.can_change_groups():
|
|
raise AccessDenied("Vous n'avez pas le droit d'effectuer cette opération !")
|
|
if isinstance(edt_id, str):
|
|
edt_id = edt_id.strip() or None
|
|
self.edt_id = edt_id
|
|
db.session.add(self)
|
|
db.session.commit()
|
|
|
|
def remove_etud(self, etud: "Identite"):
|
|
"Enlève l'étudiant de ce groupe s'il en fait partie (ne fait rien sinon)"
|
|
if etud in self.etuds:
|
|
self.etuds.remove(etud)
|
|
Scolog.logdb(
|
|
method="group_remove_etud",
|
|
etudid=etud.id,
|
|
msg=f"Retrait du groupe {self.group_name} de {self.partition.partition_name}",
|
|
)
|
|
db.session.commit()
|
|
# Update parcours
|
|
if self.partition.partition_name == scu.PARTITION_PARCOURS:
|
|
self.partition.formsemestre.update_inscriptions_parcours_from_groups(
|
|
etudid=etud.id
|
|
)
|
|
sco_cache.invalidate_formsemestre(self.partition.formsemestre_id)
|
|
|
|
|
|
group_membership = db.Table(
|
|
"group_membership",
|
|
db.Column("etudid", db.Integer, db.ForeignKey("identite.id", ondelete="CASCADE")),
|
|
db.Column("group_id", db.Integer, db.ForeignKey("group_descr.id")),
|
|
db.UniqueConstraint("etudid", "group_id"),
|
|
)
|
|
# class GroupMembership(db.Model):
|
|
# """Association groupe / étudiant"""
|
|
|
|
# __tablename__ = "group_membership"
|
|
# __table_args__ = (db.UniqueConstraint("etudid", "group_id"),)
|
|
# id = db.Column(db.Integer, primary_key=True)
|
|
# etudid = db.Column(db.Integer, db.ForeignKey("identite.id", ondelete="CASCADE"))
|
|
# group_id = db.Column(db.Integer, db.ForeignKey("group_descr.id"))
|