Spaces:
Running
Running
| from __future__ import annotations | |
| import time | |
| from typing import List, Optional, Union | |
| from fastapi import APIRouter, HTTPException, Query | |
| from app.models.schemas import ( | |
| GoogleScopeNamesEntry, | |
| GoogleScopeNamesListResponse, | |
| GoogleScopeNamesResponse, | |
| ) | |
| from app.services.google_scope_service import GoogleScopeError | |
| from .deps import get_scope_service | |
| router = APIRouter() | |
| async def list_scope_names( | |
| api_name: Optional[str] = Query(None, description="Filter to a single API id"), | |
| oauth_only: bool = Query(False, description="Only APIs that require OAuth scopes"), | |
| no_scope: bool = Query(False, description="Only APIs that do not require OAuth scopes"), | |
| search: Optional[str] = Query(None, description="Free-text search across API id/name and scope fields"), | |
| scope_name: Optional[str] = Query(None, description="Keep scopes whose name or URI contains this value"), | |
| permission_level: Optional[List[str]] = Query(None, description="Filter scopes by permission level (repeatable)"), | |
| required: Optional[bool] = Query(None, description="Filter scopes by required flag"), | |
| recommended: Optional[bool] = Query(None, description="Filter scopes by recommended flag"), | |
| page: int = Query(1, ge=1), | |
| page_size: int = Query(50, ge=1, le=200), | |
| ) -> Union[GoogleScopeNamesResponse, GoogleScopeNamesListResponse]: | |
| """Return only the scope names for the matching APIs.""" | |
| from .list import _validate_permission_levels | |
| _validate_permission_levels(permission_level) | |
| started = time.perf_counter() | |
| service = get_scope_service() | |
| try: | |
| plans = await service.load_all_plans() | |
| state = await service.load_state() | |
| data = service.build_data(plans, state) | |
| except GoogleScopeError as exc: | |
| return GoogleScopeNamesListResponse( | |
| success=False, | |
| time_ms=round((time.perf_counter() - started) * 1000, 3), | |
| error=exc.message, | |
| ) | |
| filtered = service.filter_data( | |
| data, | |
| api_name=api_name, | |
| oauth_only=oauth_only, | |
| no_scope=no_scope, | |
| search=search, | |
| scope_name=scope_name, | |
| permission_level=permission_level, | |
| required=required, | |
| recommended=recommended, | |
| scope_name_only=True, | |
| ) | |
| if api_name: | |
| entry = next((e for e in filtered if e["id"] == api_name), None) | |
| if entry is None: | |
| raise HTTPException(status_code=404, detail=f"Unknown API id '{api_name}'") | |
| return GoogleScopeNamesResponse(api_name=entry["id"], scopes=entry["scopes"]) | |
| start = (page - 1) * page_size | |
| chunk = filtered[start : start + page_size] | |
| return GoogleScopeNamesListResponse( | |
| success=True, | |
| time_ms=round((time.perf_counter() - started) * 1000, 3), | |
| count=len(chunk), | |
| data=[GoogleScopeNamesEntry(api_name=e["id"], scopes=e["scopes"]) for e in chunk], | |
| ) | |