ericwyuan/GarminHealthLab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(api): 个人资料/单位/同步偏好 + 运动详情 + 身体年龄

Browse Source 
设置 (services/settings.py, routes/settings.py)
- user_settings 表：身高/体重/出生日期/性别/单位/自动同步开关/同步频率/历史范围
- GET|PUT /api/settings，GET /api/settings/options（取值由后端给，前端不臆造）
- GET /api/settings/rating-basis：把每条参考区间的来源公开出来。
  一个把数字标成「偏低」的区间是在下判断，用户有权看到依据。

运动详情 (services/garmin.py)
- GET /api/garmin/activities/<id>/detail：概览/分段/心率区间/天气/装备/采样曲线
- 首次打开回源 Garmin 并落库，之后走缓存；?refresh=1 强制刷新
- 采样点在写入时抽稀到 300，手机图表画不了更多，也免得整行撑大

身体年龄 (services/fitness_age.py)
- 0.2.8 版 garminconnect 没有 fitnessage 接口，改为本地按公开常模推算：
  VO₂max 对应年龄为基准，静息心率与 BMI 做有上限的修正
- 返回每一步的中间值，界面照实展示，不做成一个不可追溯的分数
- 高于参考表最年轻一档时按 20 岁计——那里外推会得到「11 岁」这种结果

调度器改为每 5 分钟 tick，是否该同步按各账号自己的频率判断

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
ericwyuan
2026-08-24 00:26:26 +08:00
parent 12ef5ca06b
commit c70e7ced80
9 changed files with 789 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

177
backend/services/fitness_age.py Normal file
View File

