Spaces:
Running
Running
File size: 3,008 Bytes
55ecbeb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | 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()
@router.get("/names", response_model=Union[GoogleScopeNamesResponse, GoogleScopeNamesListResponse])
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],
)
|