bruno
2f2a5e9b4a
Introduce an admin portal (React + Nginx), WebSocket routing, and API versioning middleware with `/api/v1/` prefix deprecation. Add master API key authentication, new Prometheus metrics for AI token consumption and active WebSockets, and extend S3 config with a public endpoint URL. Update test paths and fixtures to align with the new routing structure.
35 lines
1.3 KiB
Python
35 lines
1.3 KiB
Python
"""
|
|
Client API keys management (requires admin scope)
|
|
"""
|
|
from typing import Any, Dict
|
|
from httpx import AsyncClient
|
|
|
|
from ..utils import with_retry
|
|
|
|
class AuthResource:
|
|
"""Resource for administering client accounts (admin only)."""
|
|
def __init__(self, client: AsyncClient):
|
|
self._client = client
|
|
|
|
async def _handle_request(self, method: str, path: str, **kwargs) -> Any:
|
|
async def _req():
|
|
resp = await self._client.request(method, path, **kwargs)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
return await with_retry(_req)
|
|
|
|
async def create_client(self, name: str, is_premium: bool = False, scopes: list[str] = None) -> Dict[str, Any]:
|
|
data = {"name": name, "is_premium": is_premium}
|
|
if scopes:
|
|
data["scopes"] = scopes
|
|
return await self._handle_request("POST", "/api/v1/auth/clients", json=data)
|
|
|
|
async def list_clients(self) -> list[Dict[str, Any]]:
|
|
return await self._handle_request("GET", "/api/v1/auth/clients")
|
|
|
|
async def rotate_key(self, client_id: str) -> Dict[str, str]:
|
|
return await self._handle_request("POST", f"/api/v1/auth/clients/{client_id}/rotate-key")
|
|
|
|
async def delete_client(self, client_id: str) -> None:
|
|
await self._handle_request("DELETE", f"/api/v1/auth/clients/{client_id}")
|