@@ -0,0 +1,177 @@
"""
身体年龄 (body age) — a deterministic estimate, with its working exposed.
This is NOT Garmin's Fitness Age. Garmin's model is proprietary and cannot be
reproduced; asking a language model to invent a number would produce something
unverifiable that changes between runs while looking authoritative. So the
estimate here is computed from published population reference values, and every
step it took is returned alongside the number for the UI to display.
Method
------
1. Base age from VO2max: the age at which the user's VO2max equals the median
for their sex, interpolated over the reference table below. VO2max is the
single strongest fitness predictor and is what Garmin's own model leans on.
2. Resting-heart-rate adjustment, relative to a 60 bpm reference.
3. BMI adjustment, relative to the healthy 18.5–24.9 band.
4. Clamped to within 20 years of chronological age — beyond that the
extrapolation says more about the table's edges than about the person.
Reference values are 50th-percentile VO2max (ml/kg/min) by age and sex, from
the widely published ACSM / Cooper Institute cardiorespiratory fitness norms.
They are population averages for healthy adults, not clinical thresholds.
"""
# (age, median VO2max) — men and women tabulated separately because the
# distributions differ by roughly 6–8 ml/kg/min at every age.
VO2_MEDIAN = {
"male": [(25, 44.0), (35, 41.0), (45, 37.0), (55, 33.0), (65, 29.0)],
"female": [(25, 37.0), (35, 34.0), (45, 31.0), (55, 27.0), (65, 24.0)],
}
RHR_REFERENCE = 60.0 # bpm
RHR_YEARS_PER_10BPM = 2.0
RHR_CAP = 5.0
BMI_LOW, BMI_HIGH = 18.5, 24.9
BMI_YEARS_PER_UNIT = 0.5
BMI_CAP = 5.0
MAX_DEVIATION = 20.0 # years either side of chronological age
AGE_FLOOR, AGE_CEILING = 20.0, 85.0
# Rendered verbatim in 设置 → 评分依据. Kept here, next to the constants it
# describes, so the two cannot drift apart.
BASIS = {
"title": "身体年龄的算法",
"summary": (
"由你的 VO₂max、静息心率、BMI 按公开人群参考值推算，"
"不是 Garmin 的 Fitness Age，也不是医学评估。"
),
"steps": [
{
"name": "基准：VO₂max 对应年龄",
"detail": "找出你的 VO₂max 相当于同性别人群哪个年龄的中位水平，"
"在参考表上线性插值；高于最年轻一档时按 20 岁计，"
"参考表再往上说明不了更多。",
"source": "ACSM / Cooper Institute 心肺适能人群常模（50 百分位）",
},
{
"name": "静息心率修正",
"detail": f"以 {RHR_REFERENCE:.0f} bpm 为参照，"
f"每高 10 bpm +{RHR_YEARS_PER_10BPM:.0f} 岁，"
f"每低 10 bpm −{RHR_YEARS_PER_10BPM:.0f} 岁，"
f"最多 ±{RHR_CAP:.0f} 岁。",
"source": "静息心率与心肺适能、全因死亡率的流行病学关联",
},
{
"name": "BMI 修正",
"detail": f"BMI 在 {BMI_LOW}~{BMI_HIGH} 之间不修正；"
f"每偏离 1 +{BMI_YEARS_PER_UNIT} 岁，最多 +{BMI_CAP:.0f} 岁。",
"source": "WHO 成人 BMI 分类",
},
{
"name": "收敛",
"detail": f"结果限制在实际年龄 ±{MAX_DEVIATION:.0f} 岁以内，"
f"并落在 {AGE_FLOOR:.0f}~{AGE_CEILING:.0f} 岁区间。",
"source": "参考表边界外的外推不可靠",
},
],
"caveat": "仅供长期趋势参考，不能用于诊断。有健康疑问请咨询医生。",
}
def _interpolate_age(vo2, table):
"""The age whose median VO2max equals `vo2`.
Above the youngest reference row the answer is simply "fitter than the
median 25-year-old", and the table cannot say more: extrapolating its slope
there gives absurdities (VO2max 48 reads as an eleven-year-old), so the
result floors instead.
"""
first_age, first_vo2 = table[0]
last_age, last_vo2 = table[-1]
if vo2 >= first_vo2:
return AGE_FLOOR
if vo2 <= last_vo2:
slope = (last_age - table[-2][0]) / (last_vo2 - table[-2][1])
return last_age + (vo2 - last_vo2) * slope
for (age_a, vo2_a), (age_b, vo2_b) in zip(table, table[1:]):
if vo2_b <= vo2 <= vo2_a:
share = (vo2_a - vo2) / (vo2_a - vo2_b)
return age_a + share * (age_b - age_a)
return last_age
def estimate(*, age, sex, vo2max, resting_hr=None, bmi=None):
"""Body age plus the arithmetic that produced it.
Returns None when the inputs cannot support an estimate, so the caller can
tell the user what is missing instead of showing a fabricated number.
"""
missing = []
if age is None:
missing.append("出生日期")
if sex not in VO2_MEDIAN:
missing.append("性别")
if not vo2max:
missing.append("VO₂max（需要一次户外跑步或骑行才会生成）")
if missing:
return {"value": None, "missing": missing, "basis": BASIS}
table = VO2_MEDIAN[sex]
base = _interpolate_age(float(vo2max), table)
steps = [{
"label": "VO₂max 基准",
"input": f"{float(vo2max):.0f} ml/kg/min",
"years": round(base, 1),
"kind": "base",
}]
total = base
if resting_hr:
delta = (float(resting_hr) - RHR_REFERENCE) / 10.0 * RHR_YEARS_PER_10BPM
delta = max(-RHR_CAP, min(RHR_CAP, delta))
total += delta
steps.append({
"label": "静息心率",
"input": f"{float(resting_hr):.0f} bpm",
"years": round(delta, 1),
"kind": "adjust",
})
if bmi:
value = float(bmi)
if value < BMI_LOW:
off = BMI_LOW - value
elif value > BMI_HIGH:
off = value - BMI_HIGH
else:
off = 0.0
delta = min(BMI_CAP, off * BMI_YEARS_PER_UNIT)
total += delta
steps.append({
"label": "BMI",
"input": f"{value:.1f}",
"years": round(delta, 1),
"kind": "adjust",
})
chronological = float(age)
clamped = max(chronological - MAX_DEVIATION,
min(chronological + MAX_DEVIATION, total))
clamped = max(AGE_FLOOR, min(AGE_CEILING, clamped))
value = int(round(clamped))
return {
"value": value,
"chronologicalAge": int(chronological),
"delta": value - int(chronological),
"steps": steps,
"clamped": abs(clamped - total) > 0.05,
"missing": [],
"basis": BASIS,
}

167
backend/services/garmin.py
View File

