UMBRA/backend/app/schemas/settings.py
Kyle Pope 67456c78dd Implement Track C: NTFY push notification integration
- Add ntfy columns to Settings model (server_url, topic, auth_token, enabled, per-type toggles, lead times)
- Create NtfySent dedup model to prevent duplicate notifications
- Create ntfy service with SSRF validation and async httpx send
- Create ntfy_templates service with per-type payload builders
- Create APScheduler background dispatch job (60s interval, events/reminders/todos/projects)
- Register scheduler in main.py lifespan with max_instances=1
- Update SettingsUpdate with ntfy validators (URL scheme, topic regex, lead time ranges)
- Update SettingsResponse with ntfy fields; ntfy_has_token computed, token never exposed
- Add POST /api/settings/ntfy/test endpoint
- Update GET/PUT settings to use explicit _to_settings_response() helper
- Add Alembic migration 022 for ntfy settings columns + ntfy_sent table
- Add httpx==0.27.2 and apscheduler==3.10.4 to requirements.txt

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-25 04:04:23 +08:00

167 lines
5.6 KiB
Python

import re
from pydantic import BaseModel, ConfigDict, field_validator
from datetime import datetime
from typing import Literal, Optional
AccentColor = Literal["cyan", "blue", "green", "purple", "red", "orange", "pink", "yellow"]
_NTFY_TOPIC_RE = re.compile(r'^[a-zA-Z0-9_-]{1,64}$')
def _validate_pin_length(v: str, label: str = "PIN") -> str:
if len(v) < 4:
raise ValueError(f'{label} must be at least 4 characters')
if len(v) > 72:
raise ValueError(f'{label} must be at most 72 characters')
return v
class SettingsCreate(BaseModel):
pin: str
@field_validator('pin')
@classmethod
def pin_length(cls, v: str) -> str:
return _validate_pin_length(v)
class SettingsUpdate(BaseModel):
accent_color: Optional[AccentColor] = None
upcoming_days: int | None = None
preferred_name: str | None = None
weather_city: str | None = None
weather_lat: float | None = None
weather_lon: float | None = None
first_day_of_week: int | None = None
# ntfy configuration fields
ntfy_server_url: Optional[str] = None
ntfy_topic: Optional[str] = None
# Empty string means "clear the token"; None means "leave unchanged"
ntfy_auth_token: Optional[str] = None
ntfy_enabled: Optional[bool] = None
ntfy_events_enabled: Optional[bool] = None
ntfy_reminders_enabled: Optional[bool] = None
ntfy_todos_enabled: Optional[bool] = None
ntfy_projects_enabled: Optional[bool] = None
ntfy_event_lead_minutes: Optional[int] = None
ntfy_todo_lead_days: Optional[int] = None
ntfy_project_lead_days: Optional[int] = None
@field_validator('first_day_of_week')
@classmethod
def validate_first_day(cls, v: int | None) -> int | None:
if v is not None and v not in (0, 1):
raise ValueError('first_day_of_week must be 0 (Sunday) or 1 (Monday)')
return v
@field_validator('weather_lat')
@classmethod
def validate_lat(cls, v: float | None) -> float | None:
if v is not None and (v < -90 or v > 90):
raise ValueError('Latitude must be between -90 and 90')
return v
@field_validator('weather_lon')
@classmethod
def validate_lon(cls, v: float | None) -> float | None:
if v is not None and (v < -180 or v > 180):
raise ValueError('Longitude must be between -180 and 180')
return v
@field_validator('ntfy_server_url')
@classmethod
def validate_ntfy_url(cls, v: Optional[str]) -> Optional[str]:
if v is None or v == "":
return None
from urllib.parse import urlparse
parsed = urlparse(v)
if parsed.scheme not in ("http", "https"):
raise ValueError("ntfy server URL must use http or https")
if not parsed.netloc:
raise ValueError("ntfy server URL must include a hostname")
# Strip trailing slash — ntfy base URL must not have one
return v.rstrip("/")
@field_validator('ntfy_topic')
@classmethod
def validate_ntfy_topic(cls, v: Optional[str]) -> Optional[str]:
if v is None or v == "":
return None
if not _NTFY_TOPIC_RE.match(v):
raise ValueError(
"ntfy topic must be 1-64 alphanumeric characters, hyphens, or underscores"
)
return v
@field_validator('ntfy_auth_token')
@classmethod
def validate_ntfy_token(cls, v: Optional[str]) -> Optional[str]:
# Empty string signals "clear the token" — normalise to None for storage
if v == "":
return None
if v is not None and len(v) > 500:
raise ValueError("ntfy auth token must be at most 500 characters")
return v
@field_validator('ntfy_event_lead_minutes')
@classmethod
def validate_event_lead(cls, v: Optional[int]) -> Optional[int]:
if v is not None and not (1 <= v <= 1440):
raise ValueError("ntfy_event_lead_minutes must be between 1 and 1440")
return v
@field_validator('ntfy_todo_lead_days')
@classmethod
def validate_todo_lead(cls, v: Optional[int]) -> Optional[int]:
if v is not None and not (0 <= v <= 30):
raise ValueError("ntfy_todo_lead_days must be between 0 and 30")
return v
@field_validator('ntfy_project_lead_days')
@classmethod
def validate_project_lead(cls, v: Optional[int]) -> Optional[int]:
if v is not None and not (0 <= v <= 30):
raise ValueError("ntfy_project_lead_days must be between 0 and 30")
return v
class SettingsResponse(BaseModel):
id: int
accent_color: str
upcoming_days: int
preferred_name: str | None = None
weather_city: str | None = None
weather_lat: float | None = None
weather_lon: float | None = None
first_day_of_week: int = 0
# ntfy fields — ntfy_auth_token is NEVER included here (security requirement 6.2)
ntfy_server_url: Optional[str] = None
ntfy_topic: Optional[str] = None
ntfy_enabled: bool = False
ntfy_events_enabled: bool = True
ntfy_reminders_enabled: bool = True
ntfy_todos_enabled: bool = True
ntfy_projects_enabled: bool = True
ntfy_event_lead_minutes: int = 15
ntfy_todo_lead_days: int = 1
ntfy_project_lead_days: int = 2
# Derived field: True if a token is stored, never exposes the value itself
ntfy_has_token: bool = False
created_at: datetime
updated_at: datetime
model_config = ConfigDict(from_attributes=True)
class ChangePinRequest(BaseModel):
old_pin: str
new_pin: str
@field_validator('new_pin')
@classmethod
def new_pin_length(cls, v: str) -> str:
return _validate_pin_length(v, "New PIN")