(request: Request)
| 3252 | } |
| 3253 | |
| 3254 | async def handle_connections(request: Request) -> JSONResponse: |
| 3255 | denied = _admin_auth_failure(request) |
| 3256 | if denied is not None: |
| 3257 | return denied |
| 3258 | if request.method == "GET": |
| 3259 | return JSONResponse(_current_connection_payload()) |
| 3260 | |
| 3261 | effective = resolve_primary_connection( |
| 3262 | cli_upstream=_cli_upstream_override, |
| 3263 | store=_connections_store, |
| 3264 | ) |
| 3265 | if not effective.editable: |
| 3266 | return JSONResponse({ |
| 3267 | "error": "Primary upstream is externally managed", |
| 3268 | "source": effective.source, |
| 3269 | "upstream_source": effective.upstream_source, |
| 3270 | "api_key_source": effective.api_key_source, |
| 3271 | }, status_code=409) |
| 3272 | |
| 3273 | body = await request.json() |
| 3274 | next_upstream = str(body.get("upstream", upstream)).strip() |
| 3275 | next_api_key = str(body.get("api_key", _primary_api_key)).strip() |
| 3276 | ok, payload = await _reload_primary_connection( |
| 3277 | next_upstream=next_upstream, |
| 3278 | next_api_key=next_api_key, |
| 3279 | persist=True, |
| 3280 | ) |
| 3281 | return JSONResponse(payload, status_code=200 if ok else 502) |
| 3282 | |
| 3283 | async def handle_providers(request: Request) -> JSONResponse: |
| 3284 | denied = _admin_auth_failure(request) |
nothing calls this directly
no test coverage detected