Initial commit

This commit is contained in:
2026-06-01 16:48:21 +02:00
commit 53d157535a
159 changed files with 13641 additions and 0 deletions
View File
+65
View File
@@ -0,0 +1,65 @@
"""Date calculation utilities for leave planning.
Pure functions with no I/O or state. Used by planning and analytics services.
"""
import datetime
from typing import Union
def add_months(date: datetime.date, n: int) -> datetime.date:
"""Add or subtract months from a date, handling month-end edge cases.
Examples:
add_months(datetime.date(2026, 1, 31), 1) -> 2026-02-28 (or 29 in leap year)
add_months(datetime.date(2026, 3, 31), -1) -> 2026-02-28
"""
month = date.month + n
year = date.year
while month > 12:
month -= 12
year += 1
while month < 1:
month += 12
year -= 1
# Handle month-end: if original day doesn't exist in target month,
# use last day of that month
day = min(date.day, _days_in_month(year, month))
return datetime.date(year, month, day)
def _days_in_month(year: int, month: int) -> int:
"""Return number of days in the given month."""
if month == 2:
return 29 if _is_leap_year(year) else 28
if month in (4, 6, 9, 11):
return 30
return 31
def _is_leap_year(year: int) -> bool:
"""Check if a year is a leap year."""
return year % 4 == 0 and (year % 100 != 0 or year % 400 == 0)
def age_in_months(birth_date: datetime.date, on: datetime.date) -> int:
"""Calculate exact age in months between two dates.
If on < birth_date, returns negative value (person not yet born).
"""
months = (on.year - birth_date.year) * 12 + (on.month - birth_date.month)
# Adjust if birthday hasn't occurred this month yet
if on.day < birth_date.day:
months -= 1
return months
def fourth_birthday(birth_date: datetime.date) -> datetime.date:
"""Return the date when a child turns 4 years old."""
return add_months(birth_date, 48)
+45
View File
@@ -0,0 +1,45 @@
"""Domain error types.
Use these instead of bare exceptions so services/routers can map them to
meaningful HTTP responses. Never swallow errors silently (see ARCHITECTURE.md §10).
"""
class DomainError(Exception):
"""Base class for all application-domain errors."""
class NotFoundError(DomainError):
"""Raised when an entity cannot be found by its identifier."""
class ValidationError(DomainError):
"""Raised when input fails a business rule (not just schema validation)."""
class InsufficientDaysError(DomainError):
"""Raised when a parent doesn't have enough remaining days to plan."""
class InvalidDayTypeError(ValidationError):
"""Raised when a planned-day type is not recognized."""
class DubbeldagError(ValidationError):
"""Raised when a dubbeldag breaks the 15-month window or the 60-day quota."""
class AuthError(DomainError):
"""Raised when credentials are invalid or an auth operation fails."""
class UnauthorizedError(DomainError):
"""Raised when the caller lacks permission to perform an action."""
class ConflictError(DomainError):
"""Raised when a resource already exists (e.g. duplicate email on register)."""
class InvitationError(ValidationError):
"""Raised when an invitation token is invalid, expired, or already redeemed."""
+53
View File
@@ -0,0 +1,53 @@
"""Swedish public holiday utilities for leave planning.
Uses the holidays library to get röda dagar (red days) in Sweden.
Results are cached per year to avoid repeated parsing.
"""
import datetime
from typing import Dict
import holidays
_holiday_cache: Dict[int, Dict[datetime.date, str]] = {}
def get_swedish_holidays(year: int) -> Dict[datetime.date, str]:
"""Get all Swedish public holidays (röda dagar) for a given year.
Returns a mapping of date -> holiday name in Swedish.
The result is cached per year; subsequent calls with the same year
return the cached value without re-parsing.
Args:
year: Calendar year (e.g., 2026)
Returns:
Dict mapping datetime.date to holiday name (e.g. "Juldagen", "Midsommar")
"""
if year in _holiday_cache:
return _holiday_cache[year]
# Get Swedish holidays for the year with Swedish-language names.
se_holidays = holidays.Sweden(years=year, language="sv")
# Convert to our format: date -> name
result: Dict[datetime.date, str] = {}
for holiday_date, holiday_name in sorted(se_holidays.items()):
result[holiday_date] = holiday_name
_holiday_cache[year] = result
return result
def is_swedish_holiday(date: datetime.date) -> bool:
"""Check if a date is a Swedish public holiday."""
holidays_for_year = get_swedish_holidays(date.year)
return date in holidays_for_year
def get_holiday_name(date: datetime.date) -> str | None:
"""Get the name of a Swedish public holiday, or None if not a holiday."""
holidays_for_year = get_swedish_holidays(date.year)
return holidays_for_year.get(date)
+147
View File
@@ -0,0 +1,147 @@
"""Predicate helpers for the tips engine.
These are pure functions (no I/O) that the tips service calls after loading
the workspace context. Each takes pre-fetched data and returns a bool or a
supporting value. Keeping them pure makes them trivial to unit-test.
"""
from __future__ import annotations
from datetime import date, timedelta
import holidays
from dateutil.relativedelta import relativedelta
from app.config import settings
from app.entities.child import Child
from app.entities.personal_data import PersonalData
from app.entities.planned_leave import PlannedLeaveDay
def weekly_intensity(planned_days: list[PlannedLeaveDay], period_weeks: float) -> float:
"""Average föräldrapenning days per week over a given period.
Only sjukpenning+lagsta count; vacation days are excluded (they don't draw
from the FP pool and are not the planning knob that affects SGI).
"""
if period_weeks <= 0:
return 0.0
fp_days = [d for d in planned_days if d.day_type in ("sjukpenning", "lagsta")]
return len(fp_days) / period_weeks
def days_left_at_age_4_projection(
days_used: int,
planned_fp_before_b4: int,
total_days: int,
) -> tuple[int, int]:
"""Return (projected_left_at_b4, expiring_at_b4) without hitting the DB.
Delegates the full DB-backed version to AnalyticsService. Use this in
tests or when the numbers are already loaded.
"""
rules = settings.rules
projected_left = total_days - (days_used + planned_fp_before_b4)
expiring = max(projected_left - rules.max_savable_days_at_age_4, 0)
return projected_left, expiring
def is_child_under_age(birth_date: date, months: int) -> bool:
"""True if the child's age in months is less than `months`."""
return date.today() < birth_date + relativedelta(months=months)
def is_child_over_age(birth_date: date, years: int) -> bool:
"""True if the child has turned `years`."""
return date.today() >= birth_date + relativedelta(years=years)
def sgi_at_risk(
personal_data: PersonalData,
planned_days: list[PlannedLeaveDay],
child_birth_date: date,
period_start: date,
period_end: date,
) -> bool:
"""True if the parent risks losing SGI (< 5 FP days/week after age 1).
SGI is protected automatically in the first 12 months (§6). After that a
non-working, non-jobseeking parent must take ≥5 days/week to keep full SGI.
"""
rules = settings.rules
if not is_child_over_age(child_birth_date, 1):
return False # still in the protected first year
if personal_data.employment_status in ("employed", "jobseeking"):
return False # working/registered jobseeker is protected regardless
period_days = [
d
for d in planned_days
if d.parent_personal_data_uuid == personal_data.uuid
and period_start <= d.date <= period_end
and d.day_type in ("sjukpenning", "lagsta")
]
weeks = ((period_end - period_start).days + 1) / 7
intensity = weekly_intensity(period_days, weeks)
return intensity < rules.sgi_min_days_per_week_after_age_1
def red_day_clusters_in_range(
from_date: date, to_date: date, year: int | None = None
) -> list[tuple[date, date]]:
"""Return (start, end) pairs for clusters of ≥2 consecutive public holidays.
Uses the `holidays` library (Sweden locale). A "cluster" is any stretch
where consecutive days are either weekends or public holidays. Useful for
the red-day wrapping tip.
"""
target_year = year or from_date.year
se_holidays = holidays.Sweden(years=target_year)
clusters: list[tuple[date, date]] = []
cluster_start: date | None = None
current = from_date
while current <= to_date:
is_holiday = current in se_holidays or current.weekday() >= 5
if is_holiday:
if cluster_start is None:
cluster_start = current
else:
if cluster_start is not None and (current - cluster_start).days >= 2:
clusters.append((cluster_start, current - timedelta(days=1)))
cluster_start = None
current += timedelta(days=1)
if cluster_start is not None and (to_date - cluster_start).days >= 2:
clusters.append((cluster_start, to_date))
return clusters
def preschool_buffer_needed(child: Child, default_days_per_year: int) -> int:
"""Days that should be reserved for preschool planning-day closures.
Returns the default_days_per_year unless the child has no preschool_start_date
yet (in which case no reservation is needed).
"""
if child.preschool_start_date is None:
return 0
if date.today() < child.preschool_start_date:
return 0
return default_days_per_year
def has_vacation_near_expiry(
vacation_total: int,
vacation_saved_in: int,
threshold_remaining_days: int = 5,
) -> bool:
"""True if saved vacation days are approaching expiry (simple heuristic).
Swedish law: vacation days above 20 may be saved for up to 5 years.
This predicate flags when saved days exist — the tips service adds richer
expiry logic when it has the full vacation history per year.
"""
return vacation_saved_in > 0 and vacation_total > threshold_remaining_days