Spaces:
Sleeping
Sleeping
File size: 2,003 Bytes
e549219 | 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 | 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"],
}
|