from __future__ import annotations import os, json, time from typing import List, Dict, Any from flask import current_app, g from tinydb import TinyDB, Query from tinydb.storages import Storage FILE_NAME = "orders.json" PRODUCTION = 'dev' class PrettyJSONStorage(Storage): """Custom TinyDB storage that pretty-prints JSON.""" def __init__(self, path): self.path = path def read(self): try: with open(self.path, "r", encoding="utf-8") as f: return json.load(f) except FileNotFoundError: return None def write(self, data): with open(self.path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) def get_db() -> TinyDB: """ Return a persistent TinyDB instance per Flask app context. Stored in `g` to avoid reopening the file multiple times. """ if "orders_db" not in g: db_path = os.path.join(current_app.root_path, "data", FILE_NAME) if PRODUCTION == "dev": g.orders_db = TinyDB(db_path, storage=PrettyJSONStorage ) else: g.orders_db = TinyDB(db_path) return g.orders_db def get_orders_table(): """Return the orders table.""" return get_db().table("orders") def load_orders() -> List[Dict[str, Any]]: """Return all orders as a list of dicts.""" table = get_orders_table() return table.all() def get_order_by_id(order_id: str) -> Dict[str, Any] | None: """Return a single order by its ID, or None if not found.""" table = get_orders_table() Order = Query() result = table.get(Order.id == order_id) return result def load_archived() -> List[Dict[str, Any]]: """Return all archived orders, sorted by archived/done/created date descending.""" table = get_orders_table() Order = Query() archived = table.search(Order.status == "archived") def sort_key(o: Dict[str, Any]) -> str: return o.get("archived_at") or o.get("done_at") or o.get("created_at") or "" archived.sort(key=sort_key, reverse=True) return archived def load_pending() -> List[Dict[str, Any]]: """Return all orders with status 'open', sorted by created_at descending.""" table = get_orders_table() Order = Query() pending = table.search(Order.status == "open") pending.sort(key=lambda o: o.get("created_at", ""), reverse=True) return pending def load_done() -> List[Dict[str, Any]]: """Return all orders with status 'done', sorted by done_at descending.""" table = get_orders_table() Order = Query() done = table.search(Order.status == "done") done.sort(key=lambda o: o.get("done_at") or "", reverse=True) return done def add_order(order: Dict[str, Any]) -> None: table = get_orders_table() table.insert(order) def mark_done(order_id: str) -> bool: table = get_orders_table() Order = Query() now = time.strftime("%Y-%m-%dT%H:%M:%S%z", time.gmtime()) updated = table.update({"status": "done", "done_at": now}, Order.id == order_id) return bool(updated) def mark_archived(order_id: str) -> bool: """Mark an order as archived and set archived_at timestamp.""" now = time.strftime("%Y-%m-%dT%H:%M:%S%z", time.gmtime()) table = get_orders_table() Order = Query() updated = table.update({"status": "archived", "archived_at": now}, Order.id == order_id) return bool(updated) def update_order(order_id: str, patch: Dict[str, Any]) -> bool: table = get_orders_table() Order = Query() updated = table.update(patch, Order.id == order_id) return bool(updated) def cache_costs(order_id: str, costs: Dict[str, Any]) -> bool: now = time.strftime("%Y-%m-%dT%H:%M:%S%z", time.gmtime()) return update_order(order_id, {"cookbook": costs, "last_updated": now}) def cache_materials(order_id: str, mats: Dict[str, Any]) -> bool: now = time.strftime("%Y-%m-%dT%H:%M:%S%z", time.gmtime()) return update_order(order_id, {"materials": mats, "last_updated": now})