Spaces:
Sleeping
Sleeping
File size: 4,648 Bytes
d639fb1 75ef09c d639fb1 75ef09c 92dabc7 d639fb1 75ef09c d639fb1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | """
Async SQLite database layer using aiosqlite.
Tables: users, sessions, favorites, emergency_logs.
"""
import aiosqlite
import logging
from pathlib import Path
from typing import Optional, List, Dict, Any
log = logging.getLogger(__name__)
DB_PATH = Path(__file__).parent.parent.parent / "data" / "yatri.db"
class Database:
def __init__(self, db_path: str = str(DB_PATH)):
self.db_path = db_path
async def init_db(self):
"""Create tables if they don't exist."""
Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
async with aiosqlite.connect(self.db_path) as conn:
await conn.executescript("""
CREATE TABLE IF NOT EXISTS users (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
email TEXT UNIQUE NOT NULL,
phone TEXT,
password_hash TEXT NOT NULL,
preferred_language TEXT DEFAULT 'hi',
avatar_url TEXT,
created_at TEXT NOT NULL,
last_login TEXT
);
CREATE TABLE IF NOT EXISTS sessions (
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
query_text TEXT,
response_text TEXT,
language TEXT,
query_type TEXT CHECK(query_type IN ('voice', 'text', 'emergency')),
conversation_id TEXT,
created_at TEXT NOT NULL,
FOREIGN KEY (user_id) REFERENCES users(id)
);
CREATE TABLE IF NOT EXISTS favorites (
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
place_id TEXT NOT NULL,
created_at TEXT NOT NULL,
FOREIGN KEY (user_id) REFERENCES users(id)
);
CREATE TABLE IF NOT EXISTS emergency_logs (
id TEXT PRIMARY KEY,
user_id TEXT,
scenario TEXT,
language TEXT,
location_lat REAL,
location_lon REAL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS conversations (
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
title TEXT DEFAULT 'New Chat',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY (user_id) REFERENCES users(id)
);
CREATE TABLE IF NOT EXISTS feedback (
id TEXT PRIMARY KEY,
name TEXT,
email TEXT,
type TEXT,
message TEXT,
rating INTEGER DEFAULT 0,
created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id);
CREATE INDEX IF NOT EXISTS idx_sessions_conv ON sessions(conversation_id);
CREATE INDEX IF NOT EXISTS idx_conversations_user ON conversations(user_id);
CREATE INDEX IF NOT EXISTS idx_favorites_user ON favorites(user_id);
CREATE INDEX IF NOT EXISTS idx_emergency_user ON emergency_logs(user_id);
""")
await conn.commit()
log.info(f"Database initialized at {self.db_path}")
async def execute(self, sql: str, params: tuple = ()) -> None:
"""Run a write query (INSERT, UPDATE, DELETE)."""
async with aiosqlite.connect(self.db_path) as conn:
await conn.execute(sql, params)
await conn.commit()
async def fetch_one(self, sql: str, params: tuple = ()) -> Optional[Dict[str, Any]]:
"""Fetch a single row as a dict."""
async with aiosqlite.connect(self.db_path) as conn:
conn.row_factory = aiosqlite.Row
cursor = await conn.execute(sql, params)
row = await cursor.fetchone()
if row is None:
return None
return dict(row)
async def fetch_all(self, sql: str, params: tuple = ()) -> List[Dict[str, Any]]:
"""Fetch all rows as a list of dicts."""
async with aiosqlite.connect(self.db_path) as conn:
conn.row_factory = aiosqlite.Row
cursor = await conn.execute(sql, params)
rows = await cursor.fetchall()
return [dict(r) for r in rows]
# Module-level singleton
db = Database()
|