340 lines
12 KiB
Python
340 lines
12 KiB
Python
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__), "db_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
|
|
);
|
|
"""
|
|
)
|
|
cursor.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS settings (
|
|
name TEXT PRIMARY KEY,
|
|
value TEXT
|
|
);
|
|
"""
|
|
)
|
|
|
|
# Initialize default settings
|
|
# Use JSON string for array values
|
|
cursor.execute(
|
|
"INSERT OR REPLACE INTO settings (name, value) VALUES (?, ?)",
|
|
("rate_options", "half_hour,hour"),
|
|
)
|
|
|
|
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}
|
|
|
|
# --- Settings helper functions ---
|
|
|
|
def add_setting(name: str, value: str) -> Dict[str, Any]:
|
|
"""Insert a new setting. If a setting with the same name exists, it will be replaced.
|
|
Returns a status dictionary.
|
|
"""
|
|
with _get_connection() as conn:
|
|
cursor = conn.execute(
|
|
"INSERT OR REPLACE INTO settings (name, value) VALUES (?, ?)",
|
|
(name, value),
|
|
)
|
|
conn.commit()
|
|
return {"status": "ok", "name": name}
|
|
|
|
|
|
def change_setting(name: str, value: str) -> Dict[str, Any]:
|
|
"""Update the value of an existing setting. If the setting does not exist, return not-found.
|
|
"""
|
|
with _get_connection() as conn:
|
|
cursor = conn.execute(
|
|
"UPDATE settings SET value = ? WHERE name = ?",
|
|
(value, name),
|
|
)
|
|
conn.commit()
|
|
if cursor.rowcount == 0:
|
|
return {"status": "not-found", "name": name}
|
|
return {"status": "ok", "name": name}
|
|
|
|
|
|
def view_settings(names: List[str] | None = None) -> List[Dict[str, Any]]:
|
|
"""Return a list of settings.
|
|
If names is provided, only those names are returned.
|
|
"""
|
|
if not names:
|
|
return _select("settings")
|
|
placeholders = ", ".join(["?" for _ in names])
|
|
rows = _select("settings", f"name IN ({placeholders})", tuple(names))
|
|
return rows
|
|
|
|
# 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)
|