70 lines
3.7 KiB
Python
70 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, List, Optional
|
|
from pydantic import BaseModel, Field, model_validator
|
|
|
|
class QueryRequest(BaseModel):
|
|
query: str = Field(..., description="Read-only SQL query to execute in DuckDB")
|
|
params: Optional[List[Any]] = Field(default=None, description="Positional query parameters")
|
|
|
|
class QueryResponse(BaseModel):
|
|
columns: List[str]
|
|
rows: List[Dict[str, Any]]
|
|
count: int
|
|
execution_time_ms: float
|
|
|
|
class EventCreateRequest(BaseModel):
|
|
kind: str = Field(..., description="Event kind, e.g. sleep.session, nutrition.intake, diaper.change, feeding.bottle, growth.measurement")
|
|
subject: str = Field(..., description="Event subject, e.g. mila or moritz")
|
|
payload: Dict[str, Any] = Field(default_factory=dict, description="Event-specific payload")
|
|
actor: Optional[str] = Field(default="web-app", description="Actor who logged the event")
|
|
date: Optional[str] = Field(default=None, description="Explicit log date override YYYY-MM-DD")
|
|
start: Optional[str] = Field(default=None, description="ISO-8601 start timestamp")
|
|
end: Optional[str] = Field(default=None, description="ISO-8601 end timestamp (optional)")
|
|
since: Optional[str] = Field(default=None, description="Relative start duration like '15m', '1h'")
|
|
estimated: bool = Field(default=False, description="Whether intake is estimated")
|
|
meal: Optional[str] = Field(default=None, description="Meal category, e.g. fruehstueck, mittagessen, abendessen, snack")
|
|
note: Optional[str] = Field(default=None, description="Free-text note")
|
|
source_ref: Optional[str] = Field(default=None, description="Source reference")
|
|
allow_duplicate: bool = Field(default=False, description="Allow event within 15min of duplicate")
|
|
|
|
class EventStopRequest(BaseModel):
|
|
kind: str = Field(default="sleep.session", description="Event kind to stop")
|
|
subject: str = Field(..., description="Event subject, e.g. mila")
|
|
end_time: Optional[str] = Field(default=None, description="ISO-8601 end timestamp")
|
|
since: Optional[str] = Field(default=None, description="Relative end duration like '5m'")
|
|
event_id: Optional[str] = Field(default=None, description="Specific event ID to close")
|
|
actor: Optional[str] = Field(default="web-app", description="Actor who stopped the event")
|
|
|
|
class EntityPutRequest(BaseModel):
|
|
collection: str = Field(..., description="Target collection: foods, recipes, places, people")
|
|
record: Dict[str, Any] = Field(..., description="Entity record with 'id'")
|
|
|
|
class GarminSyncRequest(BaseModel):
|
|
daily: List[Dict[str, Any]] = Field(default_factory=list, description="Garmin daily summary records")
|
|
sleep: List[Dict[str, Any]] = Field(default_factory=list, description="Garmin sleep records")
|
|
activities: List[Dict[str, Any]] = Field(default_factory=list, description="Garmin activity records")
|
|
daily_summary: Optional[Dict[str, Any]] = None
|
|
sleep_data: Optional[Dict[str, Any]] = None
|
|
|
|
@model_validator(mode="before")
|
|
@classmethod
|
|
def normalize_aliases(cls, data: Any) -> Any:
|
|
if isinstance(data, dict):
|
|
# Normalize daily_summary dict to daily list if daily is missing/empty
|
|
if "daily_summary" in data and data["daily_summary"] and not data.get("daily"):
|
|
data["daily"] = [data["daily_summary"]]
|
|
# Normalize sleep_data dict to sleep list if sleep is missing/empty
|
|
if "sleep_data" in data and data["sleep_data"] and not data.get("sleep"):
|
|
data["sleep"] = [data["sleep_data"]]
|
|
return data
|
|
|
|
class StatusResponse(BaseModel):
|
|
status: str
|
|
git_conflict: bool
|
|
last_sync_time: Optional[str]
|
|
last_commit_hash: Optional[str]
|
|
last_error: Optional[str]
|
|
data_dir: str
|
|
table_counts: Dict[str, int]
|