"""Clean FastAPI app (greenfield P0.2) for SlabUploader. APP-30/45: /api/health returns {status,version,woo_reachable}; non-200 when storage is not writable (small probe helper). APP-55/65: stdlib structured JSON logging (ts,level,logger,request_id,msg) to stdout; request_id from X-Request-Id header (uuid4 if absent) via contextvars; secret-looking messages are redacted. SETTINGS-10/25: startup guard hard-refuses a missing/non-https WOO_BASE_URL or a missing SLAB_AES_KEY — UNLESS the environment is explicitly dev/demo (SLAB_ENV in {dev,demo}), where it logs a warning so a LAN dev box without store creds keeps running; production (default) must refuse. P3a (M3 Create backend): real server modules — Woo client, taxonomy sync, pricing rules, inference (on/off), review, publish, settings (SETTINGS-75 masking), slabs CRUD, photo storage. Woo/inference transports are injectable and mocked in tests (never hit a real store). """ from __future__ import annotations import json import logging import os import sys import uuid from contextlib import asynccontextmanager from contextvars import ContextVar from datetime import datetime, timezone from pathlib import Path from typing import Any from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from pydantic import BaseModel from starlette.middleware.base import BaseHTTPMiddleware from .db import get_connection, init_db __version__ = "0.6.0-gf-p3a" DEV_ENVS = {"dev", "demo"} # --- Structured JSON logging (stdlib only) --- request_id_var: ContextVar[str] = ContextVar("request_id", default="") _SECRET_MARKERS = ( "password", "api_key", "aes_key", "slab_aes_key", "woo_app_password", "inference_api_key", ) def _redact(msg: str) -> str: """APP-65: never let secret material reach the log line.""" lower = msg.lower() if any(marker in lower for marker in _SECRET_MARKERS): return "[REDACTED secret material]" return msg class JsonFormatter(logging.Formatter): """JSON lines with request_id from the contextvar; redacts secrets.""" def format(self, record: logging.LogRecord) -> str: data: dict[str, Any] = { "ts": self.formatTime(record, self.datefmt), "level": record.levelname, "logger": record.name, "request_id": getattr(record, "request_id", None) or request_id_var.get(""), "msg": _redact(record.getMessage()), } if record.exc_info: data["exc"] = self.formatException(record.exc_info) return json.dumps(data, ensure_ascii=False, separators=(",", ":")) def _configure_logging() -> None: level = os.getenv("LOG_LEVEL", "INFO").upper() handler = logging.StreamHandler(sys.stdout) handler.setFormatter(JsonFormatter()) root = logging.getLogger() root.handlers.clear() root.addHandler(handler) root.setLevel(getattr(logging, level, logging.INFO)) logging.getLogger("uvicorn.access").setLevel(logging.WARNING) _configure_logging() json_formatter = JsonFormatter() # exposed for tests logger = logging.getLogger("slab.backend") def _is_dev_env() -> bool: return os.getenv("SLAB_ENV", "").strip().lower() in DEV_ENVS # --- Startup guards (SETTINGS-10/25) --- def _enforce_startup_guards() -> None: """Hard-refuse bad Woo URL / missing AES key; warn in dev/demo so the slice runs.""" problems: list[str] = [] woo_url = os.getenv("WOO_BASE_URL", "") if not woo_url: problems.append("SETTINGS-10: WOO_BASE_URL is required (https://...); missing") elif not woo_url.startswith("https://"): problems.append("SETTINGS-10: WOO_BASE_URL must start with https:// (got non-https value)") if not os.getenv("SLAB_AES_KEY"): problems.append("SETTINGS-25: SLAB_AES_KEY is required (32-byte base64 in env/compose); missing") if not problems: return if _is_dev_env(): for p in problems: logger.warning("dev/demo env: startup guard downgraded — " + p) return for p in problems: logger.error(p) raise RuntimeError("; ".join(problems)) # --- Request ID middleware (APP-55) --- class RequestIdMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): rid = request.headers.get("X-Request-Id") or str(uuid.uuid4()) token = request_id_var.set(rid) try: response = await call_next(request) response.headers["X-Request-Id"] = rid return response finally: request_id_var.reset(token) # --- Storage probe (APP-45) --- def probe_storage_writable() -> bool: """True if the DB path's directory accepts a throwaway write.""" db_path = os.environ.get("SLAB_DB_PATH") or "slab_demo.db" directory = Path(db_path).expanduser().resolve().parent probe = directory / f".slab_health_{os.getpid()}" try: probe.write_text("ok", encoding="utf-8") return True except OSError: return False finally: try: probe.unlink() except OSError: pass # --- App --- @asynccontextmanager async def lifespan(_: FastAPI): # Guards run at startup: a misconfigured server refuses to serve traffic. _enforce_startup_guards() init_db() logger.info("startup: schema ready (idempotent)") yield app = FastAPI( title="SlabUploader", version=__version__, docs_url=None, redoc_url=None, lifespan=lifespan, ) # Permissive CORS is explicit for LAN dev (single trust domain on the LAN); # credentials are NOT allowed, and this is a dev-only posture. app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=False, allow_methods=["*"], allow_headers=["*"], ) app.add_middleware(RequestIdMiddleware) def _error(code: str, message: str, detail: Any = None) -> dict[str, Any]: return {"error": {"code": code, "message": message, "detail": detail}} def _envelope_response(status: int, detail: Any) -> JSONResponse: if isinstance(detail, dict) and "error" in detail: return JSONResponse(status_code=status, content=detail) code = "validation" if status == 400 else str(status) return JSONResponse(status_code=status, content=_error(code, str(detail))) @app.exception_handler(HTTPException) async def _envelope_http_exception(request: Request, exc: HTTPException) -> JSONResponse: return _envelope_response(exc.status_code, exc.detail) # --- Health (APP-30/45) --- @app.get("/api/health") async def health() -> JSONResponse: if not probe_storage_writable(): logger.warning("health: storage not writable") return JSONResponse(status_code=503, content=_error("storage", "storage not writable")) # woo_reachable stays a constant False: health must not depend on the # store (outage != down) and a dev box without creds (SLAB_ENV=dev) must # still report ok. An authenticated probe lands with a deployed store. return JSONResponse(content={ "status": "ok", "version": __version__, "woo_reachable": False, }) # --- M3 routers + settings/taxonomy endpoints (P3a) --- from . import settings as settings_mod # noqa: E402 (imported after app object; router modules need it) @app.get("/api/v1/settings") async def get_settings() -> dict[str, Any]: return settings_mod.get_settings() @app.put("/api/v1/settings") async def put_settings(payload: dict[str, Any]) -> dict[str, Any]: return settings_mod.update_settings(payload) from . import taxonomy as taxonomy_mod # noqa: E402 from .woo import ( # noqa: E402 WooAuthError, WooClient, WooConnectionError, WooCredentialsMissing, WooError, WooForbiddenError, ) def _woo_credentials() -> tuple[str, str | None, str | None]: conn = get_connection() try: base = os.environ.get("WOO_BASE_URL", "") user_row = conn.execute("SELECT value FROM settings WHERE key='woo_wp_username'").fetchone() pass_row = conn.execute("SELECT value_enc FROM settings WHERE key='woo_app_password'").fetchone() return base, (user_row[0] if user_row else None), (pass_row[0] if pass_row else None) finally: conn.close() @app.post("/api/v1/settings/test-woo") async def test_woo() -> dict[str, Any]: base, username, enc_password = _woo_credentials() try: client = WooClient.from_settings(base, username, enc_password) except WooCredentialsMissing as e: return {"ok": False, "detail": str(e), "taxonomy_anchor": None, "taxonomy_last_sync_at": None} try: result = taxonomy_mod.sync(client, get_connection()) except (WooAuthError, WooForbiddenError, WooConnectionError, WooError) as e: return {"ok": False, "detail": str(e), "taxonomy_anchor": None, "taxonomy_last_sync_at": None} return { "ok": True, "detail": "taxonomy synced", "taxonomy_anchor": result["taxonomy_anchor"], "taxonomy_last_sync_at": result["taxonomy_last_sync_at"], } @app.get("/api/v1/admin/taxonomy") async def get_taxonomy() -> dict[str, Any]: conn = get_connection() try: return taxonomy_mod.get_cached(conn) finally: conn.close() @app.post("/api/v1/admin/taxonomy/sync") async def post_taxonomy_sync() -> dict[str, Any]: base, username, enc_password = _woo_credentials() try: client = WooClient.from_settings(base, username, enc_password) except WooCredentialsMissing as e: return JSONResponse(status_code=502, content=_error("woo_unavailable", str(e))) try: result = taxonomy_mod.sync(client, get_connection()) except WooError as e: return JSONResponse(status_code=502, content=_error("woo_unavailable", str(e))) return { "ok": True, "detail": "taxonomy synced", "taxonomy_anchor": result["taxonomy_anchor"], "taxonomy_last_sync_at": result["taxonomy_last_sync_at"], } from .slabs import router as slabs_router # noqa: E402 from .photos import router as photos_router # noqa: E402 from .pricing import router as pricing_router # noqa: E402 from .review import router as review_router # noqa: E402 from .publish import router as publish_router # noqa: E402 from .inference import router as inference_router # noqa: E402 from .prompts import router as prompts_router # noqa: E402 app.include_router(slabs_router) app.include_router(photos_router) app.include_router(pricing_router) app.include_router(review_router) app.include_router(publish_router) app.include_router(inference_router) app.include_router(prompts_router) def reinit_db(path: str | None = None): """Re-init schema on `path` (test isolation); returns a connection to it. With the per-call factory (P0.2), each request opens its own connection, so there is no cached binding to re-point — init_db on the new path is the entire isolation. """ p = path or os.environ.get("SLAB_DB_PATH") or "slab_demo.db" init_db(p) return get_connection(p) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=18081)