Spaces:
Running
Running
File size: 2,846 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 78 79 80 81 82 | 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],
)
|