"""Bounded read-only data API provider for Dashboard v5 Sprint 6B.""" from __future__ import annotations import hashlib import json import math import re import sqlite3 import time from collections import defaultdict from dataclasses import dataclass from datetime import date, datetime, timedelta from pathlib import Path from typing import Any from urllib.parse import parse_qs from zoneinfo import ZoneInfo from apple_health_analytics import daily_points from dashboard_v5.data_provider import ( ADMINISTERED, SYMPTOM_LABELS, parse_day, parse_score, public_text, ) from dashboard_v5.lab_registry import EXACT_LAB_NUMBER, LAB_ALLOWLIST from dashboard_v5.metric_catalog_v2 import ( APPLE_ANALYTICS_UNITS, BY_ID_V2, LAB_CATALOG_SPECS, MetricV2, SourceInventoryLimitError, catalog_search, inventory_sources, public_catalog, ) TZ_NAME = "Europe/Zurich" LOCAL_TZ = ZoneInfo(TZ_NAME) MAX_RANGE_DAYS = 3660 MAX_SERIES_ROWS = 3660 MAX_RAW_APPLE_ROWS = 100000 MAX_RAW_LAB_ROWS = 10000 MAX_EVENT_SOURCE_ROWS = 10000 MAX_DAILY_SOURCE_ROWS = 30000 MAX_EVENT_ROWS = 1000 MAX_LAB_ROWS = 100 MAX_SEARCH_ROWS_PER_GROUP = 20 MAX_QUERY_BYTES = 512 QUERY_TIMEOUT_SECONDS = 1.0 ALLOWED_RESOLUTIONS = frozenset({"day", "week"}) ALLOWED_EVENT_TYPES = frozenset( {"medication_administered", "symptom_day", "health_event", "health_period"} ) FORBIDDEN_METADATA_TEXT = re.compile( r"(?:/home/|/tmp/|/Users/|/(?:var|etc|usr|opt|srv|root|run|mnt|media|private)/|" r"~/|file:|https?://|[A-Za-z]:[\\/]|" r"(?:^|[\s:|])/(?:[^/\s]+/)+[^/\s]+|\.\.[\\/]|\\\\[^\\/\s]+[\\/]|" r"\.hermes(?:/|$))", re.IGNORECASE, ) DRIVE_ID_TOKEN = re.compile(r"(? str: return self.code def safe_metadata_text(value: Any, maximum: int, *, allow_empty: bool = False) -> str | None: text = public_text(value, maximum, allow_empty=allow_empty) drive_id_like = bool( text and any( any(char.isalpha() for char in token) and any(char.isdigit() for char in token) for token in DRIVE_ID_TOKEN.findall(text) ) ) if text is None or FORBIDDEN_METADATA_TEXT.search(text) or drive_id_like: return None return text def api_document_id(raw_id: Any, document_day: str | None, category: str, institution: str) -> str: material = json.dumps( [int(raw_id), document_day, category, institution], ensure_ascii=False, separators=(",", ":"), ).encode("utf-8") return f"api-document-{hashlib.sha256(b'health-api-document-v1:' + material).hexdigest()[:24]}" def connect_read_only(database: Path) -> sqlite3.Connection: connection = sqlite3.connect(f"{database.resolve().as_uri()}?mode=ro", uri=True) connection.row_factory = sqlite3.Row connection.execute("PRAGMA query_only=ON") connection.execute("PRAGMA trusted_schema=OFF") deadline = time.monotonic() + QUERY_TIMEOUT_SECONDS connection.set_progress_handler(lambda: int(time.monotonic() > deadline), 1000) return connection def table_exists(connection: sqlite3.Connection, table: str) -> bool: return connection.execute( "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,) ).fetchone() is not None def parse_query(query: str, allowed: set[str]) -> dict[str, str]: if len(query.encode("utf-8")) > MAX_QUERY_BYTES: raise APIError(414, "query_too_large") try: parsed = ( parse_qs( query, keep_blank_values=True, strict_parsing=True, max_num_fields=max(1, len(allowed) + 1), ) if query else {} ) except ValueError as error: raise APIError(400, "invalid_query") from error if not set(parsed) <= allowed: raise APIError(400, "unknown_parameter") if any(len(values) != 1 for values in parsed.values()): raise APIError(400, "duplicate_parameter") return {key: values[0] for key, values in parsed.items()} def parse_iso_day(raw: str, *, code: str = "invalid_date") -> date: if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", raw): raise APIError(400, code) try: return date.fromisoformat(raw) except ValueError as error: raise APIError(400, code) from error def local_today() -> date: return datetime.now(LOCAL_TZ).date() def parse_range(params: dict[str, str], *, required: bool = False) -> tuple[date | None, date | None]: raw_start = params.get("from") raw_end = params.get("to") if bool(raw_start) != bool(raw_end): raise APIError(400, "incomplete_range") if required and not raw_start: raise APIError(400, "range_required") if not raw_start: return None, None start = parse_iso_day(raw_start) end = parse_iso_day(raw_end or "") if end < start: raise APIError(400, "range_reversed") if start > local_today() or end > local_today(): raise APIError(422, "future_range_not_allowed") if (end - start).days + 1 > MAX_RANGE_DAYS: raise APIError(422, "range_too_large") return start, end def _apple_rows( connection: sqlite3.Connection, identifier: str, start: date | None, end: date | None, *, include_raw_json: bool = False, ) -> list[sqlite3.Row]: if include_raw_json: sql = """SELECT id,metric,value,unit,start_date,end_date,source_name, file_name,file_hash,raw_json FROM apple_health_records WHERE metric=? AND start_date IS NOT NULL""" else: sql = """SELECT id,metric,value,unit,start_date,end_date,source_name,file_name,file_hash FROM apple_health_records WHERE metric=? AND start_date IS NOT NULL""" parameters: list[Any] = [identifier] if start is not None: sql += " AND substr(start_date,1,10)>=?" parameters.append((start - timedelta(days=1) if start > date.min else start).isoformat()) if end is not None: sql += " AND substr(start_date,1,10)<=?" parameters.append((end + timedelta(days=1) if end < date.max else end).isoformat()) sql += " ORDER BY start_date,end_date,source_name,id LIMIT ?" parameters.append(MAX_RAW_APPLE_ROWS + 1) rows = list(connection.execute(sql, parameters)) if len(rows) > MAX_RAW_APPLE_ROWS: raise APIError(422, "source_row_limit_exceeded") return rows def _numeric(value: float, precision: int, value_type: str) -> int | float: rounded = round(float(value), precision) if value_type == "integer" or (precision == 0 and rounded.is_integer()): return int(rounded) return rounded def _apple_points( connection: sqlite3.Connection, metric: MetricV2, start: date | None, end: date | None, ) -> list[dict[str, Any]]: rows = _apple_rows(connection, metric.source_identifier, start, end) points = daily_points( metric.source_identifier, mode=metric.aggregation, stable_only=False, start_date=start.isoformat() if start else None, end_date=end.isoformat() if end else None, _rows=rows, ) result = [] for day, point in sorted(points.items()): if point.unit != EXPECTED_ANALYTICS_UNITS.get(metric.source_identifier): continue result.append( { "date": day, "value": _numeric(point.value, metric.precision, metric.value_type), "quality": point.quality, "source_class": point.source_class, } ) return result def _blood_pressure_points( connection: sqlite3.Connection, metric: MetricV2, start: date | None, end: date | None, ) -> list[dict[str, Any]]: component = "systolic" if metric.id.endswith(".systolic") else "diastolic" by_day: dict[str, tuple[datetime, float]] = {} for row in _apple_rows(connection, "blood_pressure", start, end, include_raw_json=True): try: timestamp = datetime.fromisoformat(str(row["start_date"]).replace("Z", "+00:00")) if timestamp.tzinfo is None: timestamp = timestamp.replace(tzinfo=LOCAL_TZ) day = timestamp.astimezone(LOCAL_TZ).date() payload = json.loads(str(row["raw_json"] or "")) unit = re.sub(r"[^a-z]", "", str(row["unit"] or "").casefold()) systolic = float(payload["systolic"]) diastolic = float(payload["diastolic"]) if unit != "mmhg" or not all(math.isfinite(item) for item in (systolic, diastolic)): continue value = systolic if component == "systolic" else diastolic except (ValueError, TypeError, KeyError, json.JSONDecodeError): continue if not math.isfinite(value): continue if start is not None and day < start: continue if end is not None and day > end: continue existing = by_day.get(day.isoformat()) if existing is None or timestamp > existing[0]: by_day[day.isoformat()] = (timestamp, value) return [ {"date": day, "value": _numeric(item[1], metric.precision, metric.value_type), "quality": "paired_observation", "source_class": "apple_health"} for day, item in sorted(by_day.items()) ] def _symptom_dimension_points( connection: sqlite3.Connection, metric: MetricV2, start: date | None, end: date | None, ) -> list[dict[str, Any]]: sql = "SELECT datum,schwergrad FROM symptom_log WHERE kontext='daily_quick_score' AND symptom=?" parameters: list[Any] = [metric.source_identifier] if start: sql += " AND datum>=?" parameters.append(start.isoformat()) if end: sql += " AND datum<=?" parameters.append(end.isoformat()) sql += " ORDER BY datum,id LIMIT ?" parameters.append(MAX_DAILY_SOURCE_ROWS + 1) rows = list(connection.execute(sql, parameters)) if len(rows) > MAX_DAILY_SOURCE_ROWS: raise APIError(422, "source_row_limit_exceeded") grouped: dict[str, list[int | None]] = defaultdict(list) for row in rows: day = parse_day(row["datum"]) if day: grouped[day].append(parse_score(row["schwergrad"])) return [ {"date": day, "value": scores[0], "quality": "observed"} for day, scores in sorted(grouped.items()) if len(scores) == 1 and scores[0] is not None ] def _symptom_total_points( connection: sqlite3.Connection, start: date | None, end: date | None, ) -> list[dict[str, Any]]: sql = "SELECT datum,symptom,schwergrad FROM symptom_log WHERE kontext='daily_quick_score'" parameters: list[Any] = [] if start: sql += " AND datum>=?" parameters.append(start.isoformat()) if end: sql += " AND datum<=?" parameters.append(end.isoformat()) sql += " ORDER BY datum,id LIMIT ?" parameters.append(MAX_DAILY_SOURCE_ROWS + 1) rows = list(connection.execute(sql, parameters)) if len(rows) > MAX_DAILY_SOURCE_ROWS: raise APIError(422, "source_row_limit_exceeded") grouped: dict[str, dict[str, list[int | None]]] = defaultdict(lambda: defaultdict(list)) for row in rows: day = parse_day(row["datum"]) label = str(row["symptom"] or "") if day and label in SYMPTOM_LABELS: grouped[day][label].append(parse_score(row["schwergrad"])) result = [] for day, dimensions in sorted(grouped.items()): if set(dimensions) != SYMPTOM_LABELS: continue if not all(len(scores) == 1 and scores[0] is not None for scores in dimensions.values()): continue result.append( { "date": day, "value": sum(int(scores[0]) for scores in dimensions.values()), "quality": "complete", } ) return result def _nutrition_points( connection: sqlite3.Connection, start: date | None, end: date | None, ) -> list[dict[str, Any]]: sql = """SELECT datum,histamine_score,item_count,histamine_unknown_count,histamine_label FROM nutrition_daily_summary_v2 WHERE item_count>0 AND histamine_unknown_count=0 AND histamine_score IS NOT NULL AND histamine_label IN ('green','yellow','orange','red')""" parameters: list[Any] = [] if start: sql += " AND datum>=?" parameters.append(start.isoformat()) if end: sql += " AND datum<=?" parameters.append(end.isoformat()) sql += " ORDER BY datum LIMIT ?" parameters.append(MAX_DAILY_SOURCE_ROWS + 1) rows = list(connection.execute(sql, parameters)) if len(rows) > MAX_DAILY_SOURCE_ROWS: raise APIError(422, "source_row_limit_exceeded") result = [] for row in rows: day = parse_day(row["datum"]) try: value = float(row["histamine_score"]) except (TypeError, ValueError): continue if day and math.isfinite(value): result.append({"date": day, "value": value, "quality": "complete"}) return result def _lab_key(parameter: Any, unit: Any) -> tuple[str, str]: def slug(value: Any) -> str: normalized = str(value or "").casefold().replace("µ", "u").replace("μ", "u").replace("%", " percent ") return re.sub(r"[^a-z0-9]+", "_", normalized).strip("_") return slug(parameter), slug(unit) def _verified_lab_rows(connection: sqlite3.Connection) -> list[dict[str, Any]]: if not table_exists(connection, "laborwerte"): return [] rows = list(connection.execute( """SELECT l.id,l.parameter_name,l.wert,l.einheit,l.reference_min,l.reference_max, l.abnahme_datum,l.befund_datum,l.reference_range_source,l.source_type, l.canonical_document_id,d.document_date,d.kategorie,d.institution,d.review_status FROM laborwerte l LEFT JOIN dokumente d ON d.id=l.canonical_document_id WHERE lower(trim(COALESCE(l.validierungsstatus,'')))='validiert' AND l.verified_against_original=1 AND l.reference_range_source='scanned_original' AND trim(COALESCE(l.einheit,''))<>'' ORDER BY l.id LIMIT ?""", (MAX_RAW_LAB_ROWS + 1,), )) if len(rows) > MAX_RAW_LAB_ROWS: raise APIError(422, "source_row_limit_exceeded") grouped: dict[tuple[str, str, str], list[sqlite3.Row]] = defaultdict(list) for row in rows: canonical = LAB_ALLOWLIST.get(_lab_key(row["parameter_name"], row["einheit"])) day = parse_day(row["abnahme_datum"]) or parse_day(row["befund_datum"]) if canonical and day and date.fromisoformat(day) <= local_today(): grouped[(canonical[0], canonical[1], day)].append(row) result: list[dict[str, Any]] = [] for (parameter, unit, day), same_day in sorted(grouped.items()): if len(same_day) != 1: continue row = same_day[0] raw_value = ("" if row["wert"] is None else str(row["wert"])).strip() if not EXACT_LAB_NUMBER.fullmatch(raw_value): continue numeric = float(raw_value.replace(",", ".")) if not math.isfinite(numeric): continue value: int | float = int(numeric) if numeric.is_integer() else numeric reference_min = str(row["reference_min"] or "").strip() reference_max = str(row["reference_max"] or "").strip() reference_min = ( reference_min if EXACT_LAB_NUMBER.fullmatch(reference_min) and math.isfinite(float(reference_min.replace(",", "."))) else None ) reference_max = ( reference_max if EXACT_LAB_NUMBER.fullmatch(reference_max) and math.isfinite(float(reference_max.replace(",", "."))) else None ) if reference_min is None and reference_max is None: continue item: dict[str, Any] = { "id": f"lab-observation-{int(row['id'])}", "metric_id": LAB_CATALOG_SPECS[parameter].metric_id, "parameter": parameter, "value": value, "unit": unit, "date": day, "reference": { "min": reference_min, "max": reference_max, "source": "scanned_original", }, "quality": "verified_original", } if row["canonical_document_id"] is not None and row["review_status"] == "geprueft": category = safe_metadata_text(row["kategorie"], 80, allow_empty=True) institution = safe_metadata_text(row["institution"], 120, allow_empty=True) document_day = parse_day(row["document_date"]) if category is not None and institution is not None: item["document"] = { "id": api_document_id( row["canonical_document_id"], document_day, category, institution ), "date": document_day, "category": category, "institution": institution, } result.append(item) return result def _metric_points( connection: sqlite3.Connection, metric: MetricV2, start: date | None, end: date | None, ) -> list[dict[str, Any]]: if metric.value_type == "panel": raise APIError(400, "metric_not_series") if metric.source == "apple_health": if metric.source_identifier == "blood_pressure": return _blood_pressure_points(connection, metric, start, end) return _apple_points(connection, metric, start, end) if metric.id == "symptom.total": return _symptom_total_points(connection, start, end) if metric.id == "symptom.aphthae": return _symptom_dimension_points(connection, metric, start, end) if metric.id == "nutrition.histamine": return _nutrition_points(connection, start, end) if metric.source == "laborwerte": return [ {"date": row["date"], "value": row["value"], "quality": row["quality"]} for row in _verified_lab_rows(connection) if row["metric_id"] == metric.id and (start is None or row["date"] >= start.isoformat()) and (end is None or row["date"] <= end.isoformat()) ] return [] def _weekly(points: list[dict[str, Any]], metric: MetricV2) -> list[dict[str, Any]]: grouped: dict[tuple[int, int], list[dict[str, Any]]] = defaultdict(list) for point in points: day = date.fromisoformat(point["date"]) iso = day.isocalendar() grouped[(iso.year, iso.week)].append(point) result = [] for (year, week), rows in sorted(grouped.items()): distinct = {row["date"] for row in rows} if len(distinct) != 7: continue values = [float(row["value"]) for row in rows] if metric.aggregation == "sum": value = sum(values) elif metric.aggregation == "last": value = float(max(rows, key=lambda row: row["date"])["value"]) else: value = sum(values) / len(values) result.append( { "week": f"{year}-W{week:02d}", "date": min(distinct), "value": _numeric(value, metric.precision, metric.value_type), "complete_days": 7, "quality": "complete_calendar_week", } ) return result def _coverage(points: list[dict[str, Any]], start: date | None, end: date | None) -> dict[str, Any]: observed = sorted({date.fromisoformat(point["date"]) for point in points}) effective_start = start or (observed[0] if observed else None) effective_end = end or (observed[-1] if observed else None) if effective_start is not None and effective_end is not None: span = (effective_end - effective_start).days + 1 if span > MAX_RANGE_DAYS: raise APIError(422, "range_too_large") else: span = 0 return { "from": effective_start.isoformat() if effective_start else None, "to": effective_end.isoformat() if effective_end else None, "expected_days": span, "observed_days": len(observed), "missing_days": max(0, span - len(observed)), "gaps": "omitted_not_interpolated", } def _series(connection: sqlite3.Connection, params: dict[str, str]) -> dict[str, Any]: metric_id = params.get("metric", "") metric = BY_ID_V2.get(metric_id) if metric is None: raise APIError(404, "metric_not_allowed") resolution = params.get("resolution", "day") if resolution not in ALLOWED_RESOLUTIONS: raise APIError(400, "resolution_not_allowed") start, end = parse_range(params) points = _metric_points(connection, metric, start, end) today = local_today().isoformat() points = [point for point in points if point["date"] <= today] if start is None and points: observed_start = date.fromisoformat(points[0]["date"]) observed_end = date.fromisoformat(points[-1]["date"]) if (observed_end - observed_start).days + 1 > MAX_RANGE_DAYS: raise APIError(422, "range_too_large") coverage = _coverage(points, start, end) if resolution == "week": if metric.aggregation in {"observation", "panel"}: raise APIError(400, "resolution_not_supported") points = _weekly(points, metric) if len(points) > MAX_SERIES_ROWS: raise APIError(422, "row_limit_exceeded") return { "metric": metric.id, "label": metric.label, "unit": metric.unit, "aggregation": metric.aggregation, "resolution": resolution, "source": { "type": metric.source, "identifier": metric.source_identifier, "parser": metric.parser_contract, }, "coverage": coverage, "timezone": TZ_NAME, "points": points, } def _events(connection: sqlite3.Connection, params: dict[str, str]) -> dict[str, Any]: start, end = parse_range(params, required=True) assert start is not None and end is not None raw_types = params.get("types", "") if not raw_types: selected = set(ALLOWED_EVENT_TYPES) else: selected = set(raw_types.split(",")) if "" in selected or not selected <= ALLOWED_EVENT_TYPES: raise APIError(400, "event_type_not_allowed") result: list[dict[str, Any]] = [] if "medication_administered" in selected and table_exists(connection, "medication_administrations"): administered_values = tuple(sorted(ADMINISTERED)) if len(administered_values) != 6: raise APIError(503, "api_contract_invalid") rows = list(connection.execute( """SELECT datum,medication_name,event_type FROM medication_administrations WHERE datum>=? AND datum<=? AND lower(trim(COALESCE(event_type,''))) IN (?,?,?,?,?,?) ORDER BY datum,id LIMIT ?""", (start.isoformat(), end.isoformat(), *administered_values, MAX_EVENT_SOURCE_ROWS + 1), )) if len(rows) > MAX_EVENT_SOURCE_ROWS: raise APIError(422, "source_row_limit_exceeded") for row in rows: label = safe_metadata_text(row["medication_name"], 120) day = parse_day(row["datum"]) if day and label: result.append({"date": day, "type": "medication_administered", "label": label}) if "symptom_day" in selected and table_exists(connection, "symptom_log"): for point in _symptom_total_points(connection, start, end): result.append({"date": point["date"], "type": "symptom_day", "label": "Symptome vollständig dokumentiert"}) if "health_event" in selected and table_exists(connection, "health_events"): rows = list(connection.execute( "SELECT date,category,parameter FROM health_events WHERE date>=? AND date<=? ORDER BY date,id LIMIT ?", (start.isoformat(), end.isoformat(), MAX_EVENT_SOURCE_ROWS + 1), )) if len(rows) > MAX_EVENT_SOURCE_ROWS: raise APIError(422, "source_row_limit_exceeded") for row in rows: category = safe_metadata_text(row["category"], 80, allow_empty=True) parameter = safe_metadata_text(row["parameter"], 120, allow_empty=True) day = parse_day(row["date"]) if day and category is not None and parameter is not None: result.append({"date": day, "type": "health_event", "category": category, "label": parameter}) if "health_period" in selected and table_exists(connection, "health_event_periods"): rows = list(connection.execute( """SELECT start_date,end_date,event_type,label FROM health_event_periods WHERE start_date<=? AND COALESCE(end_date,start_date)>=? ORDER BY start_date,id LIMIT ?""", (end.isoformat(), start.isoformat(), MAX_EVENT_SOURCE_ROWS + 1), )) if len(rows) > MAX_EVENT_SOURCE_ROWS: raise APIError(422, "source_row_limit_exceeded") for row in rows: event_type = safe_metadata_text(row["event_type"], 80) label = safe_metadata_text(row["label"], 120, allow_empty=True) start_day = parse_day(row["start_date"]) end_day = parse_day(row["end_date"]) if row["end_date"] else None if ( start_day is None or (end_day is not None and end_day < start_day) or (end_day is not None and date.fromisoformat(end_day) > local_today()) ): continue if event_type and label is not None: result.append({"date": start_day, "end_date": end_day, "type": "health_period", "category": event_type, "label": label}) result.sort(key=lambda item: (item["date"], item["type"], item.get("label", ""))) if len(result) > MAX_EVENT_ROWS: raise APIError(422, "row_limit_exceeded") return { "from": start.isoformat(), "to": end.isoformat(), "types": sorted(selected), "timezone": TZ_NAME, "events": result, } def _safe_documents(connection: sqlite3.Connection, needle: str) -> list[dict[str, Any]]: if not table_exists(connection, "dokumente"): return [] result = [] for row in connection.execute( """SELECT id,document_date,kategorie,institution FROM dokumente WHERE review_status='geprueft' ORDER BY document_date DESC,id DESC LIMIT 200""" ): category = safe_metadata_text(row["kategorie"], 80, allow_empty=True) institution = safe_metadata_text(row["institution"], 120, allow_empty=True) document_day = parse_day(row["document_date"]) if category is None or institution is None: continue haystack = f"{category} {institution} {document_day or ''}".casefold() if needle.casefold() not in haystack: continue result.append( { "id": api_document_id(row["id"], document_day, category, institution), "date": document_day, "category": category, "institution": institution, "drill_down_target": "document_metadata", } ) if len(result) >= MAX_SEARCH_ROWS_PER_GROUP: break return result def _search(connection: sqlite3.Connection, query: str) -> dict[str, Any]: clean = " ".join(query.split()) if len(clean) < 2 or len(clean) > 80: raise APIError(400, "query_length_invalid") matches = catalog_search(connection, clean) groups: dict[str, list[dict[str, Any]]] = { "metrics": [], "laboratory": [], "symptoms_events": [], "documents": [], "days": [], } for item in matches: target = "laboratory" if item["category"] == "laboratory" else ( "symptoms_events" if item["category"] == "symptoms" else "metrics" ) if len(groups[target]) < MAX_SEARCH_ROWS_PER_GROUP: groups[target].append(item) normalized = clean.casefold() if normalized in {"medikamente", "medikament", "medikation", "arznei"}: groups["symptoms_events"].append( { "id": "event.medication", "label": "Medikamente", "category": "events", "drill_down_target": "day", "availability": {"status": "supported"}, } ) groups["documents"] = _safe_documents(connection, clean) try: parsed = parse_iso_day(clean) except APIError: parsed = None if parsed is not None and parsed <= local_today(): groups["days"] = [{"date": parsed.isoformat(), "drill_down_target": "day"}] return {"query": clean, "timezone": TZ_NAME, "groups": groups} def _day(connection: sqlite3.Connection, day: date) -> dict[str, Any]: metrics: dict[str, dict[str, Any]] = {} for metric in BY_ID_V2.values(): if metric.value_type == "panel": continue points = _metric_points(connection, metric, day, day) if not points: continue point = points[-1] metrics[metric.id] = { "value": point["value"], "unit": metric.unit, "aggregation": metric.aggregation, "quality": point["quality"], } events = _events( connection, {"from": day.isoformat(), "to": day.isoformat(), "types": ""}, )["events"] labs = [row for row in _verified_lab_rows(connection) if row["date"] == day.isoformat()] return { "date": day.isoformat(), "timezone": TZ_NAME, "metrics": metrics, "events": events, "labs": labs, } def dispatch_api(database: Path, path: str, query: str) -> dict[str, Any]: if path not in API_PATHS and not re.fullmatch(r"/api/v1/day/\d{4}-\d{2}-\d{2}", path): raise APIError(404, "endpoint_not_found") connection = connect_read_only(database) try: if path == "/api/v1/metric-catalog": params = parse_query(query, {"q"}) query_text = params.get("q", "") metrics = catalog_search(connection, query_text) if query_text else public_catalog(connection) inventory = inventory_sources(connection) return { "version": 2, "timezone": TZ_NAME, "groups": {"metrics": metrics}, "inventory": { "apple_observed_identifiers": len(inventory["apple_health"]), "apple_released_identifiers": sum(item["status"] == "released" for item in inventory["apple_health"]), "canonical_verified_laboratory_parameters": len(inventory["laboratory"]), "observed_vital_sign_parameters": len(inventory["vital_signs"]), }, } if path == "/api/v1/series": params = parse_query(query, {"metric", "from", "to", "resolution"}) if not params.get("metric"): raise APIError(400, "metric_required") return _series(connection, params) if path.startswith("/api/v1/day/"): if query: raise APIError(400, "unknown_parameter") requested_day = parse_iso_day(path.rsplit("/", 1)[-1]) if requested_day > local_today(): raise APIError(422, "future_date_not_allowed") return _day(connection, requested_day) if path == "/api/v1/events": params = parse_query(query, {"from", "to", "types"}) return _events(connection, params) if path == "/api/v1/labs": parse_query(query, set()) rows = _verified_lab_rows(connection) if len(rows) > MAX_LAB_ROWS: raise APIError(422, "row_limit_exceeded") return { "timezone": TZ_NAME, "reference_policy": "observation_specific_verified_original", "observations": rows, } if path == "/api/v1/search": params = parse_query(query, {"q"}) return _search(connection, params.get("q", "")) raise APIError(404, "endpoint_not_found") except SourceInventoryLimitError as error: raise APIError(422, "source_inventory_limit_exceeded") from error except sqlite3.Error as error: raise APIError(503, "data_unavailable") from error finally: connection.close()