204 lines
8.2 KiB
Python
204 lines
8.2 KiB
Python
"""Соединение с SQLite и схема хаба. Схема создаётся при первом подключении; версия — в setting.schema_version."""
|
||
import itertools
|
||
import json
|
||
import sqlite3
|
||
from contextlib import contextmanager
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
from . import backup, paths
|
||
|
||
SCHEMA_VERSION = 2
|
||
|
||
SCHEMA = """
|
||
CREATE TABLE IF NOT EXISTS setting (
|
||
key TEXT PRIMARY KEY,
|
||
value TEXT NOT NULL
|
||
);
|
||
CREATE TABLE IF NOT EXISTS issue (
|
||
key TEXT PRIMARY KEY,
|
||
id TEXT,
|
||
issue_type TEXT,
|
||
summary TEXT,
|
||
description TEXT,
|
||
status TEXT,
|
||
priority TEXT,
|
||
assignee TEXT,
|
||
reporter TEXT,
|
||
duedate TEXT,
|
||
planned_start TEXT,
|
||
planned_end TEXT,
|
||
epic_key TEXT,
|
||
labels TEXT NOT NULL DEFAULT '[]',
|
||
sprint TEXT,
|
||
created TEXT,
|
||
jira_updated TEXT,
|
||
fetched_at TEXT NOT NULL,
|
||
neighbor INTEGER NOT NULL DEFAULT 0
|
||
);
|
||
CREATE TABLE IF NOT EXISTS issue_link (
|
||
source_key TEXT NOT NULL,
|
||
target_key TEXT NOT NULL,
|
||
link_type TEXT NOT NULL,
|
||
PRIMARY KEY (source_key, target_key, link_type)
|
||
);
|
||
CREATE INDEX IF NOT EXISTS ix_issue_link_target ON issue_link(target_key);
|
||
CREATE TABLE IF NOT EXISTS issue_note (
|
||
key TEXT PRIMARY KEY,
|
||
fact_status TEXT,
|
||
work_kind TEXT,
|
||
done_text TEXT,
|
||
rest_text TEXT,
|
||
owner_note TEXT,
|
||
updated_at TEXT,
|
||
updated_by TEXT
|
||
);
|
||
CREATE TABLE IF NOT EXISTS batch (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
title TEXT NOT NULL,
|
||
author TEXT NOT NULL,
|
||
created_at TEXT NOT NULL,
|
||
cancelled_at TEXT,
|
||
estimate_seconds INTEGER NOT NULL DEFAULT 0
|
||
);
|
||
CREATE TABLE IF NOT EXISTS command (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
batch_id INTEGER NOT NULL REFERENCES batch(id),
|
||
seq INTEGER NOT NULL,
|
||
type TEXT NOT NULL,
|
||
params TEXT NOT NULL,
|
||
status TEXT NOT NULL DEFAULT 'queued',
|
||
claimed_at TEXT,
|
||
deliveries INTEGER NOT NULL DEFAULT 0,
|
||
finished_at TEXT,
|
||
result TEXT,
|
||
error TEXT,
|
||
jira_calls INTEGER NOT NULL DEFAULT 0,
|
||
rate_limited INTEGER NOT NULL DEFAULT 0,
|
||
pause_seconds INTEGER,
|
||
depends_on INTEGER REFERENCES command(id)
|
||
);
|
||
CREATE INDEX IF NOT EXISTS ix_command_status ON command(status, batch_id, seq);
|
||
CREATE TABLE IF NOT EXISTS sync_run (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
started_at TEXT NOT NULL,
|
||
finished_at TEXT,
|
||
kind TEXT NOT NULL,
|
||
query TEXT,
|
||
issues_count INTEGER NOT NULL DEFAULT 0,
|
||
links_count INTEGER NOT NULL DEFAULT 0,
|
||
jira_calls INTEGER NOT NULL DEFAULT 0,
|
||
rate_limited INTEGER NOT NULL DEFAULT 0,
|
||
command_id INTEGER
|
||
);
|
||
CREATE TABLE IF NOT EXISTS agent_status (
|
||
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||
last_seen_at TEXT,
|
||
agent_version TEXT,
|
||
settings_version INTEGER,
|
||
current_command_id INTEGER,
|
||
last_jira_call_at TEXT
|
||
);
|
||
"""
|
||
|
||
# Шаги миграции: версия → SQL, переводящий схему в версию + 1 (design.md bridge-reliability, решение 5)
|
||
MIGRATIONS = {
|
||
1: ["ALTER TABLE command ADD COLUMN deliveries INTEGER NOT NULL DEFAULT 0",
|
||
"UPDATE command SET deliveries = 1 WHERE status = 'claimed'"], # уже выданные — одна выдача
|
||
}
|
||
|
||
|
||
def now() -> str:
|
||
"""Текущее время UTC в ISO-8601 с секундами — единый формат дат в базе."""
|
||
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
|
||
|
||
|
||
def connect(path) -> sqlite3.Connection:
|
||
"""Открыть базу (создать при отсутствии), применить схему. Соединение в автокоммите; для атомарных
|
||
операций — transaction()."""
|
||
is_memory = str(path) == ":memory:"
|
||
if not is_memory and Path(path).parent != Path("."): # путь без каталога — рабочий каталог процесса не трогаем
|
||
paths.secure_dir(Path(path).parent)
|
||
conn = sqlite3.connect(str(path), timeout=10, isolation_level=None, check_same_thread=False)
|
||
conn.row_factory = sqlite3.Row
|
||
if not is_memory:
|
||
conn.execute("PRAGMA journal_mode = WAL")
|
||
conn.execute("PRAGMA foreign_keys = ON")
|
||
migrate(conn, None if is_memory else path)
|
||
return conn
|
||
|
||
|
||
def _schema_version(conn) -> int | None:
|
||
row = conn.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()
|
||
return None if row is None else int(json.loads(row["value"]))
|
||
|
||
|
||
def migrate(conn: sqlite3.Connection, path=None) -> None:
|
||
"""Создать недостающие таблицы и поднять схему существующей базы до SCHEMA_VERSION. Копия базы
|
||
(<каталог базы>/backups, без ротации) и миграция — под одной блокировкой записи (BEGIN IMMEDIATE): параллельный
|
||
connect дождётся её, перечитает версию и не станет ни копировать, ни мигрировать повторно (M-2 #12). Копию делает
|
||
отдельное соединение: VACUUM INTO внутри транзакции SQLite запрещает, а читатель при WAL видит базу до миграции.
|
||
Вызывается только вне транзакции."""
|
||
conn.executescript(SCHEMA)
|
||
version = _schema_version(conn)
|
||
if version is None:
|
||
conn.execute("INSERT OR IGNORE INTO setting(key, value) VALUES ('schema_version', ?)", (json.dumps(SCHEMA_VERSION),))
|
||
elif version < SCHEMA_VERSION:
|
||
with transaction(conn):
|
||
current = _schema_version(conn)
|
||
if current < SCHEMA_VERSION:
|
||
if path is not None:
|
||
reader = sqlite3.connect(str(path), timeout=10)
|
||
try:
|
||
backup.make_backup(reader, Path(path).parent / "backups", f"schema-v{current}", keep=0)
|
||
finally:
|
||
reader.close()
|
||
while current < SCHEMA_VERSION:
|
||
for sql in MIGRATIONS[current]:
|
||
conn.execute(sql)
|
||
current += 1
|
||
conn.execute("UPDATE setting SET value = ? WHERE key = 'schema_version'", (json.dumps(current),))
|
||
conn.execute("INSERT OR IGNORE INTO agent_status(id) VALUES (1)")
|
||
|
||
|
||
_savepoints = itertools.count(1)
|
||
|
||
|
||
@contextmanager
|
||
def transaction(conn: sqlite3.Connection):
|
||
"""BEGIN IMMEDIATE … COMMIT; при исключении — ROLLBACK. Нужна там, где читаем и пишем вместе (claim, отчёт).
|
||
Вложенный вызов (транзакция уже открыта) идёт через SAVEPOINT: при успехе RELEASE, при исключении
|
||
ROLLBACK TO + RELEASE и исключение пробрасывается — вызывающий может его перехватить, внешняя
|
||
транзакция при этом остаётся целой, а записи вложенного блока не сохраняются.
|
||
Соединение не должно делиться между потоками: каждый поток/запрос открывает своё."""
|
||
if conn.in_transaction:
|
||
name = f"sp_{next(_savepoints)}"
|
||
conn.execute(f"SAVEPOINT {name}")
|
||
try:
|
||
yield conn
|
||
except BaseException:
|
||
conn.execute(f"ROLLBACK TO {name}")
|
||
conn.execute(f"RELEASE {name}")
|
||
raise
|
||
else:
|
||
conn.execute(f"RELEASE {name}")
|
||
return
|
||
conn.execute("BEGIN IMMEDIATE")
|
||
try:
|
||
yield conn
|
||
except BaseException:
|
||
conn.execute("ROLLBACK")
|
||
raise
|
||
else:
|
||
conn.execute("COMMIT")
|
||
|
||
|
||
def utcnow() -> datetime:
|
||
return datetime.now(timezone.utc).replace(microsecond=0)
|
||
|
||
|
||
def parse(ts: str) -> datetime:
|
||
"""ISO-8601 из базы → aware datetime (naive считаем UTC)."""
|
||
d = datetime.fromisoformat(ts)
|
||
return d if d.tzinfo else d.replace(tzinfo=timezone.utc)
|