Spaces:
Paused
Paused
File size: 6,426 Bytes
1ff060d | 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 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 | """
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())
|