From a5d09be46e34de4bb5aa1b82c83e4e7b18ce0cb7 Mon Sep 17 00:00:00 2001 From: Navid Filsaraee Date: Mon, 15 Jun 2026 19:56:18 +0330 Subject: [PATCH] added db --- admin/models.py | 224 +++----------------------------- db/__init__.py | 4 + db/connection.py | 67 ++++++++++ db/repositories/__init__.py | 7 + db/repositories/flow_configs.py | 98 ++++++++++++++ db/repositories/runs.py | 82 ++++++++++++ main.py | 11 +- 7 files changed, 283 insertions(+), 210 deletions(-) create mode 100644 db/__init__.py create mode 100644 db/connection.py create mode 100644 db/repositories/__init__.py create mode 100644 db/repositories/flow_configs.py create mode 100644 db/repositories/runs.py diff --git a/admin/models.py b/admin/models.py index 522927d..66fbe10 100644 --- a/admin/models.py +++ b/admin/models.py @@ -1,241 +1,55 @@ -"""SQLite schema and DB helpers (aiosqlite).""" +"""Backwards-compatibility shim — delegates to db.repositories.""" from __future__ import annotations -import json -import os -import sqlite3 -from pathlib import Path from typing import Any -DB_PATH = os.getenv("DB_PATH", "data/tracker.db") +from db.connection import DB_PATH as DB_PATH, init_db as init_db +from db.repositories import flow_config_repo, runs_repo -CREATE_DDL = """ -CREATE TABLE IF NOT EXISTS runs ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), - scenario TEXT NOT NULL, - identifier TEXT NOT NULL, - success INTEGER NOT NULL, - failure_reason TEXT, - steps_json TEXT NOT NULL DEFAULT '[]', - total_ms INTEGER NOT NULL DEFAULT 0 -); - -CREATE INDEX IF NOT EXISTS idx_runs_scenario ON runs(scenario); -CREATE INDEX IF NOT EXISTS idx_runs_created_at ON runs(created_at); -CREATE INDEX IF NOT EXISTS idx_runs_identifier ON runs(identifier); - --- Dynamic flow configurations editable from the admin panel -CREATE TABLE IF NOT EXISTS flow_configs ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), - name TEXT NOT NULL, - is_active INTEGER NOT NULL DEFAULT 0, - - -- Search step - search_texts_json TEXT NOT NULL DEFAULT '[]', -- JSON string[] - search_box_selector TEXT NOT NULL DEFAULT '', - search_box_selector_type TEXT NOT NULL DEFAULT 'css', -- 'css' | 'xpath' - search_submit_selector TEXT NOT NULL DEFAULT '', -- optional — leave blank to use Enter key - search_submit_selector_type TEXT NOT NULL DEFAULT 'css', - search_results_selector TEXT NOT NULL DEFAULT '', -- container that appears after results load - - -- Scroll step - scroll_container_selector TEXT NOT NULL DEFAULT '', -- element to scroll; blank = window - max_scrolls INTEGER NOT NULL DEFAULT 30, - scroll_pause_ms INTEGER NOT NULL DEFAULT 1200, - no_new_content_timeout_ms INTEGER NOT NULL DEFAULT 3000, -- bail if DOM height unchanged after this - - -- Item selection - item_selector_template TEXT NOT NULL DEFAULT '', - -- Template may contain {item_id} placeholder, e.g. "[data-id='{item_id}']" - -- or a plain selector like ".result-card" (picks first match) - item_selector_type TEXT NOT NULL DEFAULT 'css', - target_item_ids_json TEXT NOT NULL DEFAULT '[]' -- JSON string[] — one click per id -); -""" +FlowConfigRow = dict[str, Any] -def init_db() -> None: - """Create the DB file and tables (sync, called at startup).""" - path = Path(DB_PATH) - path.parent.mkdir(parents=True, exist_ok=True) - conn = sqlite3.connect(str(path)) - conn.executescript(CREATE_DDL) - conn.commit() - conn.close() - - -# --------------------------------------------------------------------------- -# Async helpers (used by FastAPI routes via aiosqlite) -# --------------------------------------------------------------------------- +# ── runs ───────────────────────────────────────────────────────────────────── async def insert_run( scenario: str, identifier: str, success: bool, failure_reason: str | None, - steps: list[dict[str, object]], + steps: list[dict[str, Any]], total_ms: int, ) -> int: - import aiosqlite - async with aiosqlite.connect(DB_PATH) as db: - cur = await db.execute( - """INSERT INTO runs (scenario, identifier, success, failure_reason, steps_json, total_ms) - VALUES (?, ?, ?, ?, ?, ?)""", - (scenario, identifier, int(success), failure_reason, json.dumps(steps), total_ms), - ) - await db.commit() - return int(cur.lastrowid or 0) + return await runs_repo.insert(scenario, identifier, success, failure_reason, steps, total_ms) -# --------------------------------------------------------------------------- -# flow_configs CRUD -# --------------------------------------------------------------------------- +async def get_stats() -> dict[str, Any]: + return await runs_repo.stats() -FlowConfigRow = dict[str, Any] +# ── flow_configs ────────────────────────────────────────────────────────────── async def list_flow_configs() -> list[FlowConfigRow]: - import aiosqlite - async with aiosqlite.connect(DB_PATH) as db: - db.row_factory = aiosqlite.Row - rows = await (await db.execute( - "SELECT * FROM flow_configs ORDER BY id DESC" - )).fetchall() - return [_parse_flow_row(dict(r)) for r in rows] + return await flow_config_repo.list() async def get_flow_config(cfg_id: int) -> FlowConfigRow | None: - import aiosqlite - async with aiosqlite.connect(DB_PATH) as db: - db.row_factory = aiosqlite.Row - row = await (await db.execute( - "SELECT * FROM flow_configs WHERE id=?", (cfg_id,) - )).fetchone() - return _parse_flow_row(dict(row)) if row else None + return await flow_config_repo.get(cfg_id) async def get_active_flow_config() -> FlowConfigRow | None: - import aiosqlite - async with aiosqlite.connect(DB_PATH) as db: - db.row_factory = aiosqlite.Row - row = await (await db.execute( - "SELECT * FROM flow_configs WHERE is_active=1 ORDER BY id DESC LIMIT 1" - )).fetchone() - return _parse_flow_row(dict(row)) if row else None + return await flow_config_repo.get_active() async def upsert_flow_config(data: dict[str, Any], cfg_id: int | None = None) -> int: - import aiosqlite - - fields = [ - "name", "is_active", - "search_texts_json", "search_box_selector", "search_box_selector_type", - "search_submit_selector", "search_submit_selector_type", "search_results_selector", - "scroll_container_selector", "max_scrolls", "scroll_pause_ms", "no_new_content_timeout_ms", - "item_selector_template", "item_selector_type", "target_item_ids_json", - ] - - # Serialize list fields - for key in ("search_texts_json", "target_item_ids_json"): - if key in data and isinstance(data[key], list): - data[key] = json.dumps(data[key]) - - async with aiosqlite.connect(DB_PATH) as db: - if data.get("is_active"): - await db.execute("UPDATE flow_configs SET is_active=0") - - if cfg_id is None: - placeholders = ", ".join("?" for _ in fields) - cols = ", ".join(fields) - values = [data.get(f) for f in fields] - cur = await db.execute( - f"INSERT INTO flow_configs ({cols}) VALUES ({placeholders})", values - ) - row_id = int(cur.lastrowid or 0) - else: - set_clause = ", ".join(f"{f}=?" for f in fields) - values_u = [data.get(f) for f in fields] + [cfg_id] - await db.execute(f"UPDATE flow_configs SET {set_clause} WHERE id=?", values_u) - row_id = cfg_id - - await db.commit() - return row_id + if cfg_id is None: + return await flow_config_repo.create(data) + await flow_config_repo.update(cfg_id, data) + return cfg_id async def delete_flow_config(cfg_id: int) -> None: - import aiosqlite - async with aiosqlite.connect(DB_PATH) as db: - await db.execute("DELETE FROM flow_configs WHERE id=?", (cfg_id,)) - await db.commit() + await flow_config_repo.delete(cfg_id) async def activate_flow_config(cfg_id: int) -> None: - import aiosqlite - async with aiosqlite.connect(DB_PATH) as db: - await db.execute("UPDATE flow_configs SET is_active=0") - await db.execute("UPDATE flow_configs SET is_active=1 WHERE id=?", (cfg_id,)) - await db.commit() - - -def _parse_flow_row(row: dict[str, Any]) -> FlowConfigRow: - for key in ("search_texts_json", "target_item_ids_json"): - if key in row and isinstance(row[key], str): - try: - row[key] = json.loads(row[key]) - except (json.JSONDecodeError, TypeError): - row[key] = [] - return row - - -async def get_stats() -> dict[str, object]: - import aiosqlite - async with aiosqlite.connect(DB_PATH) as db: - db.row_factory = aiosqlite.Row - - total = (await (await db.execute("SELECT COUNT(*) FROM runs")).fetchone())[0] - unique_ids = (await (await db.execute("SELECT COUNT(DISTINCT identifier) FROM runs")).fetchone())[0] - successes = (await (await db.execute("SELECT COUNT(*) FROM runs WHERE success=1")).fetchone())[0] - failures = (await (await db.execute("SELECT COUNT(*) FROM runs WHERE success=0")).fetchone())[0] - - failure_reasons_raw = await ( - await db.execute( - """SELECT failure_reason, COUNT(*) as cnt - FROM runs WHERE success=0 AND failure_reason IS NOT NULL - GROUP BY failure_reason ORDER BY cnt DESC LIMIT 10""" - ) - ).fetchall() - failure_reasons = [{"reason": r["failure_reason"], "count": r["cnt"]} for r in failure_reasons_raw] - - scenario_breakdown_raw = await ( - await db.execute( - """SELECT scenario, COUNT(*) as total, - SUM(success) as ok, - COUNT(*) - SUM(success) as fail - FROM runs GROUP BY scenario""" - ) - ).fetchall() - scenario_breakdown = [ - {"scenario": r["scenario"], "total": r["total"], "success": r["ok"], "failure": r["fail"]} - for r in scenario_breakdown_raw - ] - - recent_raw = await ( - await db.execute( - """SELECT id, created_at, scenario, identifier, success, failure_reason, total_ms - FROM runs ORDER BY id DESC LIMIT 50""" - ) - ).fetchall() - recent = [dict(r) for r in recent_raw] - - return { - "total_runs": total, - "unique_identifiers": unique_ids, - "successes": successes, - "failures": failures, - "success_rate": round(successes / total * 100, 1) if total else 0.0, - "failure_reasons": failure_reasons, - "scenario_breakdown": scenario_breakdown, - "recent_runs": recent, - } + await flow_config_repo.activate(cfg_id) diff --git a/db/__init__.py b/db/__init__.py new file mode 100644 index 0000000..ed42255 --- /dev/null +++ b/db/__init__.py @@ -0,0 +1,4 @@ +from db.connection import DB_PATH, get_db, init_db +from db.repositories import flow_config_repo, runs_repo + +__all__ = ["DB_PATH", "get_db", "init_db", "runs_repo", "flow_config_repo"] diff --git a/db/connection.py b/db/connection.py new file mode 100644 index 0000000..f037049 --- /dev/null +++ b/db/connection.py @@ -0,0 +1,67 @@ +"""Database connection management and schema initialisation.""" +from __future__ import annotations + +import os +import sqlite3 +from contextlib import asynccontextmanager +from pathlib import Path +from typing import AsyncGenerator + +import aiosqlite + +DB_PATH: str = os.getenv("DB_PATH", "data/tracker.db") + +_DDL = """ +CREATE TABLE IF NOT EXISTS runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), + scenario TEXT NOT NULL, + identifier TEXT NOT NULL, + success INTEGER NOT NULL, + failure_reason TEXT, + steps_json TEXT NOT NULL DEFAULT '[]', + total_ms INTEGER NOT NULL DEFAULT 0 +); + +CREATE INDEX IF NOT EXISTS idx_runs_scenario ON runs(scenario); +CREATE INDEX IF NOT EXISTS idx_runs_created_at ON runs(created_at); +CREATE INDEX IF NOT EXISTS idx_runs_identifier ON runs(identifier); + +CREATE TABLE IF NOT EXISTS flow_configs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), + name TEXT NOT NULL, + is_active INTEGER NOT NULL DEFAULT 0, + search_texts_json TEXT NOT NULL DEFAULT '[]', + search_box_selector TEXT NOT NULL DEFAULT '', + search_box_selector_type TEXT NOT NULL DEFAULT 'css', + search_submit_selector TEXT NOT NULL DEFAULT '', + search_submit_selector_type TEXT NOT NULL DEFAULT 'css', + search_results_selector TEXT NOT NULL DEFAULT '', + scroll_container_selector TEXT NOT NULL DEFAULT '', + max_scrolls INTEGER NOT NULL DEFAULT 30, + scroll_pause_ms INTEGER NOT NULL DEFAULT 1200, + no_new_content_timeout_ms INTEGER NOT NULL DEFAULT 3000, + item_selector_template TEXT NOT NULL DEFAULT '', + item_selector_type TEXT NOT NULL DEFAULT 'css', + target_item_ids_json TEXT NOT NULL DEFAULT '[]' +); +""" + + +def init_db() -> None: + """Create the DB file and apply the schema (sync, called once at startup).""" + path = Path(DB_PATH) + path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(str(path)) + conn.executescript(_DDL) + conn.commit() + conn.close() + + +@asynccontextmanager +async def get_db() -> AsyncGenerator[aiosqlite.Connection, None]: + """Async context manager — yields an open, Row-factory-enabled connection.""" + async with aiosqlite.connect(DB_PATH) as db: + db.row_factory = aiosqlite.Row + yield db diff --git a/db/repositories/__init__.py b/db/repositories/__init__.py new file mode 100644 index 0000000..7e61633 --- /dev/null +++ b/db/repositories/__init__.py @@ -0,0 +1,7 @@ +from db.repositories.flow_configs import FlowConfigRepository, flow_config_repo +from db.repositories.runs import RunsRepository, runs_repo + +__all__ = [ + "RunsRepository", "runs_repo", + "FlowConfigRepository", "flow_config_repo", +] diff --git a/db/repositories/flow_configs.py b/db/repositories/flow_configs.py new file mode 100644 index 0000000..c8900a6 --- /dev/null +++ b/db/repositories/flow_configs.py @@ -0,0 +1,98 @@ +"""Repository for the `flow_configs` table.""" +from __future__ import annotations + +import json +from typing import Any + +from db.connection import get_db + +FlowConfigRow = dict[str, Any] + +_FIELDS = [ + "name", "is_active", + "search_texts_json", "search_box_selector", "search_box_selector_type", + "search_submit_selector", "search_submit_selector_type", "search_results_selector", + "scroll_container_selector", "max_scrolls", "scroll_pause_ms", "no_new_content_timeout_ms", + "item_selector_template", "item_selector_type", "target_item_ids_json", +] + + +def _deserialise(row: dict[str, Any]) -> FlowConfigRow: + for key in ("search_texts_json", "target_item_ids_json"): + if isinstance(row.get(key), str): + try: + row[key] = json.loads(row[key]) + except (json.JSONDecodeError, TypeError): + row[key] = [] + return row + + +def _serialise(data: dict[str, Any]) -> dict[str, Any]: + out = dict(data) + for key in ("search_texts_json", "target_item_ids_json"): + if isinstance(out.get(key), list): + out[key] = json.dumps(out[key]) + return out + + +class FlowConfigRepository: + async def list(self) -> list[FlowConfigRow]: + async with get_db() as db: + rows = await (await db.execute( + "SELECT * FROM flow_configs ORDER BY id DESC" + )).fetchall() + return [_deserialise(dict(r)) for r in rows] + + async def get(self, cfg_id: int) -> FlowConfigRow | None: + async with get_db() as db: + row = await (await db.execute( + "SELECT * FROM flow_configs WHERE id=?", (cfg_id,) + )).fetchone() + return _deserialise(dict(row)) if row else None + + async def get_active(self) -> FlowConfigRow | None: + async with get_db() as db: + row = await (await db.execute( + "SELECT * FROM flow_configs WHERE is_active=1 ORDER BY id DESC LIMIT 1" + )).fetchone() + return _deserialise(dict(row)) if row else None + + async def create(self, data: dict[str, Any]) -> int: + data = _serialise(data) + async with get_db() as db: + if data.get("is_active"): + await db.execute("UPDATE flow_configs SET is_active=0") + placeholders = ", ".join("?" for _ in _FIELDS) + cols = ", ".join(_FIELDS) + values = [data.get(f) for f in _FIELDS] + cur = await db.execute( + f"INSERT INTO flow_configs ({cols}) VALUES ({placeholders})", values + ) + await db.commit() + return int(cur.lastrowid or 0) + + async def update(self, cfg_id: int, data: dict[str, Any]) -> None: + data = _serialise(data) + async with get_db() as db: + if data.get("is_active"): + await db.execute("UPDATE flow_configs SET is_active=0") + set_clause = ", ".join(f"{f}=?" for f in _FIELDS) + values = [data.get(f) for f in _FIELDS] + [cfg_id] + await db.execute( + f"UPDATE flow_configs SET {set_clause} WHERE id=?", values + ) + await db.commit() + + async def activate(self, cfg_id: int) -> None: + async with get_db() as db: + await db.execute("UPDATE flow_configs SET is_active=0") + await db.execute("UPDATE flow_configs SET is_active=1 WHERE id=?", (cfg_id,)) + await db.commit() + + async def delete(self, cfg_id: int) -> None: + async with get_db() as db: + await db.execute("DELETE FROM flow_configs WHERE id=?", (cfg_id,)) + await db.commit() + + +flow_config_repo = FlowConfigRepository() diff --git a/db/repositories/runs.py b/db/repositories/runs.py new file mode 100644 index 0000000..bc6b48c --- /dev/null +++ b/db/repositories/runs.py @@ -0,0 +1,82 @@ +"""Repository for the `runs` table.""" +from __future__ import annotations + +import json +from typing import Any + +from db.connection import get_db + + +class RunsRepository: + async def insert( + self, + scenario: str, + identifier: str, + success: bool, + failure_reason: str | None, + steps: list[dict[str, Any]], + total_ms: int, + ) -> int: + async with get_db() as db: + cur = await db.execute( + """INSERT INTO runs + (scenario, identifier, success, failure_reason, steps_json, total_ms) + VALUES (?, ?, ?, ?, ?, ?)""", + (scenario, identifier, int(success), failure_reason, json.dumps(steps), total_ms), + ) + await db.commit() + return int(cur.lastrowid or 0) + + async def stats(self) -> dict[str, Any]: + async with get_db() as db: + total = (await (await db.execute("SELECT COUNT(*) FROM runs")).fetchone())[0] + unique_ids = (await (await db.execute("SELECT COUNT(DISTINCT identifier) FROM runs")).fetchone())[0] + successes = (await (await db.execute("SELECT COUNT(*) FROM runs WHERE success=1")).fetchone())[0] + failures = (await (await db.execute("SELECT COUNT(*) FROM runs WHERE success=0")).fetchone())[0] + + failure_reasons = [ + {"reason": r["failure_reason"], "count": r["cnt"]} + for r in await (await db.execute( + """SELECT failure_reason, COUNT(*) as cnt + FROM runs WHERE success=0 AND failure_reason IS NOT NULL + GROUP BY failure_reason ORDER BY cnt DESC LIMIT 10""" + )).fetchall() + ] + + scenario_breakdown = [ + { + "scenario": r["scenario"], + "total": r["total"], + "success": r["ok"], + "failure": r["fail"], + } + for r in await (await db.execute( + """SELECT scenario, + COUNT(*) as total, + SUM(success) as ok, + COUNT(*) - SUM(success) as fail + FROM runs GROUP BY scenario""" + )).fetchall() + ] + + recent = [ + dict(r) + for r in await (await db.execute( + """SELECT id, created_at, scenario, identifier, success, failure_reason, total_ms + FROM runs ORDER BY id DESC LIMIT 50""" + )).fetchall() + ] + + return { + "total_runs": total, + "unique_identifiers": unique_ids, + "successes": successes, + "failures": failures, + "success_rate": round(successes / total * 100, 1) if total else 0.0, + "failure_reasons": failure_reasons, + "scenario_breakdown": scenario_breakdown, + "recent_runs": recent, + } + + +runs_repo = RunsRepository() diff --git a/main.py b/main.py index 20d1a65..7a0a993 100644 --- a/main.py +++ b/main.py @@ -8,12 +8,13 @@ import time import uvicorn -from admin.models import get_active_flow_config, init_db, insert_run +from db import flow_config_repo, init_db +from db.repositories.runs import runs_repo from config import config def _load_active_flow() -> dict[str, object]: - cfg = asyncio.run(get_active_flow_config()) + cfg = asyncio.run(flow_config_repo.get_active()) if cfg is None: print("[WARN] No active flow config — running without dynamic flow steps.", file=sys.stderr) return cfg or {} @@ -30,7 +31,7 @@ def _record( total_ms = int((time.monotonic() - t0) * 1000) if exc is not None: - asyncio.run(insert_run(scenario, identifier, False, str(exc), [], total_ms)) + asyncio.run(runs_repo.insert(scenario, identifier, False, str(exc), [], total_ms)) print(f"[FAIL] {identifier}: {exc}", file=sys.stderr) return @@ -39,13 +40,13 @@ def _record( {"name": s.name, "success": s.success, "error": s.error, "duration_ms": s.duration_ms} for s in flow_result.steps ] - asyncio.run(insert_run( + asyncio.run(runs_repo.insert( scenario, identifier, flow_result.success, flow_result.failure_reason, steps_data, total_ms, )) status = "OK" if flow_result.success else f"FAIL ({flow_result.failure_reason})" else: - asyncio.run(insert_run(scenario, identifier, True, None, [], total_ms)) + asyncio.run(runs_repo.insert(scenario, identifier, True, None, [], total_ms)) status = "OK (no flow config)" print(f"[{status}] {identifier} — {total_ms} ms")