zhi
eb57197020
- New model: minimum_workloads table with JSON config column (per-user) - Schemas: MinimumWorkloadConfig, MinimumWorkloadUpdate, MinimumWorkloadResponse - Service: CRUD operations + check_workload_warnings() entry point for BE-CAL-007 - API: GET/PUT/PATCH /calendar/workload-config (self + admin routes) - Migration: auto-create minimum_workloads table on startup - Registered calendar router in main.py
145 lines
4.6 KiB
Python
145 lines
4.6 KiB
Python
"""MinimumWorkload service — CRUD and validation helpers.
|
|
|
|
BE-CAL-004: user-level workload config read/write + future validation entry point.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
from typing import Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.minimum_workload import (
|
|
DEFAULT_WORKLOAD_CONFIG,
|
|
CATEGORIES,
|
|
PERIODS,
|
|
MinimumWorkload,
|
|
)
|
|
from app.schemas.calendar import (
|
|
MinimumWorkloadConfig,
|
|
MinimumWorkloadUpdate,
|
|
WorkloadWarningItem,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Read
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def get_workload_config(db: Session, user_id: int) -> dict:
|
|
"""Return the raw config dict for *user_id*, falling back to defaults."""
|
|
row = db.query(MinimumWorkload).filter(MinimumWorkload.user_id == user_id).first()
|
|
if row is None:
|
|
return copy.deepcopy(DEFAULT_WORKLOAD_CONFIG)
|
|
return row.config
|
|
|
|
|
|
def get_workload_row(db: Session, user_id: int) -> Optional[MinimumWorkload]:
|
|
"""Return the ORM row or None."""
|
|
return db.query(MinimumWorkload).filter(MinimumWorkload.user_id == user_id).first()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Write (upsert)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def upsert_workload_config(
|
|
db: Session,
|
|
user_id: int,
|
|
update: MinimumWorkloadUpdate,
|
|
) -> MinimumWorkload:
|
|
"""Create or update the workload config for *user_id*.
|
|
|
|
Only the periods present in *update* are overwritten; the rest keep
|
|
their current (or default) values.
|
|
"""
|
|
row = db.query(MinimumWorkload).filter(MinimumWorkload.user_id == user_id).first()
|
|
|
|
if row is None:
|
|
row = MinimumWorkload(
|
|
user_id=user_id,
|
|
config=copy.deepcopy(DEFAULT_WORKLOAD_CONFIG),
|
|
)
|
|
db.add(row)
|
|
|
|
# Merge provided periods into existing config
|
|
current = copy.deepcopy(row.config) if row.config else copy.deepcopy(DEFAULT_WORKLOAD_CONFIG)
|
|
|
|
for period in PERIODS:
|
|
period_data = getattr(update, period, None)
|
|
if period_data is not None:
|
|
current[period] = period_data.model_dump()
|
|
|
|
# Ensure JSON column is flagged as dirty for SQLAlchemy
|
|
row.config = current
|
|
db.flush()
|
|
return row
|
|
|
|
|
|
def replace_workload_config(
|
|
db: Session,
|
|
user_id: int,
|
|
config: MinimumWorkloadConfig,
|
|
) -> MinimumWorkload:
|
|
"""Full replace of the workload config for *user_id*."""
|
|
row = db.query(MinimumWorkload).filter(MinimumWorkload.user_id == user_id).first()
|
|
|
|
if row is None:
|
|
row = MinimumWorkload(user_id=user_id, config=config.model_dump())
|
|
db.add(row)
|
|
else:
|
|
row.config = config.model_dump()
|
|
|
|
db.flush()
|
|
return row
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Validation entry point (BE-CAL-007 will flesh this out)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def check_workload_warnings(
|
|
db: Session,
|
|
user_id: int,
|
|
scheduled_minutes: dict[str, dict[str, int]],
|
|
) -> list[WorkloadWarningItem]:
|
|
"""Compare *scheduled_minutes* against the user's configured thresholds.
|
|
|
|
``scheduled_minutes`` has the same shape as the config:
|
|
{"daily": {"work": N, ...}, "weekly": {...}, ...}
|
|
|
|
Returns a list of warnings for every (period, category) where the
|
|
scheduled total is below the minimum. An empty list means no warnings.
|
|
|
|
This is the entry point that BE-CAL-007 and the calendar API endpoints
|
|
will call.
|
|
"""
|
|
config = get_workload_config(db, user_id)
|
|
warnings: list[WorkloadWarningItem] = []
|
|
|
|
for period in PERIODS:
|
|
cfg_period = config.get(period, {})
|
|
sch_period = scheduled_minutes.get(period, {})
|
|
for cat in CATEGORIES:
|
|
minimum = cfg_period.get(cat, 0)
|
|
if minimum <= 0:
|
|
continue
|
|
current = sch_period.get(cat, 0)
|
|
if current < minimum:
|
|
shortfall = minimum - current
|
|
warnings.append(WorkloadWarningItem(
|
|
period=period,
|
|
category=cat,
|
|
current_minutes=current,
|
|
minimum_minutes=minimum,
|
|
shortfall_minutes=shortfall,
|
|
message=(
|
|
f"{period.capitalize()} {cat.replace('_', '-')} workload "
|
|
f"is {current} min, below minimum of {minimum} min "
|
|
f"(shortfall: {shortfall} min)"
|
|
),
|
|
))
|
|
|
|
return warnings
|