Parts inventory: catalog, search, and stock tracking
FastAPI + SQLite behind a single-page frontend, deployed as a container on mainserver behind Caddy. One flexible parts table plus key/value specs so components, filament, fasteners and tooling share a schema. Categories and locations are nestable trees whose filters and sidebar counts both roll up through descendants. Category spec templates pre-fill the properties worth recording for each kind of part, which is what makes manual entry tolerable. Every quantity change is logged. Search is FTS5 with prefix matching across names, MPNs, spec values, tags, category and location. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
+595
@@ -0,0 +1,595 @@
|
||||
"""Parts inventory API.
|
||||
|
||||
JSON API plus a single self-contained frontend. Everything a bench lookup needs
|
||||
lives behind /api; the browser app in static/ is the only consumer today, and
|
||||
the future "what could I build with this?" feature is meant to be another
|
||||
consumer of the same endpoints rather than a fork of them.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
from typing import Any, Literal
|
||||
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
from fastapi import Depends, FastAPI, HTTPException, Query, Request, Response
|
||||
from fastapi.responses import FileResponse, JSONResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from . import auth, db
|
||||
|
||||
STATIC_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "static")
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(_app: FastAPI):
|
||||
db.init()
|
||||
yield
|
||||
|
||||
|
||||
app = FastAPI(title="Parts Inventory", docs_url=None, redoc_url=None, lifespan=lifespan)
|
||||
|
||||
|
||||
# --- models -----------------------------------------------------------------
|
||||
|
||||
class SpecIn(BaseModel):
|
||||
key: str = Field(min_length=1, max_length=80)
|
||||
value: str = Field(default="", max_length=400)
|
||||
|
||||
|
||||
class PartIn(BaseModel):
|
||||
name: str = Field(min_length=1, max_length=200)
|
||||
description: str = Field(default="", max_length=2000)
|
||||
category_id: int | None = None
|
||||
location_id: int | None = None
|
||||
manufacturer: str = Field(default="", max_length=200)
|
||||
mpn: str = Field(default="", max_length=120)
|
||||
quantity: float = 0
|
||||
unit: str = Field(default="pcs", max_length=20)
|
||||
min_quantity: float | None = None
|
||||
cost_each: float | None = None
|
||||
datasheet_url: str = Field(default="", max_length=1000)
|
||||
product_url: str = Field(default="", max_length=1000)
|
||||
notes: str = Field(default="", max_length=4000)
|
||||
specs: list[SpecIn] = []
|
||||
tags: list[str] = []
|
||||
|
||||
|
||||
class PartPatch(PartIn):
|
||||
name: str | None = Field(default=None, min_length=1, max_length=200)
|
||||
description: str | None = None
|
||||
quantity: float | None = None
|
||||
unit: str | None = None
|
||||
specs: list[SpecIn] | None = None
|
||||
tags: list[str] | None = None
|
||||
|
||||
|
||||
class AdjustIn(BaseModel):
|
||||
delta: float
|
||||
reason: str = Field(default="", max_length=300)
|
||||
|
||||
|
||||
class NodeIn(BaseModel):
|
||||
name: str = Field(min_length=1, max_length=120)
|
||||
parent_id: int | None = None
|
||||
unit: str = Field(default="pcs", max_length=20)
|
||||
spec_template: list[SpecIn] = []
|
||||
notes: str = Field(default="", max_length=1000)
|
||||
sort_order: int = 0
|
||||
|
||||
|
||||
class LoginIn(BaseModel):
|
||||
password: str
|
||||
|
||||
|
||||
# --- helpers ----------------------------------------------------------------
|
||||
|
||||
def _tree_paths(conn, table: str) -> dict[int, str]:
|
||||
"""Map id -> 'Parent / Child' display path for a self-referencing table."""
|
||||
rows = conn.execute(f"SELECT id, name, parent_id FROM {table}").fetchall()
|
||||
by_id = {r["id"]: (r["name"], r["parent_id"]) for r in rows}
|
||||
paths: dict[int, str] = {}
|
||||
|
||||
def resolve(node_id: int, seen: set[int]) -> str:
|
||||
if node_id in paths:
|
||||
return paths[node_id]
|
||||
name, parent = by_id[node_id]
|
||||
# `seen` guards against a cycle introduced by a bad re-parent.
|
||||
if parent and parent in by_id and parent not in seen:
|
||||
path = resolve(parent, seen | {node_id}) + " / " + name
|
||||
else:
|
||||
path = name
|
||||
paths[node_id] = path
|
||||
return path
|
||||
|
||||
for node_id in by_id:
|
||||
resolve(node_id, set())
|
||||
return paths
|
||||
|
||||
|
||||
def _descendants(conn, table: str, root_id: int) -> list[int]:
|
||||
"""A node id plus every id beneath it, so filtering by 'Electronics'
|
||||
catches parts filed under 'Electronics / Resistors'."""
|
||||
rows = conn.execute(f"SELECT id, parent_id FROM {table}").fetchall()
|
||||
children: dict[int, list[int]] = {}
|
||||
for r in rows:
|
||||
children.setdefault(r["parent_id"], []).append(r["id"])
|
||||
out, stack = [], [root_id]
|
||||
seen = set()
|
||||
while stack:
|
||||
node = stack.pop()
|
||||
if node in seen:
|
||||
continue
|
||||
seen.add(node)
|
||||
out.append(node)
|
||||
stack.extend(children.get(node, []))
|
||||
return out
|
||||
|
||||
|
||||
def _specs_for(conn, part_ids: list[int]) -> dict[int, list[dict]]:
|
||||
if not part_ids:
|
||||
return {}
|
||||
marks = ",".join("?" * len(part_ids))
|
||||
rows = conn.execute(
|
||||
f"SELECT part_id, key, value FROM part_specs WHERE part_id IN ({marks}) ORDER BY position, key",
|
||||
part_ids,
|
||||
).fetchall()
|
||||
out: dict[int, list[dict]] = {}
|
||||
for r in rows:
|
||||
out.setdefault(r["part_id"], []).append({"key": r["key"], "value": r["value"]})
|
||||
return out
|
||||
|
||||
|
||||
def _tags_for(conn, part_ids: list[int]) -> dict[int, list[str]]:
|
||||
if not part_ids:
|
||||
return {}
|
||||
marks = ",".join("?" * len(part_ids))
|
||||
rows = conn.execute(
|
||||
f"""SELECT pt.part_id, t.name FROM part_tags pt
|
||||
JOIN tags t ON t.id = pt.tag_id
|
||||
WHERE pt.part_id IN ({marks}) ORDER BY t.name""",
|
||||
part_ids,
|
||||
).fetchall()
|
||||
out: dict[int, list[str]] = {}
|
||||
for r in rows:
|
||||
out.setdefault(r["part_id"], []).append(r["name"])
|
||||
return out
|
||||
|
||||
|
||||
def _serialise(rows, specs, tags, cat_paths, loc_paths) -> list[dict]:
|
||||
out = []
|
||||
for r in rows:
|
||||
d = dict(r)
|
||||
d["specs"] = specs.get(r["id"], [])
|
||||
d["tags"] = tags.get(r["id"], [])
|
||||
d["category_path"] = cat_paths.get(r["category_id"]) if r["category_id"] else None
|
||||
d["location_path"] = loc_paths.get(r["location_id"]) if r["location_id"] else None
|
||||
d["low_stock"] = (
|
||||
r["min_quantity"] is not None and r["quantity"] <= r["min_quantity"]
|
||||
)
|
||||
out.append(d)
|
||||
return out
|
||||
|
||||
|
||||
def _write_specs(conn, part_id: int, specs: list[SpecIn]):
|
||||
conn.execute("DELETE FROM part_specs WHERE part_id = ?", (part_id,))
|
||||
seen = set()
|
||||
position = 0
|
||||
for s in specs:
|
||||
key = s.key.strip()
|
||||
if not key or key.lower() in seen:
|
||||
continue
|
||||
seen.add(key.lower())
|
||||
conn.execute(
|
||||
"INSERT INTO part_specs(part_id, key, value, position) VALUES (?,?,?,?)",
|
||||
(part_id, key, s.value.strip(), position),
|
||||
)
|
||||
position += 1
|
||||
|
||||
|
||||
def _write_tags(conn, part_id: int, tags: list[str]):
|
||||
conn.execute("DELETE FROM part_tags WHERE part_id = ?", (part_id,))
|
||||
for raw in tags:
|
||||
name = raw.strip()
|
||||
if not name:
|
||||
continue
|
||||
conn.execute("INSERT OR IGNORE INTO tags(name) VALUES (?)", (name,))
|
||||
row = conn.execute("SELECT id FROM tags WHERE name = ? COLLATE NOCASE", (name,)).fetchone()
|
||||
if row:
|
||||
conn.execute(
|
||||
"INSERT OR IGNORE INTO part_tags(part_id, tag_id) VALUES (?,?)",
|
||||
(part_id, row["id"]),
|
||||
)
|
||||
|
||||
|
||||
def _fetch_part(conn, part_id: int) -> dict:
|
||||
row = conn.execute("SELECT * FROM parts WHERE id = ?", (part_id,)).fetchone()
|
||||
if row is None:
|
||||
raise HTTPException(404, "Part not found")
|
||||
cat_paths = _tree_paths(conn, "categories")
|
||||
loc_paths = _tree_paths(conn, "locations")
|
||||
return _serialise(
|
||||
[row], _specs_for(conn, [part_id]), _tags_for(conn, [part_id]), cat_paths, loc_paths
|
||||
)[0]
|
||||
|
||||
|
||||
# --- auth routes ------------------------------------------------------------
|
||||
|
||||
@app.get("/api/me")
|
||||
def me(request: Request):
|
||||
if not auth.auth_enabled():
|
||||
return {"authenticated": True, "auth_required": False}
|
||||
token = request.cookies.get(auth.COOKIE_NAME)
|
||||
return {
|
||||
"authenticated": bool(token and auth.token_valid(token)),
|
||||
"auth_required": True,
|
||||
}
|
||||
|
||||
|
||||
@app.post("/api/login")
|
||||
def login(body: LoginIn, response: Response, request: Request):
|
||||
if not auth.auth_enabled():
|
||||
return {"authenticated": True, "auth_required": False}
|
||||
if not auth.check_password(body.password):
|
||||
# A flat delay blunts online guessing without needing a rate-limit store.
|
||||
time.sleep(1.0)
|
||||
raise HTTPException(401, "Incorrect password")
|
||||
response.set_cookie(
|
||||
auth.COOKIE_NAME,
|
||||
auth.issue_token(),
|
||||
max_age=auth.session_days() * 86400,
|
||||
httponly=True,
|
||||
samesite="lax",
|
||||
secure=request.url.scheme == "https",
|
||||
path="/",
|
||||
)
|
||||
return {"authenticated": True, "auth_required": True}
|
||||
|
||||
|
||||
@app.post("/api/logout")
|
||||
def logout(response: Response):
|
||||
response.delete_cookie(auth.COOKIE_NAME, path="/")
|
||||
return {"authenticated": False}
|
||||
|
||||
|
||||
# --- parts ------------------------------------------------------------------
|
||||
|
||||
@app.get("/api/parts", dependencies=[Depends(auth.require_auth)])
|
||||
def list_parts(
|
||||
conn=Depends(db.get_db),
|
||||
q: str = "",
|
||||
category_id: int | None = None,
|
||||
location_id: int | None = None,
|
||||
tag: str = "",
|
||||
low_stock: bool = False,
|
||||
sort: Literal["relevance", "name", "quantity", "updated", "created", "location"] = "relevance",
|
||||
limit: int = Query(default=200, ge=1, le=1000),
|
||||
offset: int = Query(default=0, ge=0),
|
||||
):
|
||||
where: list[str] = []
|
||||
params: list[Any] = []
|
||||
joins = ""
|
||||
order = "p.name COLLATE NOCASE ASC"
|
||||
|
||||
q = q.strip()
|
||||
if q:
|
||||
match = db.fts_query(q)
|
||||
if match:
|
||||
joins += " JOIN parts_fts ON parts_fts.rowid = p.id "
|
||||
where.append("parts_fts MATCH ?")
|
||||
params.append(match)
|
||||
if sort == "relevance":
|
||||
order = "parts_fts.rank ASC, p.name COLLATE NOCASE ASC"
|
||||
|
||||
if category_id is not None:
|
||||
ids = _descendants(conn, "categories", category_id)
|
||||
where.append(f"p.category_id IN ({','.join('?' * len(ids))})")
|
||||
params.extend(ids)
|
||||
|
||||
if location_id is not None:
|
||||
ids = _descendants(conn, "locations", location_id)
|
||||
where.append(f"p.location_id IN ({','.join('?' * len(ids))})")
|
||||
params.extend(ids)
|
||||
|
||||
if tag.strip():
|
||||
where.append(
|
||||
"p.id IN (SELECT pt.part_id FROM part_tags pt JOIN tags t ON t.id = pt.tag_id "
|
||||
"WHERE t.name = ? COLLATE NOCASE)"
|
||||
)
|
||||
params.append(tag.strip())
|
||||
|
||||
if low_stock:
|
||||
where.append("p.min_quantity IS NOT NULL AND p.quantity <= p.min_quantity")
|
||||
|
||||
if sort == "name":
|
||||
order = "p.name COLLATE NOCASE ASC"
|
||||
elif sort == "quantity":
|
||||
order = "p.quantity ASC, p.name COLLATE NOCASE ASC"
|
||||
elif sort == "updated":
|
||||
order = "p.updated_at DESC, p.id DESC"
|
||||
elif sort == "created":
|
||||
order = "p.created_at DESC, p.id DESC"
|
||||
elif sort == "location":
|
||||
order = "l.name COLLATE NOCASE ASC, p.name COLLATE NOCASE ASC"
|
||||
|
||||
clause = (" WHERE " + " AND ".join(where)) if where else ""
|
||||
base = f"FROM parts p LEFT JOIN locations l ON l.id = p.location_id {joins} {clause}"
|
||||
|
||||
total = conn.execute(f"SELECT COUNT(*) AS n {base}", params).fetchone()["n"]
|
||||
rows = conn.execute(
|
||||
f"SELECT p.* {base} ORDER BY {order} LIMIT ? OFFSET ?", params + [limit, offset]
|
||||
).fetchall()
|
||||
|
||||
ids = [r["id"] for r in rows]
|
||||
items = _serialise(
|
||||
rows,
|
||||
_specs_for(conn, ids),
|
||||
_tags_for(conn, ids),
|
||||
_tree_paths(conn, "categories"),
|
||||
_tree_paths(conn, "locations"),
|
||||
)
|
||||
return {"total": total, "limit": limit, "offset": offset, "items": items}
|
||||
|
||||
|
||||
@app.post("/api/parts", dependencies=[Depends(auth.require_auth)], status_code=201)
|
||||
def create_part(body: PartIn, conn=Depends(db.get_db)):
|
||||
cur = conn.execute(
|
||||
"""INSERT INTO parts(name, description, category_id, location_id, manufacturer, mpn,
|
||||
quantity, unit, min_quantity, cost_each, datasheet_url,
|
||||
product_url, notes)
|
||||
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""",
|
||||
(
|
||||
body.name.strip(), body.description.strip(), body.category_id, body.location_id,
|
||||
body.manufacturer.strip(), body.mpn.strip(), body.quantity, body.unit.strip() or "pcs",
|
||||
body.min_quantity, body.cost_each, body.datasheet_url.strip(),
|
||||
body.product_url.strip(), body.notes.strip(),
|
||||
),
|
||||
)
|
||||
part_id = cur.lastrowid
|
||||
_write_specs(conn, part_id, body.specs)
|
||||
_write_tags(conn, part_id, body.tags)
|
||||
if body.quantity:
|
||||
conn.execute(
|
||||
"INSERT INTO stock_log(part_id, delta, quantity_after, reason) VALUES (?,?,?,?)",
|
||||
(part_id, body.quantity, body.quantity, "initial stock"),
|
||||
)
|
||||
db.reindex_part(conn, part_id)
|
||||
return _fetch_part(conn, part_id)
|
||||
|
||||
|
||||
@app.get("/api/parts/{part_id}", dependencies=[Depends(auth.require_auth)])
|
||||
def get_part(part_id: int, conn=Depends(db.get_db)):
|
||||
return _fetch_part(conn, part_id)
|
||||
|
||||
|
||||
@app.patch("/api/parts/{part_id}", dependencies=[Depends(auth.require_auth)])
|
||||
def update_part(part_id: int, body: PartPatch, conn=Depends(db.get_db)):
|
||||
if conn.execute("SELECT 1 FROM parts WHERE id = ?", (part_id,)).fetchone() is None:
|
||||
raise HTTPException(404, "Part not found")
|
||||
|
||||
fields = body.model_dump(exclude_unset=True)
|
||||
fields.pop("specs", None)
|
||||
fields.pop("tags", None)
|
||||
columns = [
|
||||
"name", "description", "category_id", "location_id", "manufacturer", "mpn",
|
||||
"quantity", "unit", "min_quantity", "cost_each", "datasheet_url", "product_url", "notes",
|
||||
]
|
||||
sets, params = [], []
|
||||
for col in columns:
|
||||
if col in fields:
|
||||
value = fields[col]
|
||||
if isinstance(value, str):
|
||||
value = value.strip()
|
||||
sets.append(f"{col} = ?")
|
||||
params.append(value)
|
||||
if sets:
|
||||
sets.append("updated_at = datetime('now')")
|
||||
conn.execute(f"UPDATE parts SET {', '.join(sets)} WHERE id = ?", params + [part_id])
|
||||
|
||||
if body.specs is not None:
|
||||
_write_specs(conn, part_id, body.specs)
|
||||
if body.tags is not None:
|
||||
_write_tags(conn, part_id, body.tags)
|
||||
db.reindex_part(conn, part_id)
|
||||
return _fetch_part(conn, part_id)
|
||||
|
||||
|
||||
@app.delete("/api/parts/{part_id}", dependencies=[Depends(auth.require_auth)])
|
||||
def delete_part(part_id: int, conn=Depends(db.get_db)):
|
||||
cur = conn.execute("DELETE FROM parts WHERE id = ?", (part_id,))
|
||||
if cur.rowcount == 0:
|
||||
raise HTTPException(404, "Part not found")
|
||||
conn.execute("DELETE FROM parts_fts WHERE rowid = ?", (part_id,))
|
||||
return {"deleted": part_id}
|
||||
|
||||
|
||||
@app.post("/api/parts/{part_id}/adjust", dependencies=[Depends(auth.require_auth)])
|
||||
def adjust_part(part_id: int, body: AdjustIn, conn=Depends(db.get_db)):
|
||||
row = conn.execute("SELECT quantity FROM parts WHERE id = ?", (part_id,)).fetchone()
|
||||
if row is None:
|
||||
raise HTTPException(404, "Part not found")
|
||||
after = round(row["quantity"] + body.delta, 4)
|
||||
if after < 0:
|
||||
after = 0.0
|
||||
conn.execute(
|
||||
"UPDATE parts SET quantity = ?, updated_at = datetime('now') WHERE id = ?",
|
||||
(after, part_id),
|
||||
)
|
||||
conn.execute(
|
||||
"INSERT INTO stock_log(part_id, delta, quantity_after, reason) VALUES (?,?,?,?)",
|
||||
(part_id, body.delta, after, body.reason.strip()),
|
||||
)
|
||||
return _fetch_part(conn, part_id)
|
||||
|
||||
|
||||
@app.get("/api/parts/{part_id}/history", dependencies=[Depends(auth.require_auth)])
|
||||
def part_history(part_id: int, conn=Depends(db.get_db), limit: int = Query(default=50, ge=1, le=500)):
|
||||
rows = conn.execute(
|
||||
"SELECT * FROM stock_log WHERE part_id = ? ORDER BY id DESC LIMIT ?", (part_id, limit)
|
||||
).fetchall()
|
||||
return {"items": [dict(r) for r in rows]}
|
||||
|
||||
|
||||
# --- categories & locations -------------------------------------------------
|
||||
|
||||
def _node_payload(conn, table: str) -> list[dict]:
|
||||
paths = _tree_paths(conn, table)
|
||||
rows = conn.execute(
|
||||
f"SELECT * FROM {table} ORDER BY sort_order, name COLLATE NOCASE"
|
||||
).fetchall()
|
||||
column = "category_id" if table == "categories" else "location_id"
|
||||
direct = {
|
||||
r["k"]: r["n"]
|
||||
for r in conn.execute(
|
||||
f"SELECT {column} AS k, COUNT(*) AS n FROM parts GROUP BY k"
|
||||
).fetchall()
|
||||
if r["k"] is not None
|
||||
}
|
||||
# Counts roll up: "Electronics" reports everything filed under its children
|
||||
# too, matching what clicking it as a filter actually returns.
|
||||
children: dict[int, list[int]] = {}
|
||||
for r in rows:
|
||||
children.setdefault(r["parent_id"], []).append(r["id"])
|
||||
|
||||
def rollup(node_id: int, seen: set[int]) -> int:
|
||||
total = direct.get(node_id, 0)
|
||||
for child in children.get(node_id, []):
|
||||
if child not in seen:
|
||||
total += rollup(child, seen | {node_id})
|
||||
return total
|
||||
|
||||
out = []
|
||||
for r in rows:
|
||||
d = dict(r)
|
||||
d["path"] = paths.get(r["id"], r["name"])
|
||||
d["part_count"] = rollup(r["id"], set())
|
||||
d["direct_count"] = direct.get(r["id"], 0)
|
||||
if table == "categories":
|
||||
try:
|
||||
d["spec_template"] = json.loads(r["spec_template"] or "[]")
|
||||
except (TypeError, ValueError):
|
||||
d["spec_template"] = []
|
||||
out.append(d)
|
||||
out.sort(key=lambda d: d["path"].lower())
|
||||
return out
|
||||
|
||||
|
||||
@app.get("/api/categories", dependencies=[Depends(auth.require_auth)])
|
||||
def list_categories(conn=Depends(db.get_db)):
|
||||
return {"items": _node_payload(conn, "categories")}
|
||||
|
||||
|
||||
@app.post("/api/categories", dependencies=[Depends(auth.require_auth)], status_code=201)
|
||||
def create_category(body: NodeIn, conn=Depends(db.get_db)):
|
||||
template = json.dumps([{"key": s.key.strip()} for s in body.spec_template if s.key.strip()])
|
||||
try:
|
||||
cur = conn.execute(
|
||||
"INSERT INTO categories(name, parent_id, unit, spec_template, sort_order) VALUES (?,?,?,?,?)",
|
||||
(body.name.strip(), body.parent_id, body.unit.strip() or "pcs", template, body.sort_order),
|
||||
)
|
||||
except Exception:
|
||||
raise HTTPException(409, "A category with that name already exists here")
|
||||
return {"id": cur.lastrowid}
|
||||
|
||||
|
||||
@app.patch("/api/categories/{cat_id}", dependencies=[Depends(auth.require_auth)])
|
||||
def update_category(cat_id: int, body: NodeIn, conn=Depends(db.get_db)):
|
||||
template = json.dumps([{"key": s.key.strip()} for s in body.spec_template if s.key.strip()])
|
||||
if body.parent_id == cat_id:
|
||||
raise HTTPException(400, "A category cannot be its own parent")
|
||||
conn.execute(
|
||||
"UPDATE categories SET name=?, parent_id=?, unit=?, spec_template=?, sort_order=? WHERE id=?",
|
||||
(body.name.strip(), body.parent_id, body.unit.strip() or "pcs", template, body.sort_order, cat_id),
|
||||
)
|
||||
return {"id": cat_id}
|
||||
|
||||
|
||||
@app.delete("/api/categories/{cat_id}", dependencies=[Depends(auth.require_auth)])
|
||||
def delete_category(cat_id: int, conn=Depends(db.get_db)):
|
||||
# ON DELETE SET NULL means parts survive as uncategorised rather than vanish.
|
||||
conn.execute("DELETE FROM categories WHERE id = ?", (cat_id,))
|
||||
return {"deleted": cat_id}
|
||||
|
||||
|
||||
@app.get("/api/locations", dependencies=[Depends(auth.require_auth)])
|
||||
def list_locations(conn=Depends(db.get_db)):
|
||||
return {"items": _node_payload(conn, "locations")}
|
||||
|
||||
|
||||
@app.post("/api/locations", dependencies=[Depends(auth.require_auth)], status_code=201)
|
||||
def create_location(body: NodeIn, conn=Depends(db.get_db)):
|
||||
try:
|
||||
cur = conn.execute(
|
||||
"INSERT INTO locations(name, parent_id, notes, sort_order) VALUES (?,?,?,?)",
|
||||
(body.name.strip(), body.parent_id, body.notes.strip(), body.sort_order),
|
||||
)
|
||||
except Exception:
|
||||
raise HTTPException(409, "A location with that name already exists here")
|
||||
return {"id": cur.lastrowid}
|
||||
|
||||
|
||||
@app.patch("/api/locations/{loc_id}", dependencies=[Depends(auth.require_auth)])
|
||||
def update_location(loc_id: int, body: NodeIn, conn=Depends(db.get_db)):
|
||||
if body.parent_id == loc_id:
|
||||
raise HTTPException(400, "A location cannot be its own parent")
|
||||
conn.execute(
|
||||
"UPDATE locations SET name=?, parent_id=?, notes=?, sort_order=? WHERE id=?",
|
||||
(body.name.strip(), body.parent_id, body.notes.strip(), body.sort_order, loc_id),
|
||||
)
|
||||
return {"id": loc_id}
|
||||
|
||||
|
||||
@app.delete("/api/locations/{loc_id}", dependencies=[Depends(auth.require_auth)])
|
||||
def delete_location(loc_id: int, conn=Depends(db.get_db)):
|
||||
conn.execute("DELETE FROM locations WHERE id = ?", (loc_id,))
|
||||
return {"deleted": loc_id}
|
||||
|
||||
|
||||
@app.get("/api/tags", dependencies=[Depends(auth.require_auth)])
|
||||
def list_tags(conn=Depends(db.get_db)):
|
||||
rows = conn.execute(
|
||||
"""SELECT t.name, COUNT(pt.part_id) AS n FROM tags t
|
||||
LEFT JOIN part_tags pt ON pt.tag_id = t.id
|
||||
GROUP BY t.id ORDER BY t.name COLLATE NOCASE"""
|
||||
).fetchall()
|
||||
return {"items": [dict(r) for r in rows]}
|
||||
|
||||
|
||||
@app.get("/api/stats", dependencies=[Depends(auth.require_auth)])
|
||||
def stats(conn=Depends(db.get_db)):
|
||||
row = conn.execute(
|
||||
"""SELECT COUNT(*) AS parts,
|
||||
COALESCE(SUM(quantity * COALESCE(cost_each, 0)), 0) AS value,
|
||||
SUM(CASE WHEN min_quantity IS NOT NULL AND quantity <= min_quantity
|
||||
THEN 1 ELSE 0 END) AS low
|
||||
FROM parts"""
|
||||
).fetchone()
|
||||
return {
|
||||
"parts": row["parts"],
|
||||
"low_stock": row["low"] or 0,
|
||||
"estimated_value": round(row["value"] or 0, 2),
|
||||
}
|
||||
|
||||
|
||||
@app.get("/healthz")
|
||||
def healthz():
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
# --- frontend ---------------------------------------------------------------
|
||||
|
||||
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
|
||||
|
||||
|
||||
@app.get("/")
|
||||
def index():
|
||||
return FileResponse(os.path.join(STATIC_DIR, "index.html"))
|
||||
|
||||
|
||||
@app.exception_handler(404)
|
||||
def not_found(request: Request, exc):
|
||||
path = request.url.path
|
||||
if path.startswith("/api/") or path.startswith("/static/") or path == "/healthz":
|
||||
return JSONResponse({"detail": "Not found"}, status_code=404)
|
||||
return FileResponse(os.path.join(STATIC_DIR, "index.html"), status_code=200)
|
||||
Reference in New Issue
Block a user