@@ -20,6 +20,7 @@ The last two are easy to confuse: `get_activities` takes an offset and a count,
so passing it a date silently asks for activity number "2026-08-23".
"""
import datetime
import json
import os
import threading
@@ -415,6 +416,172 @@ def _sync_activities(client, user_id, start_date, end_date):
return stored
# --- one activity, in full ---------------------------------------------------
# Garmin will return thousands of samples per activity. A phone chart cannot
# draw more than a few hundred usefully, and the payload is stored as a row, so
# the series are thinned on the way in rather than on every read.
DETAIL_MAX_POINTS = 300
# Descriptor key -> the name the UI charts by. Anything not listed is dropped:
# the full descriptor set runs to dozens of fields, most of them empty.
SERIES_KEYS = {
"directTimestamp": "timestamp",
"sumElapsedDuration": "elapsed",
"sumDuration": "duration",
"sumDistance": "distance",
"directHeartRate": "heartRate",
"directSpeed": "speed",
"directElevation": "elevation",
"directRunCadence": "cadence",
"directBikeCadence": "cadence",
"directDoubleCadence": "cadence",
"directPower": "power",
"directAirTemperature": "temperature",
}
def _thin(values, limit=DETAIL_MAX_POINTS):
"""Evenly sample a list down to `limit` points, keeping first and last."""
if len(values) <= limit:
return values
step = (len(values) - 1) / (limit - 1)
return [values[int(round(i * step))] for i in range(limit)]
def _series_from_details(details):
"""Turn Garmin's column-store detail payload into per-metric arrays.
The response is a descriptor list plus rows of parallel values, so every
metric has to be read out by the index its descriptor names.
"""
descriptors = details.get("metricDescriptors") or []
rows = details.get("activityDetailMetrics") or []
if not descriptors or not rows:
return {}
index = {}
for d in descriptors:
name = SERIES_KEYS.get(d.get("key"))
if name and name not in index:
index[name] = d.get("metricsIndex")
rows = _thin(rows)
out = {}
for name, position in index.items():
if position is None:
continue
column = []
for row in rows:
metrics = row.get("metrics") or []
column.append(metrics[position] if position < len(metrics) else None)
# A column of nothing but nulls is a sensor the watch does not have.
if any(v is not None for v in column):
out[name] = column
return out
def _lap_rows(splits):
laps = []
for i, lap in enumerate((splits or {}).get("lapDTOs") or [], start=1):
laps.append({
"index": lap.get("lapIndex") or i,
"duration": _num(lap.get("duration")),
"movingDuration": _num(lap.get("movingDuration")),
"distance": _num(lap.get("distance")),
"averageSpeed": _num(lap.get("averageSpeed")),
"maxSpeed": _num(lap.get("maxSpeed")),
"calories": _num(lap.get("calories")),
"averageHR": _num(lap.get("averageHR")),
"maxHR": _num(lap.get("maxHR")),
"elevationGain": _num(lap.get("elevationGain")),
"elevationLoss": _num(lap.get("elevationLoss")),
})
return laps
def _hr_zones(zones):
out = []
for z in zones or []:
out.append({
"zone": z.get("zoneNumber"),
"seconds": _num(z.get("secsInZone")) or 0,
"lowBoundary": _num(z.get("zoneLowBoundary")),
})
return sorted(out, key=lambda z: z.get("zone") or 0)
def _build_detail(client, activity_id):
"""Assemble everything Garmin knows about one activity.
Each call is wrapped: a watch without a barometer has no weather, a
treadmill run has no gear, and a missing optional endpoint must leave the
rest of the page intact rather than fail the request.
"""
summary = _safe(lambda: client.get_activity_evaluation(activity_id), {}) or {}
details = _safe(
lambda: client.get_activity_details(activity_id, maxchart=2000, maxpoly=0), {}
) or {}
return {
"activityId": str(activity_id),
"summary": summary.get("summaryDTO") or {},
"activityName": summary.get("activityName"),
"activityType": (summary.get("activityTypeDTO") or {}).get("typeKey"),
"eventType": (summary.get("eventTypeDTO") or {}).get("typeKey"),
"laps": _lap_rows(_safe(lambda: client.get_activity_splits(activity_id), {})),
"hrZones": _hr_zones(
_safe(lambda: client.get_activity_hr_in_timezones(activity_id), [])
),
"weather": _safe(lambda: client.get_activity_weather(activity_id), {}) or {},
"gear": _safe(lambda: client.get_activity_gear(activity_id), []) or [],
"exerciseSets": (
_safe(lambda: client.get_activity_exercise_sets(activity_id), {}) or {}
).get("exerciseSets") or [],
"series": _series_from_details(details),
}
def get_activity_detail(user_id, activity_id, creds=None, refresh=False):
"""Cached detail for one activity, fetched from Garmin on first open."""
activity_id = str(activity_id)
if not refresh:
row = query_one(
"SELECT payload FROM activity_details "
"WHERE user_id = ? AND activity_id = ?",
[user_id, activity_id],
)
if row and row.get("payload"):
try:
cached = json.loads(row["payload"])
cached["cached"] = True
return cached
except ValueError:
# A truncated row is worth refetching, not worth crashing on.
pass
client = _connect(creds or {}, user_id=user_id)
detail = _build_detail(client, activity_id)
cols = ["activity_id", "user_id", "payload", "fetched_at"]
values = [activity_id, user_id, json.dumps(detail, default=str),
datetime.datetime.utcnow().isoformat(timespec="seconds")]
placeholders = ", ".join(["?"] * len(cols))
if DB_TYPE == "mariadb":
updates = ", ".join(f"{c}=VALUES({c})" for c in cols if c != "activity_id")
sql = (f"INSERT INTO activity_details ({', '.join(cols)}) "
f"VALUES ({placeholders}) ON DUPLICATE KEY UPDATE {updates}")
else:
updates = ", ".join(f"{c}=excluded.{c}" for c in cols if c != "activity_id")
sql = (f"INSERT INTO activity_details ({', '.join(cols)}) VALUES "
f"({placeholders}) ON CONFLICT(activity_id) DO UPDATE SET {updates}")
execute(sql, values)
detail["cached"] = False
return detail
# Above this many days a sync is long enough that the caller must not block
# on it — a year takes roughly 20 minutes at ~3s per day.
BACKGROUND_THRESHOLD_DAYS = 14

66
backend/services/scheduler.py
View File

@@ -23,6 +23,7 @@ import time
from config import DB_TYPE
from db import execute, query_one, query_all
from services import garmin as garmin_svc
from services import settings as settings_svc
JOB_NAME = "garmin_auto_sync"
@@ -32,6 +33,11 @@ INTERVAL_SECONDS = int(os.environ.get("AUTO_SYNC_INTERVAL_SECONDS") or 3600)
SYNC_DAYS = int(os.environ.get("AUTO_SYNC_DAYS") or 2)
ENABLED = (os.environ.get("AUTO_SYNC") or "true").lower() not in ("0", "false", "no")
# The loop wakes on this cadence; whether an account is actually due is then
# decided per account from its own 同步频率 setting. A single global interval
# would mean one user's choice of 30 minutes silently applied to everyone.
TICK_SECONDS = 300
# A claim older than this is treated as abandoned — the worker holding it died
# mid-run, and without expiry the job would never run again.
CLAIM_TIMEOUT_SECONDS = 1800
@@ -99,14 +105,45 @@ def release(name=JOB_NAME, ran=True):
execute("UPDATE job_locks SET claimed_at = NULL WHERE name = ?", [name])
def sync_all_accounts(days=None):
"""Sync every account that has a stored token. Returns a per-user result."""
def due_at(user_id):
"""When this account may next be synced automatically, or None if never.
None means auto-sync is switched off for them; a time in the past means
they are due now.
"""
prefs = settings_svc.get_raw(user_id)
if not prefs.get("auto_sync"):
return None
minutes = prefs.get("auto_sync_minutes") or (INTERVAL_SECONDS // 60)
last = _parse((garmin_svc.get_sync_status(user_id) or {}).get("lastSyncTime"))
if not last:
return _now() - datetime.timedelta(seconds=1)
return last + datetime.timedelta(minutes=minutes)
def sync_all_accounts(days=None, respect_schedule=False):
"""Sync every account that has a stored token. Returns a per-user result.
`respect_schedule` is what the background loop passes: it skips accounts
that have auto-sync off or that were synced recently enough. A direct call
(a manual "sync everything") leaves it False and syncs unconditionally.
"""
days = days or SYNC_DAYS
rows = query_all("SELECT user_id FROM garmin_tokens")
results = []
for row in rows:
uid = row["user_id"]
try:
if respect_schedule:
due = due_at(uid)
if due is None:
results.append({"user": uid, "status": "skipped",
"reason": "auto-sync off"})
continue
if due > _now():
results.append({"user": uid, "status": "skipped",
"reason": "not due"})
continue
out = garmin_svc.sync_data(uid, {}, days=days)
results.append({"user": uid, "status": out.get("status"),
"records": out.get("recordsSynced")})
@@ -118,16 +155,16 @@ def sync_all_accounts(days=None):
def _loop():
while True:
try:
if claim():
if claim(interval=TICK_SECONDS):
try:
sync_all_accounts()
sync_all_accounts(respect_schedule=True)
finally:
release()
except Exception as e: # noqa: BLE001 - the loop must outlive any single failure
print(f"[scheduler] tick failed: {e}")
# Checked more often than the interval so a worker that starts late
# still picks the job up promptly rather than waiting a full hour.
time.sleep(min(300, INTERVAL_SECONDS))
time.sleep(TICK_SECONDS)
def start():
@@ -144,14 +181,27 @@ def start():
print(f"[scheduler] auto-sync every {INTERVAL_SECONDS}s, {SYNC_DAYS} day(s) back")
def status():
def status(user_id=None):
"""Scheduler state, and — when a user is named — their own next due time."""
row = query_one("SELECT * FROM job_locks WHERE name = ?", [JOB_NAME])
last = _parse(row.get("last_run_at")) if row else None
return {
out = {
"enabled": ENABLED,
"intervalSeconds": INTERVAL_SECONDS,
"tickSeconds": TICK_SECONDS,
"days": SYNC_DAYS,
"lastRunAt": _iso(last) if last else None,
"nextRunAt": _iso(last + datetime.timedelta(seconds=INTERVAL_SECONDS)) if last else None,
"nextRunAt": _iso(last + datetime.timedelta(seconds=TICK_SECONDS)) if last else None,
"running": bool(row and row.get("claimed_at")),
}
if user_id:
prefs = settings_svc.get_raw(user_id)
due = due_at(user_id)
out["account"] = {
"autoSync": bool(prefs.get("auto_sync")),
"intervalMinutes": prefs.get("auto_sync_minutes"),
"dueAt": _iso(due) if due else None,
}
return out

213
backend/services/settings.py Normal file
View File

@@ -0,0 +1,213 @@
"""
Per-user profile and preferences.
Two unrelated things share one table because they share a lifetime and an
edit screen: body measurements (height, weight, birth date, sex) that the
rating bands and the fitness-age estimate read, and preferences (units, how
often to sync, how far back to pull).
Every value is optional. An account with an empty profile still works — it
just falls back to general-population bands instead of personalised ones.
"""
import datetime
from db import execute, query_one
from config import DB_TYPE
# Only these are accepted from a request body; anything else is ignored rather
# than rejected, so an older client posting an unknown field still saves.
FIELDS = (
"height_cm", "weight_kg", "birth_date", "sex", "units",
"auto_sync", "auto_sync_minutes", "history_days",
)
DEFAULTS = {
"height_cm": None,
"weight_kg": None,
"birth_date": None,
"sex": None,
"units": "metric",
"auto_sync": 1,
"auto_sync_minutes": 60,
# How far back a full sync reaches. 0 means "everything Garmin has".
"history_days": 365,
}
SEXES = ("male", "female", "other")
UNITS = ("metric", "imperial")
# Offered in the UI as a picker; anything else is snapped to the nearest.
INTERVALS = (30, 60, 180, 360, 720, 1440)
HISTORY = (7, 30, 90, 180, 365, 730, 0)
CAMEL = {
"height_cm": "heightCm",
"weight_kg": "weightKg",
"birth_date": "birthDate",
"sex": "sex",
"units": "units",
"auto_sync": "autoSync",
"auto_sync_minutes": "autoSyncMinutes",
"history_days": "historyDays",
}
class InvalidSetting(ValueError):
"""A value the UI should not have sent; the message is user-facing."""
def _number(value, low, high, label):
if value is None or value == "":
return None
try:
n = float(value)
except (TypeError, ValueError):
raise InvalidSetting(f"{label}必须是数字")
if not low <= n <= high:
raise InvalidSetting(f"{label}应在 {low:g}~{high:g} 之间")
return n
def _date(value, label):
if not value:
return None
text = str(value)[:10]
try:
d = datetime.date.fromisoformat(text)
except ValueError:
raise InvalidSetting(f"{label}格式应为 YYYY-MM-DD")
today = datetime.date.today()
if not 0 < (today - d).days < 365 * 120:
raise InvalidSetting(f"{label}看起来不对")
return text
def _choice(value, allowed, label, default=None):
if value is None or value == "":
return default
if value not in allowed:
raise InvalidSetting(f"{label}只能是 {'/'.join(map(str, allowed))}")
return value
def _snap(value, allowed, default):
"""Nearest allowed number rather than an error.
The interval and history controls are pickers, so an off-list value means
a stale client, not a user mistake — snapping keeps them working.
"""
if value is None or value == "":
return default
try:
n = int(value)
except (TypeError, ValueError):
return default
if n in allowed:
return n
if n <= 0:
return 0 if 0 in allowed else default
return min((a for a in allowed if a > 0), key=lambda a: abs(a - n))
def clean(patch):
"""Validate an incoming patch into storable columns."""
out = {}
if "heightCm" in patch or "height_cm" in patch:
out["height_cm"] = _number(
patch.get("heightCm", patch.get("height_cm")), 80, 250, "身高"
)
if "weightKg" in patch or "weight_kg" in patch:
out["weight_kg"] = _number(
patch.get("weightKg", patch.get("weight_kg")), 25, 300, "体重"
)
if "birthDate" in patch or "birth_date" in patch:
out["birth_date"] = _date(
patch.get("birthDate", patch.get("birth_date")), "出生日期"
)
if "sex" in patch:
out["sex"] = _choice(patch.get("sex"), SEXES, "性别")
if "units" in patch:
out["units"] = _choice(patch.get("units"), UNITS, "单位", "metric")
if "autoSync" in patch or "auto_sync" in patch:
out["auto_sync"] = 1 if patch.get("autoSync", patch.get("auto_sync")) else 0
if "autoSyncMinutes" in patch or "auto_sync_minutes" in patch:
out["auto_sync_minutes"] = _snap(
patch.get("autoSyncMinutes", patch.get("auto_sync_minutes")),
INTERVALS, DEFAULTS["auto_sync_minutes"],
)
if "historyDays" in patch or "history_days" in patch:
out["history_days"] = _snap(
patch.get("historyDays", patch.get("history_days")),
HISTORY, DEFAULTS["history_days"],
)
return out
def get_raw(user_id):
"""Settings as column names, with defaults filled in for missing rows."""
row = query_one("SELECT * FROM user_settings WHERE user_id = ?", [user_id])
merged = dict(DEFAULTS)
if row:
for key in FIELDS:
value = row.get(key)
if value is not None:
merged[key] = value
# DATE comes back as a date object from MariaDB and a string from SQLite.
if merged["birth_date"] is not None:
merged["birth_date"] = str(merged["birth_date"])[:10]
return merged
def get_settings(user_id):
"""Settings in the camelCase the UI consumes, plus derived age."""
raw = get_raw(user_id)
out = {CAMEL[k]: raw[k] for k in FIELDS}
out["autoSync"] = bool(raw["auto_sync"])
out["age"] = age_from(raw["birth_date"])
out["bmi"] = bmi_from(raw["height_cm"], raw["weight_kg"])
return out
def save_settings(user_id, patch):
changes = clean(patch)
if not changes:
return get_settings(user_id)
current = get_raw(user_id)
current.update(changes)
current["updated_at"] = datetime.datetime.utcnow().isoformat(timespec="seconds")
cols = ["user_id"] + list(FIELDS) + ["updated_at"]
values = [user_id] + [current[k] for k in FIELDS] + [current["updated_at"]]
placeholders = ", ".join(["?"] * len(cols))
updatable = [c for c in cols if c != "user_id"]
if DB_TYPE == "mariadb":
updates = ", ".join(f"{c}=VALUES({c})" for c in updatable)
sql = (f"INSERT INTO user_settings ({', '.join(cols)}) "
f"VALUES ({placeholders}) ON DUPLICATE KEY UPDATE {updates}")
else:
updates = ", ".join(f"{c}=excluded.{c}" for c in updatable)
sql = (f"INSERT INTO user_settings ({', '.join(cols)}) "
f"VALUES ({placeholders}) ON CONFLICT(user_id) DO UPDATE SET {updates}")
execute(sql, values)
return get_settings(user_id)
def age_from(birth_date):
if not birth_date:
return None
try:
born = datetime.date.fromisoformat(str(birth_date)[:10])
except ValueError:
return None
today = datetime.date.today()
# Subtract a year when this year's birthday has not happened yet.
return today.year - born.year - (
(today.month, today.day) < (born.month, born.day)
)
def bmi_from(height_cm, weight_kg):
if not height_cm or not weight_kg:
return None
metres = float(height_cm) / 100
return round(float(weight_kg) / (metres * metres), 1)