feat(hub): вложенные транзакции присоединяются к внешней, db.utcnow и db.parse

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014Typ5sXWNMFirr2va2uL8o
This commit is contained in:
Andrey Epifancev 2026-10-11 22:07:07 +04:00
parent 68018c3e5e
commit f353d90ec6
2 changed files with 47 additions and 1 deletions

View File

@ -128,7 +128,12 @@ def migrate(conn: sqlite3.Connection) -> None:
@contextmanager
def transaction(conn: sqlite3.Connection):
"""BEGIN IMMEDIATE … COMMIT; при исключении — ROLLBACK. Нужна там, где читаем и пишем вместе (claim, отчёт)."""
"""BEGIN IMMEDIATE … COMMIT; при исключении — ROLLBACK. Нужна там, где читаем и пишем вместе (claim, отчёт).
Вложенный вызов (транзакция уже открыта) входит во внешнюю: своих BEGIN/COMMIT/ROLLBACK не делает,
исключение уходит внешнему блоку, который откатит всё."""
if conn.in_transaction:
yield conn
return
conn.execute("BEGIN IMMEDIATE")
try:
yield conn
@ -137,3 +142,13 @@ def transaction(conn: sqlite3.Connection):
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)

View File

@ -25,3 +25,34 @@ def test_transaction_rolls_back_on_error(tmp_path):
except RuntimeError:
pass
assert conn.execute("SELECT COUNT(*) AS n FROM batch").fetchone()["n"] == 0
def test_nested_transaction_joins_outer(tmp_path):
conn = db.connect(tmp_path / "hub.sqlite")
try:
with db.transaction(conn):
with db.transaction(conn):
conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('x', 'y', 'z')")
assert conn.in_transaction # вложенный блок не закоммитил сам
assert conn.execute("SELECT COUNT(*) AS n FROM batch").fetchone()["n"] == 1
raise RuntimeError("boom")
except RuntimeError:
pass
assert conn.execute("SELECT COUNT(*) AS n FROM batch").fetchone()["n"] == 0 # откат внешней откатил и вложенную
def test_nested_transaction_error_propagates_to_outer(tmp_path):
conn = db.connect(tmp_path / "hub.sqlite")
try:
with db.transaction(conn):
conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('a', 'y', 'z')")
with db.transaction(conn):
raise RuntimeError("inner")
except RuntimeError:
pass
assert conn.execute("SELECT COUNT(*) AS n FROM batch").fetchone()["n"] == 0
def test_parse_and_utcnow():
assert db.parse("2020-01-01T00:00:00").tzinfo is not None
assert db.parse(db.now()) <= db.utcnow()