Initial pve cloud backup app
This commit is contained in:
@@ -0,0 +1,37 @@
|
||||
import sqlite3
|
||||
from contextlib import contextmanager
|
||||
from datetime import datetime, timezone
|
||||
from typing import Iterator
|
||||
|
||||
from .config import DB_PATH, ensure_dirs
|
||||
|
||||
|
||||
def utc_now() -> str:
|
||||
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
|
||||
|
||||
|
||||
def connect() -> sqlite3.Connection:
|
||||
ensure_dirs()
|
||||
conn = sqlite3.connect(DB_PATH, timeout=30)
|
||||
conn.row_factory = sqlite3.Row
|
||||
conn.execute("PRAGMA foreign_keys = ON")
|
||||
conn.execute("PRAGMA journal_mode = WAL")
|
||||
conn.execute("PRAGMA busy_timeout = 30000")
|
||||
return conn
|
||||
|
||||
|
||||
@contextmanager
|
||||
def db() -> Iterator[sqlite3.Connection]:
|
||||
conn = connect()
|
||||
try:
|
||||
yield conn
|
||||
conn.commit()
|
||||
except Exception:
|
||||
conn.rollback()
|
||||
raise
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def row_to_dict(row: sqlite3.Row | None) -> dict | None:
|
||||
return None if row is None else dict(row)
|
||||
Reference in New Issue
Block a user