131 lines
5.1 KiB
Python
131 lines
5.1 KiB
Python
"""Durable, account-scoped saved Search views."""
|
|
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
|
|
_REPOSITORY = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$")
|
|
_VIEW_ID = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
|
|
|
|
|
|
class SavedSearchConflict(ValueError):
|
|
"""Raised when a client attempts to replace a stale collection."""
|
|
|
|
def __init__(self, snapshot: dict):
|
|
super().__init__("saved searches changed on another device")
|
|
self.snapshot = snapshot
|
|
|
|
|
|
class SavedSearchStore:
|
|
def __init__(self, path: str | Path, *, limit: int = 20, 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 sqlite3.connect(self.path, timeout=self.timeout) as connection:
|
|
connection.execute("PRAGMA journal_mode=WAL")
|
|
connection.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS saved_searches (
|
|
login TEXT PRIMARY KEY,
|
|
revision INTEGER NOT NULL,
|
|
views TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
|
|
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 _snapshot(row) -> dict:
|
|
return {"revision": 0, "views": []} if row is None else {
|
|
"revision": int(row[0]), "views": json.loads(row[1])
|
|
}
|
|
|
|
def get(self, login: str) -> dict:
|
|
with self._connect() as connection:
|
|
row = connection.execute(
|
|
"SELECT revision, views FROM saved_searches WHERE login = ?",
|
|
(self._login(login),),
|
|
).fetchone()
|
|
return self._snapshot(row)
|
|
|
|
def _normalize(self, views: list[dict]) -> list[dict]:
|
|
if not isinstance(views, list):
|
|
raise ValueError("views must be a list")
|
|
if len(views) > self.limit:
|
|
raise ValueError(f"saved searches are limited to {self.limit}")
|
|
normalized = []
|
|
seen = set()
|
|
for raw in views:
|
|
if not isinstance(raw, dict):
|
|
raise ValueError("saved search must be an object")
|
|
view_id = raw.get("id")
|
|
if not isinstance(view_id, str) or not _VIEW_ID.fullmatch(view_id):
|
|
raise ValueError("id is invalid")
|
|
if view_id in seen:
|
|
raise ValueError("saved search ids must be unique")
|
|
name = raw.get("name")
|
|
if not isinstance(name, str) or not name.strip():
|
|
raise ValueError("name is required")
|
|
name = name.strip()
|
|
if len(name) > 60:
|
|
raise ValueError("name must be at most 60 characters")
|
|
query = raw.get("query")
|
|
if not isinstance(query, str) or not 2 <= len(query.strip()) <= 200:
|
|
raise ValueError("query must be between 2 and 200 characters")
|
|
kind = raw.get("kind", "all")
|
|
state = raw.get("state", "all")
|
|
if kind not in {"all", "issue", "pull"}:
|
|
raise ValueError("kind is invalid")
|
|
if state not in {"all", "open", "closed"}:
|
|
raise ValueError("state is invalid")
|
|
repository = raw.get("repository", "")
|
|
if not isinstance(repository, str) or (repository and not _REPOSITORY.fullmatch(repository)):
|
|
raise ValueError("repository is invalid")
|
|
seen.add(view_id)
|
|
normalized.append({
|
|
"id": view_id,
|
|
"name": name,
|
|
"query": query.strip(),
|
|
"kind": kind,
|
|
"state": state,
|
|
"repository": repository,
|
|
})
|
|
return normalized
|
|
|
|
def replace(self, login: str, expected_revision: int, views: list[dict]) -> dict:
|
|
login = self._login(login)
|
|
if not isinstance(expected_revision, int) or isinstance(expected_revision, bool) or expected_revision < 0:
|
|
raise ValueError("revision is invalid")
|
|
normalized = self._normalize(views)
|
|
serialized = json.dumps(normalized, separators=(",", ":"))
|
|
with self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
row = connection.execute(
|
|
"SELECT revision, views FROM saved_searches WHERE login = ?", (login,)
|
|
).fetchone()
|
|
current = self._snapshot(row)
|
|
if current["revision"] != expected_revision:
|
|
raise SavedSearchConflict(current)
|
|
revision = expected_revision + 1
|
|
connection.execute(
|
|
"INSERT INTO saved_searches(login, revision, views) VALUES (?, ?, ?) "
|
|
"ON CONFLICT(login) DO UPDATE SET revision=excluded.revision, views=excluded.views",
|
|
(login, revision, serialized),
|
|
)
|
|
return {"revision": revision, "views": normalized}
|