Kyle Pope
be1fdc4551
Phase 1: Recurrence safety — MAX_OCCURRENCES=730 hard cap, adaptive 90-day horizon for daily events (interval<7), RecurrenceRule cross-field validation, ID bounds on location_id/calendar_id schemas. Phase 2: Dashboard correctness — shared calendar events now included in /dashboard and /upcoming via get_accessible_calendar_ids helper. Project stats consolidated into single GROUP BY query (saves 1 DB round-trip). Phase 3: Write performance — bulk db.add_all() for child events, removed redundant SELECT in this_and_future delete path. Phase 4: Frontend query efficiency — staleTime: 30_000 on calendar events query eliminates redundant refetches on mount/view switch. Backend LIMIT 2000 safety guard on events endpoint. Phase 5: Rate limiting — nginx limit_req zone on /api/events (30r/m) to prevent DB flooding via recurrence amplification. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
118 lines
4.1 KiB
Python
118 lines
4.1 KiB
Python
import json as _json
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
|
|
from datetime import datetime
|
|
from typing import Literal, Optional
|
|
|
|
|
|
class RecurrenceRule(BaseModel):
|
|
"""Structured recurrence rule — serialized to/from JSON string in the DB column."""
|
|
type: Literal["every_n_days", "weekly", "monthly_nth_weekday", "monthly_date"]
|
|
# every_n_days
|
|
interval: Optional[int] = Field(None, ge=1, le=365)
|
|
# weekly / monthly_nth_weekday
|
|
weekday: Optional[int] = Field(None, ge=0, le=6) # 0=Mon … 6=Sun
|
|
# monthly_nth_weekday
|
|
week: Optional[int] = Field(None, ge=1, le=4)
|
|
# monthly_date
|
|
day: Optional[int] = Field(None, ge=1, le=31)
|
|
|
|
@model_validator(mode="after")
|
|
def validate_required_fields(self):
|
|
"""Enforce required fields per rule type."""
|
|
if self.type == "every_n_days" and self.interval is None:
|
|
raise ValueError("every_n_days rule requires 'interval'")
|
|
if self.type == "monthly_nth_weekday":
|
|
if self.week is None or self.weekday is None:
|
|
raise ValueError("monthly_nth_weekday rule requires both 'week' and 'weekday'")
|
|
if self.type == "monthly_date" and self.day is None:
|
|
raise ValueError("monthly_date rule requires 'day'")
|
|
return self
|
|
|
|
|
|
def _coerce_recurrence_rule(v):
|
|
"""Accept None, dict, RecurrenceRule, or JSON/legacy strings gracefully."""
|
|
if v is None or v == "" or v == "null":
|
|
return None
|
|
if isinstance(v, dict):
|
|
return v
|
|
if isinstance(v, RecurrenceRule):
|
|
return v
|
|
if isinstance(v, str):
|
|
try:
|
|
parsed = _json.loads(v)
|
|
if isinstance(parsed, dict):
|
|
return parsed
|
|
except (_json.JSONDecodeError, TypeError):
|
|
pass
|
|
# Legacy simple strings like "daily", "weekly" — discard (not structured)
|
|
return None
|
|
return v
|
|
|
|
|
|
class CalendarEventCreate(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
title: str = Field(min_length=1, max_length=255)
|
|
description: Optional[str] = Field(None, max_length=5000)
|
|
start_datetime: datetime
|
|
end_datetime: datetime
|
|
all_day: bool = False
|
|
color: Optional[str] = Field(None, max_length=20)
|
|
location_id: Optional[int] = Field(None, ge=1, le=2147483647)
|
|
recurrence_rule: Optional[RecurrenceRule] = None
|
|
is_starred: bool = False
|
|
calendar_id: Optional[int] = Field(None, ge=1, le=2147483647)
|
|
|
|
@field_validator("recurrence_rule", mode="before")
|
|
@classmethod
|
|
def coerce_recurrence(cls, v):
|
|
return _coerce_recurrence_rule(v)
|
|
|
|
|
|
class CalendarEventUpdate(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
title: Optional[str] = Field(None, min_length=1, max_length=255)
|
|
description: Optional[str] = Field(None, max_length=5000)
|
|
start_datetime: Optional[datetime] = None
|
|
end_datetime: Optional[datetime] = None
|
|
all_day: Optional[bool] = None
|
|
color: Optional[str] = Field(None, max_length=20)
|
|
location_id: Optional[int] = Field(None, ge=1, le=2147483647)
|
|
recurrence_rule: Optional[RecurrenceRule] = None
|
|
is_starred: Optional[bool] = None
|
|
calendar_id: Optional[int] = Field(None, ge=1, le=2147483647)
|
|
# Controls which occurrences an edit applies to; absent = non-recurring or whole-series
|
|
edit_scope: Optional[Literal["this", "this_and_future"]] = None
|
|
|
|
@field_validator("recurrence_rule", mode="before")
|
|
@classmethod
|
|
def coerce_recurrence(cls, v):
|
|
return _coerce_recurrence_rule(v)
|
|
|
|
|
|
class CalendarEventResponse(BaseModel):
|
|
id: int
|
|
title: str
|
|
description: Optional[str]
|
|
start_datetime: datetime
|
|
end_datetime: datetime
|
|
all_day: bool
|
|
color: Optional[str]
|
|
location_id: Optional[int]
|
|
recurrence_rule: Optional[str] # raw JSON string from DB; frontend parses
|
|
is_starred: bool
|
|
calendar_id: int
|
|
calendar_name: str
|
|
calendar_color: str
|
|
is_virtual: bool = False
|
|
# Recurrence fields
|
|
parent_event_id: Optional[int] = None
|
|
is_recurring: bool = False
|
|
original_start: Optional[datetime] = None
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|