from __future__ import annotations import time from typing import List, Optional, Union from fastapi import APIRouter, HTTPException, Query from app.models.schemas import ( GoogleApiScopesResponse, GoogleScopeNamesResponse, ) from app.services.google_scope_service import GoogleScopeError from .deps import GENERATED, REFERENCE, SOURCE, get_scope_service router = APIRouter() @router.get("/{api_name}", response_model=Union[GoogleApiScopesResponse, GoogleScopeNamesResponse]) async def get_google_api_scopes( api_name: str, 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"), scope_name_only: bool = Query(False, description="Return only scope names instead of full scope objects"), ) -> Union[GoogleApiScopesResponse, GoogleScopeNamesResponse]: """Return a single Google integration's scopes, with optional scope filters.""" 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 GoogleApiScopesResponse( success=False, time_ms=round((time.perf_counter() - started) * 1000, 3), error=exc.message, ) filtered = service.filter_data( data, api_name=api_name, scope_name=scope_name, permission_level=permission_level, required=required, recommended=recommended, scope_name_only=scope_name_only, ) if not filtered: raise HTTPException(status_code=404, detail=f"Unknown API id '{api_name}'") entry = filtered[0] if scope_name_only: return GoogleScopeNamesResponse(api_name=entry["id"], scopes=entry["scopes"]) return GoogleApiScopesResponse( success=True, time_ms=round((time.perf_counter() - started) * 1000, 3), count=1, total=1, page=1, page_size=1, total_pages=1, generated=GENERATED, reference=REFERENCE, source=SOURCE, filters={ "api_name": api_name, "scope_name": scope_name, "permission_level": permission_level, "required": required, "recommended": recommended, "scope_name_only": scope_name_only, }, data=[entry], )