Spaces:
Running
Running
File size: 6,185 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 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 | from __future__ import annotations
import time
from typing import List, Optional, Union
from fastapi import APIRouter, HTTPException, Query
from app.models.schemas import (
GoogleApiIdsResponse,
GoogleApiScopesResponse,
GoogleScopeNamesEntry,
GoogleScopeNamesListResponse,
GoogleScopeNamesResponse,
)
from app.services.google_scope_service import GoogleScopeError
from .deps import GENERATED, REFERENCE, SOURCE, get_scope_service
router = APIRouter()
_VALID_PERMISSION_LEVELS = frozenset({"non-sensitive", "sensitive", "restricted"})
def _filters_echo(
api_name: Optional[str],
oauth_only: bool,
no_scope: bool,
search: Optional[str],
scope_name: Optional[str],
permission_level: Optional[List[str]],
required: Optional[bool],
recommended: Optional[bool],
scope_name_only: bool,
ids_only: bool,
) -> dict:
return {
"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": scope_name_only,
"ids_only": ids_only,
}
def _validate_permission_levels(permission_level: Optional[List[str]]) -> None:
if permission_level:
bad = [p for p in permission_level if p.strip().lower() not in _VALID_PERMISSION_LEVELS]
if bad:
raise HTTPException(
status_code=422,
detail=f"Invalid permission_level values: {', '.join(bad)}. "
"Allowed: non-sensitive, sensitive, restricted.",
)
@router.get("", response_model=Union[GoogleApiScopesResponse, GoogleScopeNamesResponse, GoogleScopeNamesListResponse, GoogleApiIdsResponse])
async def list_google_scopes(
api_name: Optional[str] = Query(None, description="Filter to a single API id (e.g. gmail, youtube, drive)"),
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"),
scope_name_only: bool = Query(False, description="Return only scope names instead of full scope objects"),
ids_only: bool = Query(False, description="Return only the matching API ids"),
page: int = Query(1, ge=1, description="Page number"),
page_size: int = Query(50, ge=1, le=200, description="Items per page"),
) -> Union[GoogleApiScopesResponse, GoogleScopeNamesResponse, GoogleScopeNamesListResponse, GoogleApiIdsResponse]:
"""List all Google integrations with their OAuth scopes and current selection.
Supports combining every filter: ``api_name``, ``oauth_only``/``no_scope``,
``search``, ``scope_name``, ``permission_level``, ``required``,
``recommended``, ``scope_name_only``, ``ids_only`` and pagination.
"""
_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,
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=scope_name_only,
)
if ids_only:
return GoogleApiIdsResponse(
success=True,
time_ms=round((time.perf_counter() - started) * 1000, 3),
count=len(filtered),
ids=[e["id"] for e in filtered],
)
# Compact single-API shape for api_name + scope_name_only (see example).
if scope_name_only and 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"])
if scope_name_only:
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],
)
total = len(filtered)
total_pages = max(1, (total + page_size - 1) // page_size)
start = (page - 1) * page_size
chunk = filtered[start : start + page_size]
return GoogleApiScopesResponse(
success=True,
time_ms=round((time.perf_counter() - started) * 1000, 3),
count=len(chunk),
total=total,
page=page,
page_size=page_size,
total_pages=total_pages,
generated=GENERATED,
reference=REFERENCE,
source=SOURCE,
filters=_filters_echo(
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=scope_name_only,
ids_only=ids_only,
),
data=chunk,
)
|