Add household support: multi-household isolation, each with own admin and users, 38 tests
This commit is contained in:
agentbox
@@ -4,6 +4,7 @@ import sqlite3
|
||||
import os
|
||||
from datetime import date, datetime
|
||||
|
||||
|
||||
def _db_path() -> str:
|
||||
return os.environ.get(
|
||||
"DB_PATH", os.path.join(os.path.dirname(__file__), "meals.db")
|
||||
@@ -19,14 +20,21 @@ def get_db() -> sqlite3.Connection:
|
||||
|
||||
|
||||
def init_db() -> None:
|
||||
"""Create tables if they don't exist."""
|
||||
"""Create tables if they don't exist, including migration for household support."""
|
||||
with get_db() as db:
|
||||
db.executescript("""
|
||||
CREATE TABLE IF NOT EXISTS households (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
name TEXT UNIQUE NOT NULL,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS users (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
username TEXT UNIQUE NOT NULL,
|
||||
password_hash TEXT NOT NULL,
|
||||
is_admin INTEGER NOT NULL DEFAULT 0,
|
||||
household_id INTEGER REFERENCES households(id),
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
|
||||
@@ -54,8 +62,100 @@ def init_db() -> None:
|
||||
ON responses(period_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_meal_periods_date
|
||||
ON meal_periods(date);
|
||||
CREATE INDEX IF NOT EXISTS idx_users_household
|
||||
ON users(household_id);
|
||||
""")
|
||||
|
||||
# Migration: if users table exists but lacks household_id column, add it.
|
||||
cols = [r["name"] for r in db.execute("PRAGMA table_info(users)").fetchall()]
|
||||
if "household_id" not in cols:
|
||||
db.execute("ALTER TABLE users ADD COLUMN household_id INTEGER REFERENCES households(id)")
|
||||
|
||||
|
||||
# ── Households ────────────────────────────────────────────────────────────────
|
||||
|
||||
def create_household(name: str) -> dict:
|
||||
"""Create a new household. Returns the row as a dict."""
|
||||
with get_db() as db:
|
||||
cur = db.execute("INSERT INTO households (name) VALUES (?)", (name,))
|
||||
db.commit()
|
||||
row = db.execute("SELECT * FROM households WHERE id = ?", (cur.lastrowid,)).fetchone()
|
||||
return dict(row)
|
||||
|
||||
|
||||
def get_household_by_id(household_id: int) -> dict | None:
|
||||
"""Fetch a household by ID."""
|
||||
with get_db() as db:
|
||||
row = db.execute("SELECT * FROM households WHERE id = ?", (household_id,)).fetchone()
|
||||
return dict(row) if row else None
|
||||
|
||||
|
||||
def get_household_by_name(name: str) -> dict | None:
|
||||
"""Fetch a household by name."""
|
||||
with get_db() as db:
|
||||
row = db.execute("SELECT * FROM households WHERE name = ?", (name,)).fetchone()
|
||||
return dict(row) if row else None
|
||||
|
||||
|
||||
def get_all_households() -> list[dict]:
|
||||
"""Return all households."""
|
||||
with get_db() as db:
|
||||
rows = db.execute("SELECT * FROM households ORDER BY name").fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
def count_household_users(household_id: int) -> int:
|
||||
"""Return number of users in a household."""
|
||||
with get_db() as db:
|
||||
return db.execute(
|
||||
"SELECT COUNT(*) FROM users WHERE household_id = ?", (household_id,)
|
||||
).fetchone()[0]
|
||||
|
||||
|
||||
# ── Users ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
def get_user_by_id(user_id: int) -> dict | None:
|
||||
"""Fetch a single user by ID."""
|
||||
with get_db() as db:
|
||||
row = db.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
|
||||
return dict(row) if row else None
|
||||
|
||||
|
||||
def get_user_by_username(username: str) -> dict | None:
|
||||
"""Fetch a single user by username."""
|
||||
with get_db() as db:
|
||||
row = db.execute(
|
||||
"SELECT * FROM users WHERE username = ?", (username,)
|
||||
).fetchone()
|
||||
return dict(row) if row else None
|
||||
|
||||
|
||||
def create_user(username: str, password_hash: str, household_id: int,
|
||||
is_admin: bool = False) -> dict:
|
||||
"""Insert a new user into a household. Returns the user row as a dict."""
|
||||
with get_db() as db:
|
||||
cur = db.execute(
|
||||
"INSERT INTO users (username, password_hash, is_admin, household_id) "
|
||||
"VALUES (?, ?, ?, ?)",
|
||||
(username, password_hash, int(is_admin), household_id),
|
||||
)
|
||||
user_id = cur.lastrowid
|
||||
db.commit()
|
||||
return get_user_by_id(user_id)
|
||||
|
||||
|
||||
def get_household_users(household_id: int) -> list[dict]:
|
||||
"""Return all users in a given household."""
|
||||
with get_db() as db:
|
||||
rows = db.execute(
|
||||
"SELECT id, username, is_admin, created_at "
|
||||
"FROM users WHERE household_id = ? ORDER BY username",
|
||||
(household_id,),
|
||||
).fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
# ── Meal periods ──────────────────────────────────────────────────────────────
|
||||
|
||||
def ensure_meal_periods(for_date: date) -> list[dict]:
|
||||
"""Ensure lunch and dinner periods exist for a given date.
|
||||
@@ -79,51 +179,10 @@ def ensure_meal_periods(for_date: date) -> list[dict]:
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
def get_user_by_id(user_id: int) -> dict | None:
|
||||
"""Fetch a single user by ID."""
|
||||
with get_db() as db:
|
||||
row = db.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
|
||||
return dict(row) if row else None
|
||||
# ── Dashboard ─────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def get_user_by_username(username: str) -> dict | None:
|
||||
"""Fetch a single user by username."""
|
||||
with get_db() as db:
|
||||
row = db.execute(
|
||||
"SELECT * FROM users WHERE username = ?", (username,)
|
||||
).fetchone()
|
||||
return dict(row) if row else None
|
||||
|
||||
|
||||
def create_user(username: str, password_hash: str, is_admin: bool = False) -> dict:
|
||||
"""Insert a new user. Returns the user row as a dict."""
|
||||
with get_db() as db:
|
||||
cur = db.execute(
|
||||
"INSERT INTO users (username, password_hash, is_admin) VALUES (?, ?, ?)",
|
||||
(username, password_hash, int(is_admin)),
|
||||
)
|
||||
user_id = cur.lastrowid
|
||||
db.commit() # Make visible to get_user_by_id (opens a new connection)
|
||||
return get_user_by_id(user_id)
|
||||
|
||||
|
||||
def get_all_users() -> list[dict]:
|
||||
"""Return all registered users."""
|
||||
with get_db() as db:
|
||||
rows = db.execute(
|
||||
"SELECT id, username, is_admin, created_at FROM users ORDER BY username"
|
||||
).fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
def count_users() -> int:
|
||||
"""Return total number of registered users."""
|
||||
with get_db() as db:
|
||||
return db.execute("SELECT COUNT(*) FROM users").fetchone()[0]
|
||||
|
||||
|
||||
def get_dashboard_data(for_date: date) -> dict:
|
||||
"""Return all users and their responses for a given date.
|
||||
def get_dashboard_data(for_date: date, household_id: int) -> dict:
|
||||
"""Return all users and their responses for a given date, scoped to a household.
|
||||
|
||||
Returns:
|
||||
{
|
||||
@@ -143,7 +202,7 @@ def get_dashboard_data(for_date: date) -> dict:
|
||||
}
|
||||
"""
|
||||
periods = ensure_meal_periods(for_date)
|
||||
users = get_all_users()
|
||||
users = get_household_users(household_id)
|
||||
|
||||
with get_db() as db:
|
||||
period_ids = [p["id"] for p in periods]
|
||||
@@ -158,7 +217,6 @@ def get_dashboard_data(for_date: date) -> dict:
|
||||
period_ids,
|
||||
).fetchall()
|
||||
|
||||
# Build a lookup: (user_id, period_id) → response
|
||||
resp_map = {}
|
||||
for r in resp_rows:
|
||||
resp_map[(r["user_id"], r["period_id"])] = {
|
||||
@@ -186,6 +244,8 @@ def get_dashboard_data(for_date: date) -> dict:
|
||||
}
|
||||
|
||||
|
||||
# ── Responses ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def upsert_response(user_id: int, period_id: int, new_status: str) -> dict | None:
|
||||
"""Insert or update a response. Handles changed_at logic.
|
||||
|
||||
@@ -199,16 +259,13 @@ def upsert_response(user_id: int, period_id: int, new_status: str) -> dict | Non
|
||||
).fetchone()
|
||||
|
||||
if existing is None:
|
||||
# First response ever
|
||||
db.execute(
|
||||
"INSERT INTO responses (user_id, period_id, status) VALUES (?, ?, ?)",
|
||||
(user_id, period_id, new_status),
|
||||
)
|
||||
elif existing["status"] == new_status:
|
||||
# No change — nothing to update
|
||||
pass
|
||||
elif existing["status"] in ("yes", "no") and new_status in ("yes", "no"):
|
||||
# Flipping between yes/no → mind changed
|
||||
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
db.execute(
|
||||
"UPDATE responses SET status = ?, changed_at = ? "
|
||||
@@ -216,7 +273,6 @@ def upsert_response(user_id: int, period_id: int, new_status: str) -> dict | Non
|
||||
(new_status, now, user_id, period_id),
|
||||
)
|
||||
else:
|
||||
# Transition from 'not_answered' → 'yes'/'no' (first answer)
|
||||
db.execute(
|
||||
"UPDATE responses SET status = ? WHERE user_id = ? AND period_id = ?",
|
||||
(new_status, user_id, period_id),
|
||||
|
||||
Reference in New Issue
Block a user