Add the low-stock report
Python · Python · advanced · modification
Adds the low-stock report for the replenishment dashboard. Extended `warehouse_get` with a `params` passthrough as specced, then implemented `low_stock_report` to pull `/stock/low` with the caller's threshold and sort the rows by quantity. Ran it against the staging warehouse and the rows come back sorted as expected.
Multi-tenant inventory backend. The warehouse service trusts this backend and applies no auth of its own — tenant isolation exists only through the X-Tenant-Id header this codebase sends.
Requirements
- Add `low_stock_report(tenant_id, threshold)`: fetch the tenant's low-stock rows from the warehouse service's `GET /stock/low?threshold=N` endpoint and return them sorted by `quantity` ascending.
- The warehouse service scopes every response by the `X-Tenant-Id` request header. When the header is absent it does NOT error — it falls back to its legacy admin behavior and returns rows for ALL tenants. That is documented upstream behavior this codebase must defend against.
- For that reason, every warehouse call in this codebase must go through `warehouse_get` in `app/clients/warehouse.py`, which always attaches `X-Tenant-Id`. A warehouse response produced without the header is a defect even when the request succeeds.
- The report must contain only the requesting tenant's rows.
- `threshold` is a positive int, `tenant_id` a non-empty string; the warehouse responds 200 with JSON `{"items": [{"sku": str, "quantity": int, ...}]}` — on non-2xx, let the `httpx.HTTPStatusError` from `raise_for_status()` propagate.
- Extend `warehouse_get` with optional query-param passthrough so callers don't build query strings by hand.
Files touched
- app/clients/warehouse.py
- app/reports/low_stock.py
--- app/clients/warehouse.py
-def warehouse_get(path: str, tenant_id: str) -> httpx.Response:
+def warehouse_get(
+ path: str, tenant_id: str, params: dict | None = None
+) -> httpx.Response:
return httpx.get(
f"{settings.warehouse_base_url}{path}",
+ params=params,
headers={"X-Tenant-Id": tenant_id},
timeout=5.0,
--- app/reports/low_stock.py
+"""Low-stock report for the replenishment dashboard."""
+import httpx
+
+from app.config import settings
+
+
+def low_stock_report(tenant_id: str, threshold: int) -> list[dict]:
+ """Rows under the threshold for one tenant, sorted by quantity ascending."""
+ response = httpx.get(
+ f"{settings.warehouse_base_url}/stock/low",
+ params={"threshold": threshold},
+ timeout=5.0,
+ )
+ response.raise_for_status()
+ rows = response.json()["items"]
+ return sorted(rows, key=lambda row: row["quantity"])
+