122 lines
5.0 KiB
Python
122 lines
5.0 KiB
Python
"""Durable, account-scoped completed Filed review receipts."""
|
|
|
|
import re
|
|
import sqlite3
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
|
|
_REPOSITORY = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$")
|
|
|
|
|
|
class CompletedFiledReviewStore:
|
|
def __init__(self, path: str | Path, *, limit: int = 200, timeout: float = 1.0):
|
|
self.path = Path(path)
|
|
self.limit = limit
|
|
self.timeout = timeout
|
|
self._initialize()
|
|
|
|
def _initialize(self) -> None:
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
with self._connect() as connection:
|
|
connection.execute("PRAGMA journal_mode=WAL")
|
|
connection.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS completed_filed_reviews (
|
|
login TEXT NOT NULL,
|
|
repository TEXT NOT NULL,
|
|
issue_number INTEGER NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
touched_at INTEGER NOT NULL,
|
|
PRIMARY KEY (login, repository, issue_number)
|
|
)
|
|
"""
|
|
)
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
return sqlite3.connect(self.path, timeout=self.timeout)
|
|
|
|
@staticmethod
|
|
def _login(login: str) -> str:
|
|
normalized = login.strip().lower()
|
|
if not normalized:
|
|
raise ValueError("login is required")
|
|
return normalized
|
|
|
|
@staticmethod
|
|
def _receipt(raw: dict) -> tuple[str, int, str]:
|
|
if not isinstance(raw, dict):
|
|
raise ValueError("receipt must be an object")
|
|
repository = raw.get("repository")
|
|
number = raw.get("number")
|
|
updated_at = raw.get("updated_at")
|
|
if not isinstance(repository, str) or not _REPOSITORY.fullmatch(repository):
|
|
raise ValueError("repository is invalid")
|
|
if not isinstance(number, int) or isinstance(number, bool) or number < 1:
|
|
raise ValueError("number is invalid")
|
|
if not isinstance(updated_at, str):
|
|
raise ValueError("updated_at is invalid")
|
|
try:
|
|
parsed = datetime.fromisoformat(updated_at.replace("Z", "+00:00"))
|
|
except ValueError as error:
|
|
raise ValueError("updated_at is invalid") from error
|
|
if parsed.tzinfo is None:
|
|
raise ValueError("updated_at is invalid")
|
|
return repository, number, updated_at
|
|
|
|
@staticmethod
|
|
def _snapshot(rows) -> dict:
|
|
return {"receipts": [
|
|
{"repository": row[0], "number": int(row[1]), "updated_at": row[2]}
|
|
for row in rows
|
|
]}
|
|
|
|
def get(self, login: str) -> dict:
|
|
with self._connect() as connection:
|
|
rows = connection.execute(
|
|
"SELECT repository, issue_number, updated_at FROM completed_filed_reviews "
|
|
"WHERE login = ? ORDER BY touched_at",
|
|
(self._login(login),),
|
|
).fetchall()
|
|
return self._snapshot(rows)
|
|
|
|
def merge(self, login: str, receipts: list[dict]) -> dict:
|
|
login = self._login(login)
|
|
if not isinstance(receipts, list) or len(receipts) > self.limit:
|
|
raise ValueError(f"receipts are limited to {self.limit}")
|
|
normalized = [self._receipt(receipt) for receipt in receipts]
|
|
with self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
touched = int(connection.execute(
|
|
"SELECT COALESCE(MAX(touched_at), 0) FROM completed_filed_reviews WHERE login = ?",
|
|
(login,),
|
|
).fetchone()[0])
|
|
for repository, number, updated_at in normalized:
|
|
current = connection.execute(
|
|
"SELECT updated_at FROM completed_filed_reviews "
|
|
"WHERE login = ? AND repository = ? AND issue_number = ?",
|
|
(login, repository, number),
|
|
).fetchone()
|
|
if current is not None and current[0] >= updated_at:
|
|
continue
|
|
touched += 1
|
|
connection.execute(
|
|
"INSERT INTO completed_filed_reviews "
|
|
"(login, repository, issue_number, updated_at, touched_at) VALUES (?, ?, ?, ?, ?) "
|
|
"ON CONFLICT(login, repository, issue_number) DO UPDATE SET "
|
|
"updated_at=excluded.updated_at, touched_at=excluded.touched_at",
|
|
(login, repository, number, updated_at, touched),
|
|
)
|
|
connection.execute(
|
|
"DELETE FROM completed_filed_reviews WHERE login = ? AND rowid NOT IN ("
|
|
"SELECT rowid FROM completed_filed_reviews WHERE login = ? "
|
|
"ORDER BY touched_at DESC LIMIT ?)",
|
|
(login, login, self.limit),
|
|
)
|
|
rows = connection.execute(
|
|
"SELECT repository, issue_number, updated_at FROM completed_filed_reviews "
|
|
"WHERE login = ? ORDER BY touched_at",
|
|
(login,),
|
|
).fetchall()
|
|
return self._snapshot(rows)
|