Source code for pas.forms.attendence
import datetime
from decimal import Decimal, ROUND_HALF_UP
from datetime import date
from wtforms import HiddenField
from onegov.form import Form
from onegov.form.fields import ChosenSelectField
from onegov.form.fields import MultiCheckboxField
from onegov.form.utils import coerce_uuid
from onegov.pas import _
from onegov.pas.collections import PASCommissionCollection
from onegov.pas.collections import PASParliamentarianCollection
from onegov.pas.utils import get_attendance_parliamentarians
from onegov.pas.utils import get_attendance_settlement_run
from onegov.pas.custom import AttendenceCollection
from onegov.pas.models import PASCommission, PASCommissionMembership
from onegov.pas.models import SettlementRun
from onegov.pas.models.attendence import TYPES
from wtforms.fields import BooleanField
from wtforms.fields import DateField
from wtforms.fields import DecimalField
from wtforms.fields import RadioField
from wtforms.fields import SelectChoice
from wtforms.validators import InputRequired, NumberRange, ValidationError
from onegov.user import User
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Collection
from onegov.core.request import CoreRequest
from onegov.pas.models import Attendence
from onegov.pas.request import PasRequest
from typing import Any
from uuid import UUID
[docs]
class SettlementRunBoundMixin:
if TYPE_CHECKING:
# forward declaration of required attributes
request: CoreRequest
[docs]
def ensure_date(self) -> bool:
if self.date.data:
query = self.request.session.query(SettlementRun)
query = query.filter(
SettlementRun.active.is_(True),
SettlementRun.start <= self.date.data,
)
if not query.first():
assert isinstance(self.date.errors, list)
self.date.errors.append(
_('No within an active settlement run.')
)
return False
return True
[docs]
def set_default_value_to_settlement_run_start(self) -> None:
if self.request.method == 'POST':
return
settlement_run = get_attendance_settlement_run(self.request)
if settlement_run is not None:
self.date.data = settlement_run.start
[docs]
def hours_to_minutes(hours: Decimal | None) -> Decimal:
"""A duration is entered in hours and stored in minutes. Both are
exact, 2.01 h is stored as 120.60 and not as 120.
"""
return (hours or Decimal('0')) * 60
[docs]
def minutes_to_hours(minutes: Decimal) -> Decimal:
"""The inverse of :func:`hours_to_minutes`. Rows written before the
duration became fractional are not exact, they get cut down to the two
places the field accepts.
"""
return (minutes / 60).quantize(Decimal('0.01'), rounding=ROUND_HALF_UP)
[docs]
class AttendenceForm(Form, SettlementRunBoundMixin):
[docs]
date = DateField(
label=_('Date'),
validators=[InputRequired()],
default=datetime.date.today
)
[docs]
duration = DecimalField(
places=2,
label=_('Duration in hours'),
validators=[InputRequired(), NumberRange(min=0)],
)
[docs]
type = RadioField(
label=_('Type'),
choices=[SelectChoice(value, label) for value, label in TYPES.items()],
validators=[InputRequired()],
default='plenary'
)
[docs]
parliamentarian_id = ChosenSelectField(
label=_('Parliamentarian'),
validators=[InputRequired()],
coerce=coerce_uuid,
)
[docs]
commission_id = ChosenSelectField(
label=_('Commission'),
validators=[InputRequired()],
coerce=coerce_uuid,
depends_on=('type', '!plenary'),
)
[docs]
def _can_edit_parliamentarian(
self, parliamentarian_id: UUID
) -> tuple[bool, str | None]:
"""Check if current user can edit attendance for given parl ID.
Returns: (can_edit, error_message)
"""
if not hasattr(self.request.identity, 'role'):
return (False, _('No role found for user.'))
role = self.request.identity.role
if role in ('admin', 'editor'):
return (True, None)
if role not in ('parliamentarian', 'commission_president'):
return (False, _('Insufficient permissions.'))
user = (
self.request.session.query(User)
.filter_by(username=self.request.identity.userid)
.first()
)
if not user or not user.parliamentarian: # type: ignore[attr-defined]
return (False, _('User has no parliamentarian record.'))
user_parl_id = user.parliamentarian.id # type: ignore[attr-defined]
if parliamentarian_id == user_parl_id:
return (True, None)
if role == 'parliamentarian':
return (False, _('You can only edit your own attendance.'))
target_parl = PASParliamentarianCollection(self.request.app).by_id(
parliamentarian_id
)
if not target_parl:
return (False, _('Target parliamentarian not found.'))
today = date.today()
presidencies = user.parliamentarian.commission_memberships_on( # type: ignore[attr-defined]
on_date=today,
role='president',
)
target_memberships = target_parl.commission_memberships_on(
on_date=today,
)
for presidency in presidencies:
if any(
membership.commission_id == presidency.commission_id
for membership in target_memberships
):
return (True, None)
return (
False,
_(
'You can only edit your own or your commission '
"members' attendance."
),
)
[docs]
def validate_parliamentarian_id(self, field: ChosenSelectField) -> None:
"""Prevent parliamentarians from editing other people's attendance."""
can_edit, error_msg = self._can_edit_parliamentarian(field.data)
if not can_edit and error_msg is not None:
raise ValidationError(error_msg)
[docs]
def ensure_commission(self) -> bool:
if (
self.type.data
and self.type.data != 'plenary'
and self.commission_id.data
and self.parliamentarian_id.data
):
collection = PASParliamentarianCollection(self.request.app)
parliamentarian = collection.by_id(self.parliamentarian_id.data)
if parliamentarian:
if self.commission_id.data not in (
membership.commission_id
for membership in parliamentarian.commission_memberships
):
assert isinstance(self.commission_id.errors, list)
self.commission_id.errors.append(
_('Parliamentarian is not in this commission.')
)
return False
return True
[docs]
def process_obj(self, obj: Attendence) -> None: # type:ignore
super().process_obj(obj)
self.duration.data = minutes_to_hours(obj.duration)
[docs]
def populate_obj( # type: ignore[override]
self,
obj: Attendence, # type: ignore[override]
exclude: Collection[str] | None = None,
include: Collection[str] | None = None
) -> None:
super().populate_obj(obj, exclude, include)
obj.commission_id = obj.commission_id or None
obj.duration = hours_to_minutes(obj.duration)
if obj.type == 'plenary':
obj.commission_id = None
[docs]
def get_useful_data(self) -> dict[str, Any]: # type:ignore[override]
result = super().get_useful_data()
result['commission_id'] = result.get('commission_id') or None
if result.get('type', '') == 'plenary':
result['commission_id'] = None
result['duration'] = hours_to_minutes(result.get('duration'))
return result
[docs]
def on_request(self) -> None:
self.set_default_value_to_settlement_run_start()
if not self.request.is_admin:
self.type.choices = [
choice
for choice in self.type.choices
if choice.value != 'plenary'
]
if (
hasattr(self.request.identity, 'role')
and self.request.identity.role == 'parliamentarian'
):
user = (
self.request.session.query(User)
.filter_by(username=self.request.identity.userid)
.first()
)
if user and (parliamentarian := user.parliamentarian): # type: ignore[attr-defined]
self.parliamentarian_id.choices = [SelectChoice(
parliamentarian.id.hex,
parliamentarian.title
)]
else:
self.parliamentarian_id.choices = []
self.request.warn_no_parliamentarian()
elif (
hasattr(self.request.identity, 'role')
and self.request.identity.role == 'commission_president'
):
user = (
self.request.session.query(User)
.filter_by(username=self.request.identity.userid)
.first()
)
if user and (parliamentarian := user.parliamentarian): # type: ignore[attr-defined]
choices = [SelectChoice(
parliamentarian.id.hex,
parliamentarian.title,
)]
today = date.today()
presidencies = user.parliamentarian.commission_memberships_on( # type: ignore[attr-defined]
on_date=today,
role='president',
)
seen = {parliamentarian.id}
for presidency in presidencies:
memberships = self.request.session.query(
PASCommissionMembership
).filter_by(commission_id=presidency.commission_id)
for membership in memberships:
if (
membership.is_active_on(today)
and membership.parliamentarian_id
!= user.parliamentarian.id # type: ignore[attr-defined]
):
member = membership.parliamentarian
if member.id in seen:
continue
seen.add(member.id)
choices.append(
SelectChoice(member.id.hex, member.title)
)
self.parliamentarian_id.choices = choices
else:
self.parliamentarian_id.choices = []
self.request.warn_no_parliamentarian()
else:
self.parliamentarian_id.choices = [
SelectChoice(p.id.hex, p.title)
for p in get_attendance_parliamentarians(self.request)
]
# Filter commission choices based on user role
if hasattr(
self.request.identity, 'role'
) and self.request.identity.role in (
'parliamentarian',
'commission_president',
):
user = (
self.request.session.query(User)
.filter_by(username=self.request.identity.userid)
.first()
)
if user and user.parliamentarian: # type: ignore[attr-defined]
parliamentarian = user.parliamentarian # type: ignore[attr-defined]
memberships = parliamentarian.commission_memberships_on(
on_date=date.today(),
)
commission_ids = [
membership.commission_id
for membership in memberships
]
self.commission_id.choices = [
SelectChoice(commission.id.hex, commission.title)
for commission in PASCommissionCollection(
self.request.session
).query()
if commission.id in commission_ids
]
else:
self.commission_id.choices = []
else:
self.commission_id.choices = [
SelectChoice(commission.id.hex, commission.title)
for commission in PASCommissionCollection(
self.request.session
).query()
]
self.commission_id.choices.insert(0, SelectChoice('', '-'))
[docs]
class AttendenceAddForm(AttendenceForm):
[docs]
def on_request(self) -> None:
super().on_request()
# Limit choices based on user role
if (hasattr(self.request.identity, 'role')
and self.request.identity.role == 'parliamentarian'):
# Regular parliamentarians can only select themselves
user = self.request.session.query(User).filter_by(
username=self.request.identity.userid).first()
if user and (parliamentarian := user.parliamentarian): # type: ignore[attr-defined]
self.parliamentarian_id.choices = [
SelectChoice(parliamentarian.id.hex, parliamentarian.title)
]
else:
self.parliamentarian_id.choices = []
elif (hasattr(self.request.identity, 'role')
and self.request.identity.role == 'commission_president'):
# Commission presidents can select themselves + commission members
user = self.request.session.query(User).filter_by(
username=self.request.identity.userid).first()
if user and (parliamentarian := user.parliamentarian): # type: ignore[attr-defined]
choices = [SelectChoice(
parliamentarian.id.hex,
parliamentarian.title
)]
# Add commission members
today = date.today()
presidencies = user.parliamentarian.commission_memberships_on( # type: ignore[attr-defined]
on_date=today,
role='president',
)
seen = {parliamentarian.id}
for presidency in presidencies:
memberships = self.request.session.query(
PASCommissionMembership
).filter_by(commission_id=presidency.commission_id)
for membership in memberships:
if (
membership.is_active_on(today)
and membership.parliamentarian_id
!= user.parliamentarian.id # type: ignore[attr-defined]
):
member = membership.parliamentarian
if member.id in seen:
continue
seen.add(member.id)
choices.append(
SelectChoice(member.id.hex, member.title)
)
self.parliamentarian_id.choices = choices
else:
self.parliamentarian_id.choices = []
else:
self.parliamentarian_id.choices = [
SelectChoice(p.id.hex, p.title)
for p in get_attendance_parliamentarians(self.request)
]
[docs]
class AttendenceAddPlenaryForm(Form, SettlementRunBoundMixin):
[docs]
date = DateField(
label=_('Date'),
validators=[InputRequired()],
default=datetime.date.today
)
[docs]
duration = DecimalField(
places=2,
label=_('Duration in hours'),
validators=[InputRequired(), NumberRange(min=0)],
)
[docs]
parliamentarian_id = MultiCheckboxField(
label=_('Parliamentarian'),
validators=[InputRequired()],
coerce=coerce_uuid,
)
[docs]
def get_useful_data(self) -> dict[str, Any]: # type:ignore[override]
result = super().get_useful_data()
result['duration'] = hours_to_minutes(result.get('duration'))
return result
[docs]
def on_request(self) -> None:
self.set_default_value_to_settlement_run_start()
self.parliamentarian_id.choices = [
SelectChoice(p.id.hex, p.title)
for p in get_attendance_parliamentarians(self.request)
]
# FIXME: Setting this unconditionally seems wrong
self.parliamentarian_id.data = [
value
for choice in self.parliamentarian_id.choices
if (value := coerce_uuid(choice.value)) is not None
]
[docs]
BULK_MEETING_TYPES: dict[str, str] = {
'commission': _('Commission meeting'),
'shortest': _('Shortest meeting'),
}
[docs]
class AttendenceAddCommissionBulkForm(Form, SettlementRunBoundMixin):
"""Bulk form for commission-based meetings (commission or shortest)."""
[docs]
date = DateField(
label=_('Date'),
validators=[InputRequired()],
default=datetime.date.today
)
[docs]
type = RadioField(
label=_('Type'),
choices=[
SelectChoice(value, label)
for value, label in BULK_MEETING_TYPES.items()
],
validators=[InputRequired()],
default='commission',
)
[docs]
duration = DecimalField(
places=2,
label=_('Duration in hours'),
validators=[InputRequired(), NumberRange(min=0)],
)
[docs]
commission_id = ChosenSelectField(
label=_('Commission'),
validators=[InputRequired()],
coerce=coerce_uuid
)
[docs]
parliamentarian_id = MultiCheckboxField(
label=_('Parliamentarian'),
validators=[InputRequired()],
coerce=coerce_uuid
)
[docs]
def get_useful_data(self) -> dict[str, Any]: # type:ignore[override]
result = super().get_useful_data()
result['duration'] = hours_to_minutes(result.get('duration'))
return result
[docs]
def on_request(self) -> None:
self.set_default_value_to_settlement_run_start()
commissions = PASCommissionCollection(self.request.session).query()
parliamentarians = get_attendance_parliamentarians(self.request)
if (
hasattr(self.request.identity, 'role')
and self.request.identity.role == 'commission_president'
):
user = (
self.request.session.query(User)
.filter_by(username=self.request.identity.userid)
.first()
)
if user and user.parliamentarian: # type: ignore[attr-defined]
today = date.today()
led_commission_ids = {
membership.commission_id
for membership in (
user.parliamentarian.commission_memberships_on( # type: ignore[attr-defined]
on_date=today,
role='president',
)
)
}
commissions = commissions.filter(
PASCommission.id.in_(led_commission_ids)
)
member_ids = {
membership.parliamentarian_id
for membership in self.request.session.query(
PASCommissionMembership
)
if membership.commission_id in led_commission_ids
and membership.is_active_on(today)
}
parliamentarians = [
parliamentarian
for parliamentarian in parliamentarians
if parliamentarian.id in member_ids
]
else:
commissions = commissions.filter(PASCommission.id.in_([]))
parliamentarians = []
self.commission_id.choices = [
SelectChoice(commission.id.hex, commission.title)
for commission in commissions
]
self.parliamentarian_id.choices = [
SelectChoice(p.id.hex, p.title)
for p in parliamentarians
]
# FIXME: Setting this unconditionally seems wrong
self.parliamentarian_id.data = []
self.request.warn_no_parliamentarian() # type: ignore[attr-defined]
[docs]
class AttendenceAddFileStudyBulkForm(AttendenceAddCommissionBulkForm):
""" Bulk form for file studies. """
[docs]
def get_useful_data(self) -> dict[str, Any]: # type:ignore[override]
result = super().get_useful_data()
result['type'] = 'study'
return result
[docs]
class AttendenceEditBulkForm(Form, SettlementRunBoundMixin):
""" Edit form for bulk attendance changes. """
[docs]
date = DateField(
label=_('Date'),
validators=[InputRequired()],
default=datetime.date.today
)
[docs]
duration = DecimalField(
places=2,
label=_('Duration in hours'),
validators=[InputRequired(), NumberRange(min=0)],
)
[docs]
commission_id = ChosenSelectField(
label=_('Commission'),
validators=[InputRequired()],
coerce=coerce_uuid,
# FIXME: This only applies client-side in order to protect this
# from being changed, you need to exclude the field in
# populate_obj
render_kw={'readonly': True}
)
[docs]
parliamentarian_id = MultiCheckboxField(
label=_('Parliamentarian'),
validators=[InputRequired()],
coerce=coerce_uuid
)
[docs]
def get_useful_data(self) -> dict[str, Any]: # type:ignore[override]
result = super().get_useful_data()
result['duration'] = hours_to_minutes(result.get('duration'))
return result
[docs]
def on_request(self) -> None:
self.set_default_value_to_settlement_run_start()
self.commission_id.choices = [
SelectChoice(commission.id.hex, commission.title)
for commission
in PASCommissionCollection(self.request.session).query()
]
self.parliamentarian_id.choices = [
SelectChoice(p.id.hex, p.title)
for p in get_attendance_parliamentarians(self.request)
]
[docs]
class AttendenceCommissionBulkEditForm(AttendenceEditBulkForm):
""" Edit form for commission bulk attendance changes. """
[docs]
def process_obj(self, obj: Attendence) -> None: # type: ignore[override]
super().process_obj(obj)
memberships = self.request.session.query(
PASCommissionMembership).filter(
PASCommissionMembership.commission_id == obj.commission_id
).all()
eligible = {
p.id for p in get_attendance_parliamentarians(self.request)
}
self.parliamentarian_id.choices = [
SelectChoice(value, label)
for value, label in {
m.parliamentarian_id.hex: m.parliamentarian.title
for m in memberships
if m.parliamentarian_id in eligible
}.items()
]
self.duration.data = minutes_to_hours(obj.duration)
self.abschluss.data = obj.abschluss
attendences = AttendenceCollection(
self.request.session).query().filter_by(
bulk_edit_id=obj.bulk_edit_id
)
assert obj.commission is not None
self.commission_id.choices = [
SelectChoice(obj.commission.id.hex, obj.commission.title)
]
self.commission_id.data = obj.commission_id
self.parliamentarian_id.data = [
attendence.parliamentarian_id for attendence in attendences
]
[docs]
def populate_obj(self, obj: Attendence) -> None: # type: ignore[override]
obj.duration = hours_to_minutes(self.duration.data)
obj.date = self.date.data # type: ignore[assignment]
obj.commission_id = self.commission_id.data
obj.abschluss = self.abschluss.data
[docs]
class AttendencePlenaryBulkEditForm(AttendenceEditBulkForm):
""" Edit form for plenary bulk attendance changes. """
[docs]
def get_useful_data(self) -> dict[str, Any]: # type:ignore[override]
result = super().get_useful_data()
result['type'] = 'plenary'
return result
[docs]
def process_obj(self, obj: Attendence) -> None: # type: ignore[override]
super().process_obj(obj)
self.duration.data = minutes_to_hours(obj.duration)
attendences = AttendenceCollection(
self.request.session).query().filter_by(
bulk_edit_id=obj.bulk_edit_id
)
self.parliamentarian_id.choices = [
SelectChoice(p.id.hex, p.title)
for p in get_attendance_parliamentarians(self.request)
]
self.parliamentarian_id.data = [
attendence.parliamentarian.id
for attendence in attendences
]
[docs]
def populate_obj(self, obj: Attendence) -> None: # type: ignore[override]
obj.duration = hours_to_minutes(self.duration.data)
obj.date = self.date.data # type: ignore[assignment]
[docs]
class AttendenceAddCommissionForm(Form, SettlementRunBoundMixin):
[docs]
date = DateField(
label=_('Date'),
validators=[InputRequired()],
default=datetime.date.today
)
[docs]
duration = DecimalField(
places=2,
label=_('Duration in hours'),
validators=[InputRequired(), NumberRange(min=0)],
)
[docs]
type = RadioField(
label=_('Type'),
choices=[
SelectChoice(key, value)
for key, value in TYPES.items()
if key != 'plenary'
],
validators=[InputRequired()],
default='commission'
)
[docs]
parliamentarian_id = MultiCheckboxField(
label=_('Parliamentarian'),
validators=[InputRequired()],
coerce=coerce_uuid
)
[docs]
def get_useful_data(self) -> dict[str, Any]: # type:ignore[override]
result = super().get_useful_data()
result['commission_id'] = self.model.id
result['duration'] = hours_to_minutes(result.get('duration'))
return result
[docs]
def on_request(self) -> None:
self.set_default_value_to_settlement_run_start()
self.parliamentarian_id.choices = [
SelectChoice(
membership.parliamentarian.id.hex,
membership.parliamentarian.title
)
for membership in self.model.memberships
]
# FIXME: Setting this uncondtionally seems wrong
self.parliamentarian_id.data = [
value
for choice in self.parliamentarian_id.choices
if (value := coerce_uuid(choice.value)) is not None
]