complete overhaul

2026-10-16 02:58:58 -07:00
1 parent 0bc92cdb8b
commit ee6c7719cf
@@ -0,0 +1,131 @@
+from flask import Flask, request, jsonify
+from database import *
+
+app = Flask(__name__)
+
+# Initialise database on startup
+initialize_db()
+
+# ----------------------------- Users -----------------------------
+
+@app.route("/users", methods=["GET", "POST"])
+def users():
+    if request.method == "GET":
+        # Optional query parameters
+        name = request.args.get("name")
+        id = request.args.get("id", type=int)
+        username = request.args.get("username")
+        fields = request.args.get("fields")
+        fields = fields.split(",") if fields else None
+        users = get_user(name=name, id=id, username=username, fields=fields)
+        return jsonify(users)
+
+    if request.method == "POST":
+        data = request.get_json()
+        if not data:
+            return jsonify({"error": "Missing JSON body"}), 400
+        name = data.get("name")
+        age = data.get("age")
+        photo = data.get("photo")
+        username = data.get("username")
+        if not all([name, age, username]):
+            return jsonify({"error": "name, age and username required"}), 400
+        result = add_user(name, age, photo, username)
+        return jsonify(result), 201
+
+@app.route("/users/<int:user_id>", methods=["PUT"])
+def update_user_route(user_id: int):
+    data = request.get_json()
+    if not data:
+        return jsonify({"error": "Missing JSON body"}), 400
+    name = data.get("name")
+    photo = data.get("photo")
+    settings = data.get("settings")
+    result = update_user(user_id, name=name, photo=photo, settings=settings)
+    return jsonify(result)
+
+# ----------------------------- Providers -----------------------------
+
+@app.route("/providers", methods=["GET", "POST"])
+def providers():
+    if request.method == "GET":
+        name = request.args.get("name")
+        id = request.args.get("id", type=int)
+        phone = request.args.get("phone")
+        fields = request.args.get("fields")
+        fields = fields.split(",") if fields else None
+        provs = get_provider(name=name, id=id, phone=phone, fields=fields)
+        return jsonify(provs)
+
+    if request.method == "POST":
+        data = request.get_json()
+        if not data:
+            return jsonify({"error": "Missing JSON body"}), 400
+        name = data.get("name")
+        age = data.get("age")
+        photo = data.get("photo")
+        phone = data.get("phone")
+        rates = data.get("rates", {})
+        private = data.get("private", False)
+        if not all([name, age, phone]):
+            return jsonify({"error": "name, age and phone required"}), 400
+        result = add_provider(name, age, photo, phone, rates, private)
+        return jsonify(result), 201
+
+@app.route("/providers/<int:prov_id>", methods=["PUT"])
+def update_provider_route(prov_id: int):
+    data = request.get_json()
+    if not data:
+        return jsonify({"error": "Missing JSON body"}), 400
+    name = data.get("name")
+    phone = data.get("phone")
+    rates = data.get("rates")
+    private = data.get("private")
+    photo = data.get("photo")
+    result = update_provider(prov_id, name=name, phone=phone, rates=rates, private=private, photo=photo)
+    return jsonify(result)
+
+# ----------------------------- Visits -----------------------------
+
+@app.route("/visits", methods=["GET", "POST"])
+def visits():
+    if request.method == "GET":
+        provider = request.args.get("provider", type=int)
+        user = request.args.get("user", type=int)
+        start = request.args.get("start")
+        end = request.args.get("end")
+        date_range = [start, end] if start and end else None
+        visits_list = get_visits(provider=provider, user=user, date_range=date_range)
+        return jsonify(visits_list)
+
+    if request.method == "POST":
+        data = request.get_json()
+        if not data:
+            return jsonify({"error": "Missing JSON body"}), 400
+        user_ids = data.get("user_ids", [])
+        provider_ids = data.get("provider_ids", [])
+        cost = data.get("cost")
+        location = data.get("location")
+        time = data.get("time")
+        if not (user_ids and provider_ids and time):
+            return jsonify({"error": "user_ids, provider_ids and time required"}), 400
+        result = add_visit(user_ids, provider_ids, cost, location, time)
+        return jsonify(result), 201
+
+@app.route("/visits/<int:visit_id>", methods=["PUT"])
+def update_visit_route(visit_id: int):
+    data = request.get_json()
+    if not data:
+        return jsonify({"error": "Missing JSON body"}), 400
+    user_ids = data.get("user_ids")
+    provider_ids = data.get("provider_ids")
+    cost = data.get("cost")
+    location = data.get("location")
+    time = data.get("time")
+    result = update_visit(visit_id, user_ids=user_ids, provider_ids=provider_ids,
+                          cost=cost, location=location, time=time)
+    return jsonify(result)
+
+# Run the app if this file is executed directly
+if __name__ == "__main__":
+    app.run(host="0.0.0.0", port=5000, debug=False)
@@ -0,0 +1,284 @@
+import sqlite3
+import json
+import os
+from typing import List, Dict, Any, Optional
+
+# Path to the SQLite database file
+DB_PATH = os.path.join(os.path.dirname(__file__), "fitness.db")
+
+# Ensure the database directory exists
+os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
+
+# ---------------------------------------------------------------------------
+# Helper functions
+# ---------------------------------------------------------------------------
+
+def _get_connection() -> sqlite3.Connection:
+    """Return a new SQLite connection with row factory set to sqlite3.Row."""
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    return conn
+
+# ---------------------------------------------------------------------------
+# Database initialization
+# ---------------------------------------------------------------------------
+
+def initialize_db() -> None:
+    """Create the database file and tables if they do not already exist.
+
+    This function is idempotent - calling it multiple times will not alter an
+    already-existing schema.
+    """
+    with _get_connection() as conn:
+        cursor = conn.cursor()
+        cursor.execute(
+            """
+            CREATE TABLE IF NOT EXISTS users (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                name TEXT NOT NULL,
+                username TEXT UNIQUE NOT NULL,
+                age INTEGER,
+                photo TEXT,
+                settings TEXT
+            );
+            """
+        )
+        cursor.execute(
+            """
+            CREATE TABLE IF NOT EXISTS providers (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                name TEXT NOT NULL,
+                phone TEXT,
+                age INTEGER,
+                photo TEXT,
+                rates TEXT,
+                private INTEGER NOT NULL DEFAULT 0
+            );
+            """
+        )
+        cursor.execute(
+            """
+            CREATE TABLE IF NOT EXISTS visits (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                user_ids TEXT NOT NULL,
+                provider_ids TEXT NOT NULL,
+                cost REAL,
+                location TEXT,
+                timestamp DATETIME NOT NULL
+            );
+            """
+        )
+        conn.commit()
+
+# ---------------------------------------------------------------------------
+# CRUD functions
+# ---------------------------------------------------------------------------
+
+def _select(table: str, where_clause: str = "", params: tuple = (), fields: List[str] = None) -> List[Dict[str, Any]]:
+    """Internal helper to execute SELECT queries.
+
+    Parameters
+    ----------
+    table: str
+        Name of the table to query.
+    where_clause: str
+        Optional WHERE clause (without the leading "WHERE").
+    params: tuple
+        Parameters for the query.
+    fields: List[str] | None
+        List of columns to retrieve. If None, all columns are returned.
+    """
+    cols = ", ".join(fields) if fields else "*"
+    sql = f"SELECT {cols} FROM {table}"
+    if where_clause:
+        sql += f" WHERE {where_clause}"
+    with _get_connection() as conn:
+        cursor = conn.execute(sql, params)
+        rows = [dict(row) for row in cursor.fetchall()]
+    return rows
+
+# Users ---------------------------------------------------------------------
+
+def get_user(name: Optional[str] = None, id: Optional[int] = None, username: Optional[str] = None, fields: List[str] = None) -> List[Dict[str, Any]]:
+    where_clauses = []
+    params = []
+    if name:
+        where_clauses.append("name = ?")
+        params.append(name)
+    if id is not None:
+        where_clauses.append("id = ?")
+        params.append(id)
+    if username:
+        where_clauses.append("username = ?")
+        params.append(username)
+    where = " AND ".join(where_clauses) if where_clauses else ""
+    return _select("users", where, tuple(params), fields)
+
+
+def add_user(name: str, age: int, photo: str, username: str) -> Dict[str, Any]:
+    """Insert a new user.
+
+    The ``settings`` column is initialised to an empty JSON object.
+    """
+    with _get_connection() as conn:
+        cursor = conn.execute(
+            "INSERT INTO users (name, username, age, photo, settings) VALUES (?, ?, ?, ?, ?)",
+            (name, username, age, photo, json.dumps({})),
+        )
+        conn.commit()
+        return {"status": "ok", "id": cursor.lastrowid}
+
+
+def update_user(id: int, name: Optional[str] = None, photo: Optional[str] = None, settings: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    updates = []
+    params = []
+    if name is not None:
+        updates.append("name = ?")
+        params.append(name)
+    if photo is not None:
+        updates.append("photo = ?")
+        params.append(photo)
+    if settings is not None:
+        updates.append("settings = ?")
+        params.append(json.dumps(settings))
+    if not updates:
+        return {"status": "no-op"}
+    params.append(id)
+    with _get_connection() as conn:
+        conn.execute(f"UPDATE users SET {', '.join(updates)} WHERE id = ?", tuple(params))
+        conn.commit()
+    return {"status": "ok", "id": id}
+
+# Providers -----------------------------------------------------------------
+
+def get_provider(name: Optional[str] = None, id: Optional[int] = None, phone: Optional[str] = None, fields: List[str] = None) -> List[Dict[str, Any]]:
+    where_clauses = []
+    params = []
+    if name:
+        where_clauses.append("name = ?")
+        params.append(name)
+    if id is not None:
+        where_clauses.append("id = ?")
+        params.append(id)
+    if phone:
+        where_clauses.append("phone = ?")
+        params.append(phone)
+    where = " AND ".join(where_clauses) if where_clauses else ""
+    return _select("providers", where, tuple(params), fields)
+
+
+def add_provider(name: str, age: int, photo: str, phone: str, rates: Dict[str, Any], private: bool) -> Dict[str, Any]:
+    with _get_connection() as conn:
+        cursor = conn.execute(
+            "INSERT INTO providers (name, phone, age, photo, rates, private) VALUES (?, ?, ?, ?, ?, ?)",
+            (name, phone, age, photo, json.dumps(rates), int(private)),
+        )
+        conn.commit()
+        return {"status": "ok", "id": cursor.lastrowid}
+
+
+def update_provider(id: int, name: Optional[str] = None, phone: Optional[str] = None, rates: Optional[Dict[str, Any]] = None, private: Optional[bool] = None, photo: Optional[str] = None) -> Dict[str, Any]:
+    updates = []
+    params = []
+    if name is not None:
+        updates.append("name = ?")
+        params.append(name)
+    if phone is not None:
+        updates.append("phone = ?")
+        params.append(phone)
+    if rates is not None:
+        updates.append("rates = ?")
+        params.append(json.dumps(rates))
+    if private is not None:
+        updates.append("private = ?")
+        params.append(int(private))
+    if photo is not None:
+        updates.append("photo = ?")
+        params.append(photo)
+    if not updates:
+        return {"status": "no-op"}
+    params.append(id)
+    with _get_connection() as conn:
+        conn.execute(f"UPDATE providers SET {', '.join(updates)} WHERE id = ?", tuple(params))
+        conn.commit()
+    return {"status": "ok", "id": id}
+
+# Visits ---------------------------------------------------------------------
+
+def _parse_user_provider_ids(ids_json: str) -> List[int]:
+    try:
+        return json.loads(ids_json)
+    except json.JSONDecodeError:
+        return []
+
+
+def get_visits(provider: Optional[int] = None, user: Optional[int] = None, date_range: Optional[List[str]] = None) -> List[Dict[str, Any]]:
+    """Return visits filtered by optional provider, user, and date range.
+
+    SQLite does not have a native ``IN`` operator that works directly on a JSON
+    array column.  We use ``EXISTS`` with ``json_each`` to test whether a value
+    appears in the stored JSON array.  This keeps the query simple and works
+    across SQLite versions.
+    """
+    where_clauses: List[str] = []
+    params: List[Any] = []
+    if provider is not None:
+        where_clauses.append("EXISTS (SELECT 1 FROM json_each(provider_ids) WHERE value = ?)")
+        params.append(provider)
+    if user is not None:
+        where_clauses.append("EXISTS (SELECT 1 FROM json_each(user_ids) WHERE value = ?)")
+        params.append(user)
+    if date_range:
+        where_clauses.append("timestamp BETWEEN ? AND ?")
+        params.extend(date_range)
+    where = " AND ".join(where_clauses) if where_clauses else ""
+    rows = _select("visits", where, tuple(params))
+    # Convert JSON id fields back to lists
+    for row in rows:
+        row["user_ids"] = _parse_user_provider_ids(row["user_ids"])
+        row["provider_ids"] = _parse_user_provider_ids(row["provider_ids"])
+    return rows
+
+
+def add_visit(user_ids: List[int], provider_ids: List[int], cost: float, location: str, time: str) -> Dict[str, Any]:
+    with _get_connection() as conn:
+        cursor = conn.execute(
+            "INSERT INTO visits (user_ids, provider_ids, cost, location, timestamp) VALUES (?, ?, ?, ?, ?)",
+            (json.dumps(user_ids), json.dumps(provider_ids), cost, location, time),
+        )
+        conn.commit()
+        return {"status": "ok", "id": cursor.lastrowid}
+
+
+def update_visit(id: int, user_ids: Optional[List[int]] = None, provider_ids: Optional[List[int]] = None, cost: Optional[float] = None, location: Optional[str] = None, time: Optional[str] = None) -> Dict[str, Any]:
+    updates = []
+    params = []
+    if user_ids is not None:
+        updates.append("user_ids = ?")
+        params.append(json.dumps(user_ids))
+    if provider_ids is not None:
+        updates.append("provider_ids = ?")
+        params.append(json.dumps(provider_ids))
+    if cost is not None:
+        updates.append("cost = ?")
+        params.append(cost)
+    if location is not None:
+        updates.append("location = ?")
+        params.append(location)
+    if time is not None:
+        updates.append("timestamp = ?")
+        params.append(time)
+    if not updates:
+        return {"status": "no-op"}
+    params.append(id)
+    with _get_connection() as conn:
+        conn.execute(f"UPDATE visits SET {', '.join(updates)} WHERE id = ?", tuple(params))
+        conn.commit()
+    return {"status": "ok", "id": id}
+
+# Example usage (to be removed or guarded by ``if __name__ == '__main__'``)
+if __name__ == "__main__":
+    initialize_db()
+    # Demo: add a user
+    res = add_user("Alice", 30, "alice.jpg", "alice123")
+    print("Added user:", res)