Spaces:
Sleeping
Sleeping
| import os | |
| import sqlite3 | |
| from typing import Optional, Dict | |
| from datetime import datetime, UTC | |
| def get_default_db_path() -> str: | |
| """Return default DB path or APP_DB_PATH env override.""" | |
| base = os.environ.get("APP_DB_PATH") | |
| if base: | |
| return base | |
| data_dir = os.path.join(os.getcwd(), "data") | |
| os.makedirs(data_dir, exist_ok=True) | |
| return os.path.join(data_dir, "state.db") | |
| def init_db(db_path: str) -> None: | |
| os.makedirs(os.path.dirname(db_path), exist_ok=True) | |
| with sqlite3.connect(db_path) as conn: | |
| conn.execute( | |
| """ | |
| CREATE TABLE IF NOT EXISTS tasks ( | |
| task TEXT PRIMARY KEY, | |
| repo_url TEXT NOT NULL, | |
| pages_url TEXT NOT NULL, | |
| commit_sha TEXT NOT NULL, | |
| updated_at TEXT NOT NULL | |
| ) | |
| """ | |
| ) | |
| conn.commit() | |
| def upsert_task(db_path: str, task: str, repo_url: str, pages_url: str, commit_sha: str) -> None: | |
| now = datetime.now(UTC) | |
| with sqlite3.connect(db_path) as conn: | |
| conn.execute( | |
| """ | |
| INSERT INTO tasks(task, repo_url, pages_url, commit_sha, updated_at) | |
| VALUES(?, ?, ?, ?, ?) | |
| ON CONFLICT(task) DO UPDATE SET | |
| repo_url=excluded.repo_url, | |
| pages_url=excluded.pages_url, | |
| commit_sha=excluded.commit_sha, | |
| updated_at=excluded.updated_at | |
| """, | |
| (task, repo_url, pages_url, commit_sha, now), | |
| ) | |
| conn.commit() | |
| def get_task(db_path: str, task: str) -> Optional[Dict[str, str]]: | |
| with sqlite3.connect(db_path) as conn: | |
| conn.row_factory = sqlite3.Row | |
| cur = conn.execute("SELECT * FROM tasks WHERE task = ?", (task,)) | |
| row = cur.fetchone() | |
| if not row: | |
| return None | |
| return { | |
| "repo_url": row["repo_url"], | |
| "pages_url": row["pages_url"], | |
| "commit_sha": row["commit_sha"], | |
| } | |