Spaces:
Paused
Paused
| """ | |
| DB — the only file that talks to Supabase | |
| ========================================= | |
| Everything the app needs from the database comes through the two | |
| functions at the bottom of this file. If we ever swap databases, this | |
| is the one file that changes. Same reason a shop puts all the dust | |
| collection in one trunk line instead of running hose everywhere. | |
| Secrets (URL + service_role key) come from the .env file, which git | |
| ignores. NEVER paste the key into code. The service_role key bypasses | |
| row-level security, so it belongs on the server only — never in a | |
| browser or a phone app. | |
| .env ──> load_dotenv() ──> os.environ ──> create_client() | |
| """ | |
| import os | |
| from dotenv import load_dotenv # reads .env into os.environ | |
| from supabase import create_client # the official Supabase client | |
| # Read .env once when this file is first imported. | |
| load_dotenv() | |
| # One shared connection, created the first time it's needed ("lazy"). | |
| # Creating it at import time would make the tests need a real key. | |
| _client = None | |
| def get_client(): | |
| """Return the shared Supabase client, creating it on first use.""" | |
| global _client | |
| if _client is None: | |
| url = os.environ.get("SUPABASE_URL", "") | |
| key = os.environ.get("SUPABASE_SERVICE_ROLE_KEY", "") | |
| if not url or not key or key.startswith("paste_"): | |
| raise RuntimeError( | |
| "Supabase is not configured. Copy .env.example to .env and " | |
| "fill in SUPABASE_URL and SUPABASE_SERVICE_ROLE_KEY." | |
| ) | |
| # Easy mistake: pasting the dashboard's web address. The API lives | |
| # at https://<project-id>.supabase.co; the dashboard page just | |
| # returns HTML, which the client can't parse into rows. | |
| if "supabase.com/dashboard" in url or not url.endswith(".supabase.co"): | |
| raise RuntimeError( | |
| f"SUPABASE_URL looks wrong: {url!r}. Use the project API URL " | |
| "(Settings -> API -> Project URL), e.g. " | |
| "https://your-project-id.supabase.co" | |
| ) | |
| # Another easy mistake: pasting the PUBLIC key. Row-level security | |
| # blocks it, so every query silently returns zero rows. The server | |
| # needs the secret key: "sb_secret_..." (new style) or the legacy | |
| # "service_role" JWT. | |
| if key.startswith("sb_publishable_") or key.startswith("sb_anon_"): | |
| raise RuntimeError( | |
| "SUPABASE_SERVICE_ROLE_KEY holds the PUBLIC (publishable) key. " | |
| "Use the secret key instead: Dashboard -> Project Settings -> " | |
| "API Keys -> Secret keys (starts with sb_secret_)." | |
| ) | |
| _client = create_client(url, key) | |
| return _client | |
| def fetch_price_items() -> list: | |
| """Return every row of price_items as a list of dicts.""" | |
| response = ( | |
| get_client() | |
| .table("price_items") | |
| .select("sku, category, series, description, config, width_in, " | |
| "drawers, special, price, glass_add, needs_verification") | |
| .order("category") | |
| .order("sku") | |
| .execute() | |
| ) | |
| return response.data | |
| def fetch_shop_settings() -> dict: | |
| """Return the shop's pricing defaults, in the engine's naming.""" | |
| response = ( | |
| get_client() | |
| .table("shops") | |
| .select("name, finder_fee_pct, buyout_markup_pct, default_tax_pct") | |
| .limit(1) | |
| .execute() | |
| ) | |
| if not response.data: | |
| raise RuntimeError("No row found in the shops table.") | |
| shop = response.data[0] | |
| # Rename default_tax_pct -> tax_pct so the engine has one vocabulary. | |
| return { | |
| "shop_name": shop["name"], | |
| "finder_fee_pct": float(shop["finder_fee_pct"]), | |
| "buyout_markup_pct": float(shop["buyout_markup_pct"]), | |
| "tax_pct": float(shop["default_tax_pct"]), | |
| } | |
| # --------------------------------------------------------------------- | |
| # JOBS — read a saved job and write its prices back | |
| # Used by POST /jobs/{id}/price, which n8n calls whenever a job changes. | |
| # --------------------------------------------------------------------- | |
| def fetch_job(job_id: str) -> dict | None: | |
| """Return one jobs row as a dict, or None if it doesn't exist.""" | |
| response = ( | |
| get_client() | |
| .table("jobs") | |
| .select("id, customer, project, gc_referral, gc_fee_pct, " | |
| "buyout_markup_pct, tax_pct, status") | |
| .eq("id", job_id) | |
| .limit(1) | |
| .execute() | |
| ) | |
| return response.data[0] if response.data else None | |
| def fetch_job_lines(job_id: str) -> list: | |
| """Return every job_lines row for a job, oldest first.""" | |
| response = ( | |
| get_client() | |
| .table("job_lines") | |
| .select("id, kind, sku, description, with_glass, qty, cost") | |
| .eq("job_id", job_id) | |
| .order("created_at") | |
| .execute() | |
| ) | |
| return response.data | |
| def save_job_pricing(job_id: str, line_ids: list, quote: dict) -> None: | |
| """Write a finished quote back: each line's price, then the job totals. | |
| `line_ids` is the list of job_lines ids in the SAME ORDER as | |
| quote["lines"], so we can zip them together. | |
| """ | |
| client = get_client() | |
| for line_id, priced in zip(line_ids, quote["lines"]): | |
| (client.table("job_lines") | |
| .update({"unit_price": priced["unit_price"], | |
| "line_total": priced["extended_price"]}) | |
| .eq("id", line_id) | |
| .execute()) | |
| (client.table("jobs") | |
| .update({"subtotal": quote["subtotal"], | |
| "finder_fee": quote["finder_fee"], | |
| "tax": quote["tax"], | |
| "total": quote["total"], | |
| "priced_at": "now()", | |
| "pricing_error": None, | |
| "needs_pricing": False}) # leave the queue | |
| .eq("id", job_id) | |
| .execute()) | |
| def save_job_pricing_error(job_id: str, message: str) -> None: | |
| """Record why a job could not be priced, so it shows in the app.""" | |
| (get_client().table("jobs") | |
| .update({"pricing_error": message, | |
| "needs_pricing": False}) # don't retry bad input forever | |
| .eq("id", job_id) | |
| .execute()) | |