GET /v1/scenes — list all scenes. POST /v1/scenes — add/remove/import.
(request: Request)
| 3574 | return JSONResponse(payload) |
| 3575 | |
| 3576 | async def handle_scenes(request: Request) -> JSONResponse: |
| 3577 | """GET /v1/scenes — list all scenes. POST /v1/scenes — add/remove/import.""" |
| 3578 | if request.method == "GET": |
| 3579 | return JSONResponse(_scene_store.export()) |
| 3580 | |
| 3581 | denied = _admin_auth_failure(request) |
| 3582 | if denied is not None: |
| 3583 | return denied |
| 3584 | body = await request.json() |
| 3585 | action = str(body.get("action", "")).strip().lower() |
| 3586 | try: |
| 3587 | if action == "add": |
| 3588 | name = str(body.get("name", "")).strip() |
| 3589 | primary = str(body.get("primary", "")).strip() |
| 3590 | if not name or not primary: |
| 3591 | return JSONResponse({"error": "name and primary are required"}, status_code=400) |
| 3592 | fallback_raw = body.get("fallback", []) |
| 3593 | if isinstance(fallback_raw, str): |
| 3594 | fallback = [p.strip() for p in fallback_raw.split(",") if p.strip()] |
| 3595 | elif isinstance(fallback_raw, list): |
| 3596 | fallback = [str(f).strip() for f in fallback_raw if str(f).strip()] |
| 3597 | else: |
| 3598 | fallback = [] |
| 3599 | tier_floor_raw = body.get("tier_floor") |
| 3600 | tier_floor = Tier(str(tier_floor_raw).upper()) if tier_floor_raw else None |
| 3601 | tier_cap_raw = body.get("tier_cap") |
| 3602 | tier_cap = Tier(str(tier_cap_raw).upper()) if tier_cap_raw else None |
| 3603 | allowed_providers_raw = body.get("allowed_providers", []) |
| 3604 | allowed_providers = ( |
| 3605 | [str(p).strip() for p in allowed_providers_raw if str(p).strip()] |
| 3606 | if isinstance(allowed_providers_raw, list) else [] |
| 3607 | ) |
| 3608 | max_cost_raw = body.get("max_cost_per_request") |
| 3609 | max_cost = float(max_cost_raw) if max_cost_raw is not None else None |
| 3610 | if max_cost is not None and max_cost <= 0: |
| 3611 | return JSONResponse({"error": "max_cost_per_request must be positive"}, status_code=400) |
| 3612 | scene = SceneConfig( |
| 3613 | name=name, |
| 3614 | primary=primary, |
| 3615 | fallback=fallback, |
| 3616 | hard_pin=bool(body.get("hard_pin", False)), |
| 3617 | description=str(body.get("description", "")), |
| 3618 | tier_floor=tier_floor, |
| 3619 | tier_cap=tier_cap, |
| 3620 | allowed_providers=allowed_providers, |
| 3621 | max_cost_per_request=max_cost, |
| 3622 | ) |
| 3623 | stored = _scene_store.add(scene) |
| 3624 | return JSONResponse({"ok": True, "scene": _serialize_scene_response(stored)}) |
| 3625 | elif action == "remove": |
| 3626 | name = str(body.get("name", "")).strip() |
| 3627 | if not name: |
| 3628 | return JSONResponse({"error": "name is required"}, status_code=400) |
| 3629 | removed = _scene_store.remove(name) |
| 3630 | return JSONResponse({"ok": removed, "name": name}) |
| 3631 | elif action == "import": |
| 3632 | data = body.get("data", {}) |
| 3633 | count = _scene_store.import_scenes(data) |
nothing calls this directly
no test coverage detected