Spaces:
Runtime error
Runtime error
File size: 1,354 Bytes
9d9d2a1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 | import sqlite3
from typing import Optional, Dict, List
from src.utils.sqlite_retry import run_tx
from src.utils.time_utils import utc_now_iso
def is_closed(conn: sqlite3.Connection, business_date: str) -> bool:
row = conn.execute("SELECT 1 FROM day_closings WHERE business_date=?", (business_date,)).fetchone()
return bool(row)
def save_closing(conn: sqlite3.Connection, business_date: str, closed_by: str, note: str, pdf_path: str, pdf_bytes: bytes) -> None:
now = utc_now_iso()
def _tx(c):
c.execute(
"""
INSERT INTO day_closings(business_date, closed_at_utc, closed_by, note, pdf_path, pdf_bytes)
VALUES(?,?,?,?,?,?)
""",
(business_date, now, closed_by, note, pdf_path, pdf_bytes),
)
run_tx(conn, _tx)
def get_closing(conn: sqlite3.Connection, business_date: str) -> Optional[Dict]:
row = conn.execute(
"SELECT * FROM day_closings WHERE business_date=?",
(business_date,),
).fetchone()
return dict(row) if row else None
def recent_closings(conn: sqlite3.Connection, limit: int = 14) -> List[Dict]:
rows = conn.execute(
"SELECT business_date, closed_at_utc, closed_by, note, pdf_path FROM day_closings ORDER BY business_date DESC LIMIT ?",
(limit,),
).fetchall()
return [dict(r) for r in rows]
|