Pointage enregistré et statistiques de présence
ci-cd / build-backend (push) Successful in 29s
ci-cd / build-frontend (push) Successful in 19s
ci-cd / deploy (push) Successful in 12s

Étape 58 : le présent / absent du référent est enregistré sur
l'affectation (Assignment.attendance + heure, migration 0030,
PUT .../assignments/{id}/attendance) au lieu de disparaître au
rechargement.

Étape 59 : section « Présence » de l'écran Statistiques — créneaux en
cours et créneaux terminés, par poste, rafraîchie toutes les 30 s.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
AdamThomas
2026-10-28 08:06:32 +01:00
co-authored by Claude Opus 5.5
parent a37f6c2c41
commit c57c70569c
19 changed files with 600 additions and 22 deletions
+26 -1
View File
@@ -17,6 +17,7 @@ from app.api.schemas import (
AssignmentValidateRequest,
AssignmentValidateResult,
AssignmentWriteResult,
AttendanceUpdate,
)
from app.auth.deps import (
current_account_in_edition,
@@ -24,7 +25,7 @@ from app.auth.deps import (
require_staff,
)
from app.db import get_session
from app.models import Account, Assignment, Edition, Role, Shift, Volunteer
from app.models import Account, Assignment, Edition, Role, Shift, Volunteer, local_now
from app.validation import has_blocking, validate
from app.validation.loader import build_validation_input
@@ -206,6 +207,30 @@ def update_assignment(
return AssignmentWriteResult(assignment=assignment, warnings=warnings)
@router.put("/{assignment_id}/attendance", response_model=AssignmentRead)
def set_attendance(
assignment_id: uuid.UUID,
payload: AttendanceUpdate,
edition: Edition = Depends(get_edition),
account: Account = Depends(current_account_in_edition),
session: Session = Depends(get_session),
) -> Assignment:
"""Pointe un bénévole présent ou absent sur son créneau (`None` annule) —
le référent du poste ou la coordination. Enregistré en base : le
pointage survit à un rechargement et se retrouve sur tout appareil."""
assignment = owned_or_404(
session.get(Assignment, assignment_id), edition.id
)
ensure_can_write_shift(
account, session, session.get(Shift, assignment.shift_id)
)
assignment.attendance = payload.attendance
assignment.attendance_marked_at = local_now() if payload.attendance else None
session.commit()
session.refresh(assignment)
return assignment
@router.delete("/{assignment_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_assignment(
assignment_id: uuid.UUID,
+40
View File
@@ -7,6 +7,7 @@ from pydantic import BaseModel, ConfigDict, Field
from app.models import (
AssignmentSource,
AttendanceStatus,
BadgeAlign,
BadgeFace,
BadgeFieldType,
@@ -305,10 +306,17 @@ class AssignmentRead(ApiModel):
shift_id: uuid.UUID
source: AssignmentSource
locked: bool
attendance: AttendanceStatus | None
attendance_marked_at: dt.datetime | None
created_at: dt.datetime
updated_at: dt.datetime
class AttendanceUpdate(ApiModel):
# `None` = annuler le pointage.
attendance: AttendanceStatus | None
class AssignmentWriteResult(ApiModel):
assignment: AssignmentRead
warnings: list[Violation] = Field(default_factory=list)
@@ -758,9 +766,41 @@ class StatsNeeds(ApiModel):
volunteers_with_assignment: int
class StatsAttendanceCounts(ApiModel):
"""Pointage (`Assignment.attendance`) d'un ensemble d'affectations :
`assigned` = `present` + `absent` + `unmarked`."""
assigned: int = 0
present: int = 0
absent: int = 0
unmarked: int = 0
class StatsAttendancePosition(ApiModel):
position_id: uuid.UUID
name: str
# Créneaux terminés (`ends_at <= as_of`) et créneaux en cours.
past: StatsAttendanceCounts
live: StatsAttendanceCounts
live_shifts: int
class StatsAttendance(ApiModel):
"""Présents / absents pointés par les référents, par poste : sur les
créneaux déjà terminés et sur ceux en cours à `as_of` (heure du calcul,
locale). Les créneaux à venir ne comptent pas."""
as_of: dt.datetime
past: StatsAttendanceCounts
live: StatsAttendanceCounts
live_shifts: int
by_position: list[StatsAttendancePosition]
class EditionStats(ApiModel):
registrations: StatsRegistrations
answers: StatsAnswers
payments: StatsPayments
artist_wishes: list[StatsArtistWish]
needs: StatsNeeds
attendance: StatsAttendance
+58
View File
@@ -16,6 +16,9 @@ from app.api.schemas import (
EditionStats,
StatsAnswers,
StatsArtistWish,
StatsAttendance,
StatsAttendanceCounts,
StatsAttendancePosition,
StatsNeedDay,
StatsNeedPosition,
StatsNeeds,
@@ -30,6 +33,7 @@ from app.models import (
Account,
Artist,
Assignment,
AttendanceStatus,
DayPart,
Edition,
FeePaymentStatus,
@@ -108,6 +112,59 @@ def _payments(
)
def _count(counts: StatsAttendanceCounts, assignment: Assignment) -> None:
counts.assigned += 1
if assignment.attendance == AttendanceStatus.PRESENT:
counts.present += 1
elif assignment.attendance == AttendanceStatus.ABSENT:
counts.absent += 1
else:
counts.unmarked += 1
def _attendance(
positions: list[Position], shifts: list[Shift], assignments: list[Assignment]
) -> StatsAttendance:
"""Pointage par poste, sur les créneaux terminés et ceux en cours."""
now = local_now()
past = StatsAttendanceCounts()
live = StatsAttendanceCounts()
by_position = {
p.id: StatsAttendancePosition(
position_id=p.id,
name=p.name,
past=StatsAttendanceCounts(),
live=StatsAttendanceCounts(),
live_shifts=0,
)
for p in positions
}
# Créneau terminé ⇒ « past », en cours ⇒ « live », à venir : ignoré.
phase: dict = {}
for s in shifts:
if s.ends_at <= now:
phase[s.id] = ("past", s.position_id)
elif s.starts_at <= now:
phase[s.id] = ("live", s.position_id)
by_position[s.position_id].live_shifts += 1
for a in assignments:
if a.shift_id not in phase:
continue
when, position_id = phase[a.shift_id]
row = by_position[position_id]
_count(past if when == "past" else live, a)
_count(row.past if when == "past" else row.live, a)
return StatsAttendance(
as_of=now,
past=past,
live=live,
live_shifts=sum(r.live_shifts for r in by_position.values()),
by_position=[
r for r in by_position.values() if r.past.assigned or r.live.assigned or r.live_shifts
],
)
@router.get("", response_model=EditionStats)
def get_stats(
edition: Edition = Depends(get_edition),
@@ -256,4 +313,5 @@ def get_stats(
payments=_payments(volunteers, pending_unfinished),
artist_wishes=artist_stats,
needs=needs,
attendance=_attendance(positions, shifts, assignments),
)
@@ -0,0 +1,34 @@
"""Pointage des bénévoles
Revision ID: 0030_assignment_attendance
Revises: 0029_volunteer_messages
Create Date: 2026-10-28
`assignment.attendance` (`PRESENT`/`ABSENT`, `NULL` = pas pointé) et
`attendance_marked_at` : le pointage du référent était gardé dans le
navigateur seulement et se perdait au rechargement.
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
revision: str = "0030_assignment_attendance"
down_revision: str | None = "0029_volunteer_messages"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
_ATTENDANCE = sa.Enum("PRESENT", "ABSENT", name="attendancestatus", native_enum=False)
def upgrade() -> None:
op.add_column("assignment", sa.Column("attendance", _ATTENDANCE, nullable=True))
op.add_column(
"assignment", sa.Column("attendance_marked_at", sa.DateTime(), nullable=True)
)
def downgrade() -> None:
op.drop_column("assignment", "attendance_marked_at")
op.drop_column("assignment", "attendance")
+2
View File
@@ -27,6 +27,7 @@ from app.models.domain import (
)
from app.models.enums import (
AssignmentSource,
AttendanceStatus,
BadgeAlign,
BadgeFace,
BadgeFieldType,
@@ -42,6 +43,7 @@ __all__ = [
"Artist",
"Assignment",
"AssignmentSource",
"AttendanceStatus",
"BadgeAlign",
"BadgeFace",
"BadgeField",
+9
View File
@@ -17,6 +17,7 @@ from sqlmodel import Field, Relationship, SQLModel, UniqueConstraint
from app.models.base import TimestampMixin
from app.models.enums import (
AssignmentSource,
AttendanceStatus,
BadgeAlign,
BadgeFace,
BadgeFieldType,
@@ -414,6 +415,14 @@ class Assignment(TimestampMixin, table=True):
default=False,
sa_column=Column(Boolean, nullable=False, server_default="false"),
)
# Pointage par le référent du poste (ou la coordination) : `None` tant
# que personne n'a pointé. `attendance_marked_at` : heure du dernier
# pointage, affichée sur l'écran référent.
attendance: AttendanceStatus | None = Field(
default=None,
sa_column=Column(SAEnum(AttendanceStatus, native_enum=False), nullable=True),
)
attendance_marked_at: dt.datetime | None = None
volunteer: Volunteer = Relationship(back_populates="assignments")
shift: Shift = Relationship(back_populates="assignments")
+7
View File
@@ -18,6 +18,13 @@ class AssignmentSource(enum.StrEnum):
MANUAL = "MANUAL"
class AttendanceStatus(enum.StrEnum):
"""Pointage d'un bénévole sur son créneau, par le référent du poste."""
PRESENT = "PRESENT"
ABSENT = "ABSENT"
class PositionCategory(enum.StrEnum):
BAR = "BAR"
TICKETING = "TICKETING"
+72
View File
@@ -0,0 +1,72 @@
"""Pointage présent / absent d'un bénévole sur son créneau
(`PUT .../assignments/{id}/attendance`) : enregistré en base, par le
référent du poste ou la coordination."""
from app.models import Assignment, Role
from tests.factories import dt_at, make_position, make_shift, make_volunteer
def _assignment(session, edition, lead=None):
position = make_position(session, edition)
if lead is not None:
position.lead_account_id = lead.id
session.add(position)
session.commit()
shift = make_shift(session, edition, position, start=dt_at(10, 18), end=dt_at(10, 22))
volunteer = make_volunteer(session, edition)
assignment = Assignment(edition_id=edition.id, volunteer_id=volunteer.id, shift_id=shift.id)
session.add(assignment)
session.commit()
session.refresh(assignment)
return assignment
def _url(edition, assignment):
return f"/editions/{edition.id}/assignments/{assignment.id}/attendance"
def test_lead_marks_attendance_and_it_is_kept(edition, session, make_account, as_account):
lead = make_account(Role.LEAD)
assignment = _assignment(session, edition, lead)
client = as_account(lead)
resp = client.put(_url(edition, assignment), json={"attendance": "PRESENT"})
assert resp.status_code == 200
assert resp.json()["attendance"] == "PRESENT"
assert resp.json()["attendance_marked_at"] is not None
# Relu ensuite (rechargement de l'écran) : le pointage est toujours là.
(listed,) = client.get(f"/editions/{edition.id}/assignments").json()
assert listed["attendance"] == "PRESENT"
def test_attendance_can_be_switched_and_cleared(client, edition, session):
assignment = _assignment(session, edition)
assert client.put(_url(edition, assignment), json={"attendance": "ABSENT"}).json()[
"attendance"
] == "ABSENT"
cleared = client.put(_url(edition, assignment), json={"attendance": None}).json()
assert cleared["attendance"] is None
assert cleared["attendance_marked_at"] is None
def test_lead_of_another_position_cannot_mark(edition, session, make_account, as_account):
assignment = _assignment(session, edition, make_account(Role.LEAD))
resp = as_account(make_account(Role.LEAD)).put(
_url(edition, assignment), json={"attendance": "PRESENT"}
)
assert resp.status_code == 403
def test_volunteer_cannot_mark(edition, session, make_account, as_account):
assignment = _assignment(session, edition)
resp = as_account(make_account(Role.VOLUNTEER)).put(
_url(edition, assignment), json={"attendance": "PRESENT"}
)
assert resp.status_code == 403
def test_unknown_status_is_rejected(client, edition, session):
assignment = _assignment(session, edition)
resp = client.put(_url(edition, assignment), json={"attendance": "LATE"})
assert resp.status_code == 422
+46
View File
@@ -187,3 +187,49 @@ def test_filled_slots_never_count_overstaffing(client, edition, session):
assert position["assigned"] == 3
assert position["filled"] == 2
assert position["ideal_slots"] == 4
def test_attendance_splits_past_and_live_shifts_per_position(
client, edition, session, monkeypatch
):
import app.api.stats as stats_api
from app.models import AttendanceStatus
# Maintenant : le 10 à 20:00 — 14h–18h terminé, 18h–22h en cours,
# 22h–02h à venir (ignoré).
monkeypatch.setattr(stats_api, "local_now", lambda: dt_at(10, 20))
bar = make_position(session, edition, name="Bar")
parking = make_position(session, edition, name="Parking")
past = make_shift(session, edition, bar, start=dt_at(10, 14), end=dt_at(10, 18))
live = make_shift(session, edition, bar, start=dt_at(10, 18), end=dt_at(10, 22))
later = make_shift(session, edition, parking, start=dt_at(10, 22), end=dt_at(11, 2))
def assign(shift, name, attendance=None):
volunteer = make_volunteer(session, edition, first_name=name)
session.add(
Assignment(
edition_id=edition.id,
volunteer_id=volunteer.id,
shift_id=shift.id,
attendance=attendance,
)
)
assign(past, "A", AttendanceStatus.PRESENT)
assign(past, "B", AttendanceStatus.ABSENT)
assign(past, "C")
assign(live, "D", AttendanceStatus.PRESENT)
assign(live, "E")
assign(later, "F", AttendanceStatus.PRESENT)
session.commit()
att = client.get(_url(edition)).json()["attendance"]
assert att["past"] == {"assigned": 3, "present": 1, "absent": 1, "unmarked": 1}
assert att["live"] == {"assigned": 2, "present": 1, "absent": 0, "unmarked": 1}
assert att["live_shifts"] == 1
# Parking n'a encore aucun créneau commencé : absent du détail.
(row,) = att["by_position"]
assert row["name"] == "Bar"
assert row["past"]["assigned"] == 3
assert row["live"]["present"] == 1
assert row["live_shifts"] == 1