bout right
This commit is contained in:
lda committed
1 parent
34bee8bf31
commit
2fc7d1cbfa
12 files changed
+1305
-335
No files matched your search
@@ -0,0 +1,18 @@
|
||||
"""Database abstraction layer.
|
||||
|
||||
Provides a Repository protocol and concrete implementations for SQLite and MySQL.
|
||||
The server doesn't care which one it talks to.
|
||||
"""
|
||||
|
||||
from db.repository import Repository
|
||||
from db.sqlite import SQLiteRepository
|
||||
|
||||
__all__ = ["Repository", "SQLiteRepository"]
|
||||
|
||||
# Optional MySQL import (requires mysql-connector-python)
|
||||
try:
|
||||
from db.mysql import MySQLRepository
|
||||
|
||||
__all__.append("MySQLRepository")
|
||||
except ImportError:
|
||||
pass
|
||||
+135
@@ -0,0 +1,135 @@
|
||||
"""MySQL implementation of the Repository."""
|
||||
|
||||
import os
|
||||
from typing import Optional
|
||||
|
||||
import mysql.connector # type: ignore[import-untyped]
|
||||
|
||||
from db.repository import DetectionRecord, User
|
||||
|
||||
|
||||
class MySQLRepository:
|
||||
"""MySQL-backed repository. Needs a running MySQL server + .env config."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
user: Optional[str] = None,
|
||||
password: Optional[str] = None,
|
||||
host: Optional[str] = None,
|
||||
database: Optional[str] = None,
|
||||
):
|
||||
self.config = {
|
||||
"user": user or os.getenv("DB_USER", "smoke_admin"),
|
||||
"password": password or os.getenv("DB_PASSWORD", "123456"),
|
||||
"host": host or os.getenv("DB_HOST", "localhost"),
|
||||
"database": database or os.getenv("DB_NAME", "smoke_detection"),
|
||||
}
|
||||
|
||||
def _connect(self):
|
||||
return mysql.connector.connect(**self.config)
|
||||
|
||||
def initialize(self) -> None:
|
||||
conn = self._connect()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
"""CREATE TABLE IF NOT EXISTS users (
|
||||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||||
username VARCHAR(50) UNIQUE NOT NULL,
|
||||
password VARCHAR(255) NOT NULL
|
||||
)"""
|
||||
)
|
||||
cursor.execute(
|
||||
"""CREATE TABLE IF NOT EXISTS detections (
|
||||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||||
user_id INT,
|
||||
image_path VARCHAR(255) NOT NULL,
|
||||
confidence FLOAT,
|
||||
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
camera_source VARCHAR(50) DEFAULT 'Laptop Webcam',
|
||||
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
|
||||
)"""
|
||||
)
|
||||
cursor.execute(
|
||||
"INSERT IGNORE INTO users (username, password) VALUES ('admin', '123456')"
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
# --- Users ---
|
||||
|
||||
def get_user_by_username(self, username: str) -> Optional[User]:
|
||||
conn = self._connect()
|
||||
cursor = conn.cursor(dictionary=True)
|
||||
cursor.execute("SELECT * FROM users WHERE username = %s", (username,))
|
||||
row = cursor.fetchone()
|
||||
conn.close()
|
||||
return User(**row) if row else None
|
||||
|
||||
def get_user_by_credentials(self, username: str, password: str) -> Optional[User]:
|
||||
conn = self._connect()
|
||||
cursor = conn.cursor(dictionary=True)
|
||||
cursor.execute(
|
||||
"SELECT * FROM users WHERE username = %s AND password = %s",
|
||||
(username, password),
|
||||
)
|
||||
row = cursor.fetchone()
|
||||
conn.close()
|
||||
return User(**row) if row else None
|
||||
|
||||
def create_user(self, username: str, password: str) -> int:
|
||||
conn = self._connect()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
"INSERT INTO users (username, password) VALUES (%s, %s)",
|
||||
(username, password),
|
||||
)
|
||||
conn.commit()
|
||||
new_id = cursor.lastrowid
|
||||
conn.close()
|
||||
return new_id # type: ignore[return-value]
|
||||
|
||||
# --- Detections ---
|
||||
|
||||
def save_detection(
|
||||
self,
|
||||
user_id: int,
|
||||
image_path: str,
|
||||
confidence: float,
|
||||
camera_source: str = "Laptop Webcam",
|
||||
) -> int:
|
||||
conn = self._connect()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
"INSERT INTO detections (user_id, image_path, confidence, camera_source) VALUES (%s, %s, %s, %s)",
|
||||
(user_id, image_path, confidence, camera_source),
|
||||
)
|
||||
conn.commit()
|
||||
new_id = cursor.lastrowid
|
||||
conn.close()
|
||||
return new_id # type: ignore[return-value]
|
||||
|
||||
def get_detections(self, user_id: Optional[int] = None) -> list[DetectionRecord]:
|
||||
conn = self._connect()
|
||||
cursor = conn.cursor(dictionary=True)
|
||||
if user_id is not None:
|
||||
cursor.execute(
|
||||
"SELECT * FROM detections WHERE user_id = %s ORDER BY timestamp DESC",
|
||||
(user_id,),
|
||||
)
|
||||
else:
|
||||
cursor.execute("SELECT * FROM detections ORDER BY timestamp DESC")
|
||||
rows = cursor.fetchall()
|
||||
conn.close()
|
||||
return [DetectionRecord(**r) for r in rows]
|
||||
|
||||
def clear_detections(self, user_id: int) -> int:
|
||||
conn = self._connect()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("DELETE FROM detections WHERE user_id = %s", (user_id,))
|
||||
conn.commit()
|
||||
count = cursor.rowcount
|
||||
conn.close()
|
||||
return count
|
||||
|
||||
def close(self) -> None:
|
||||
pass # MySQL uses per-call connections
|
||||
@@ -0,0 +1,68 @@
|
||||
"""Repository protocol - the interface all DB backends must implement."""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional, Protocol, runtime_checkable
|
||||
|
||||
|
||||
@dataclass
|
||||
class User:
|
||||
id: int
|
||||
username: str
|
||||
password: str # plaintext for now (legacy compat) - hash this later
|
||||
|
||||
|
||||
@dataclass
|
||||
class DetectionRecord:
|
||||
id: int
|
||||
user_id: int
|
||||
image_path: str
|
||||
confidence: float
|
||||
timestamp: str
|
||||
camera_source: str
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class Repository(Protocol):
|
||||
"""Database operations for the smoking detection server."""
|
||||
|
||||
def initialize(self) -> None:
|
||||
"""Create tables if they don't exist."""
|
||||
...
|
||||
|
||||
# --- Users ---
|
||||
|
||||
def get_user_by_username(self, username: str) -> Optional[User]:
|
||||
"""Find a user by username. Returns None if not found."""
|
||||
...
|
||||
|
||||
def get_user_by_credentials(self, username: str, password: str) -> Optional[User]:
|
||||
"""Find a user by username + password. Returns None if invalid."""
|
||||
...
|
||||
|
||||
def create_user(self, username: str, password: str) -> int:
|
||||
"""Create a new user. Returns the new user ID."""
|
||||
...
|
||||
|
||||
# --- Detections ---
|
||||
|
||||
def save_detection(
|
||||
self,
|
||||
user_id: int,
|
||||
image_path: str,
|
||||
confidence: float,
|
||||
camera_source: str = "Laptop Webcam",
|
||||
) -> int:
|
||||
"""Save a detection record. Returns the new record ID."""
|
||||
...
|
||||
|
||||
def get_detections(self, user_id: Optional[int] = None) -> list[DetectionRecord]:
|
||||
"""Get detection history, optionally filtered by user. Newest first."""
|
||||
...
|
||||
|
||||
def clear_detections(self, user_id: int) -> int:
|
||||
"""Delete all detections for a user. Returns number of rows deleted."""
|
||||
...
|
||||
|
||||
def close(self) -> None:
|
||||
"""Close the database connection."""
|
||||
...
|
||||
+121
@@ -0,0 +1,121 @@
|
||||
"""SQLite implementation of the Repository."""
|
||||
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from db.repository import DetectionRecord, User
|
||||
|
||||
# Schema that works with SQLite (no MySQL-specific syntax)
|
||||
SCHEMA = """\
|
||||
CREATE TABLE IF NOT EXISTS users (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
username TEXT UNIQUE NOT NULL,
|
||||
password TEXT NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS detections (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
user_id INTEGER,
|
||||
image_path TEXT NOT NULL,
|
||||
confidence REAL,
|
||||
timestamp TEXT DEFAULT (datetime('now', 'localtime')),
|
||||
camera_source TEXT DEFAULT 'Laptop Webcam',
|
||||
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
|
||||
);
|
||||
|
||||
INSERT OR IGNORE INTO users (username, password) VALUES ('admin', '123456');
|
||||
"""
|
||||
|
||||
|
||||
class SQLiteRepository:
|
||||
"""SQLite-backed repository. Zero setup, file-based."""
|
||||
|
||||
def __init__(self, db_path: str | Path = "data/smoke.db"):
|
||||
self.db_path = Path(db_path)
|
||||
self.db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
self._conn: Optional[sqlite3.Connection] = None
|
||||
|
||||
def _get_conn(self) -> sqlite3.Connection:
|
||||
if self._conn is None:
|
||||
self._conn = sqlite3.connect(str(self.db_path), check_same_thread=False)
|
||||
self._conn.row_factory = sqlite3.Row
|
||||
self._conn.execute("PRAGMA journal_mode=WAL")
|
||||
self._conn.execute("PRAGMA foreign_keys=ON")
|
||||
return self._conn
|
||||
|
||||
def initialize(self) -> None:
|
||||
conn = self._get_conn()
|
||||
conn.executescript(SCHEMA)
|
||||
conn.commit()
|
||||
|
||||
# --- Users ---
|
||||
|
||||
def get_user_by_username(self, username: str) -> Optional[User]:
|
||||
row = (
|
||||
self._get_conn()
|
||||
.execute("SELECT * FROM users WHERE username = ?", (username,))
|
||||
.fetchone()
|
||||
)
|
||||
return User(**dict(row)) if row else None
|
||||
|
||||
def get_user_by_credentials(self, username: str, password: str) -> Optional[User]:
|
||||
row = (
|
||||
self._get_conn()
|
||||
.execute(
|
||||
"SELECT * FROM users WHERE username = ? AND password = ?",
|
||||
(username, password),
|
||||
)
|
||||
.fetchone()
|
||||
)
|
||||
return User(**dict(row)) if row else None
|
||||
|
||||
def create_user(self, username: str, password: str) -> int:
|
||||
conn = self._get_conn()
|
||||
cursor = conn.execute(
|
||||
"INSERT INTO users (username, password) VALUES (?, ?)",
|
||||
(username, password),
|
||||
)
|
||||
conn.commit()
|
||||
return cursor.lastrowid # type: ignore[return-value]
|
||||
|
||||
# --- Detections ---
|
||||
|
||||
def save_detection(
|
||||
self,
|
||||
user_id: int,
|
||||
image_path: str,
|
||||
confidence: float,
|
||||
camera_source: str = "Laptop Webcam",
|
||||
) -> int:
|
||||
conn = self._get_conn()
|
||||
cursor = conn.execute(
|
||||
"INSERT INTO detections (user_id, image_path, confidence, camera_source) VALUES (?, ?, ?, ?)",
|
||||
(user_id, image_path, confidence, camera_source),
|
||||
)
|
||||
conn.commit()
|
||||
return cursor.lastrowid # type: ignore[return-value]
|
||||
|
||||
def get_detections(self, user_id: Optional[int] = None) -> list[DetectionRecord]:
|
||||
conn = self._get_conn()
|
||||
if user_id is not None:
|
||||
rows = conn.execute(
|
||||
"SELECT * FROM detections WHERE user_id = ? ORDER BY timestamp DESC",
|
||||
(user_id,),
|
||||
).fetchall()
|
||||
else:
|
||||
rows = conn.execute(
|
||||
"SELECT * FROM detections ORDER BY timestamp DESC"
|
||||
).fetchall()
|
||||
return [DetectionRecord(**dict(r)) for r in rows]
|
||||
|
||||
def clear_detections(self, user_id: int) -> int:
|
||||
conn = self._get_conn()
|
||||
cursor = conn.execute("DELETE FROM detections WHERE user_id = ?", (user_id,))
|
||||
conn.commit()
|
||||
return cursor.rowcount
|
||||
|
||||
def close(self) -> None:
|
||||
if self._conn:
|
||||
self._conn.close()
|
||||
self._conn = None
|
||||
Reference in new issue
Block a user