Spaces:
Running
Running
| from __future__ import annotations | |
| from typing import Any, Dict, List, Optional | |
| from app.services.google_scopes_data import GOOGLE_APIS | |
| # OIDC short names that are conventionally sent without a full URI. They are | |
| # already present as scope names for the oauth_signin plan, but accepting them | |
| # explicitly keeps validation predictable for all APIs. | |
| _OIDC_SHORT_NAMES = frozenset({"openid", "email", "profile"}) | |
| class GoogleScopeError(Exception): | |
| """Raised for invalid scope configuration requests.""" | |
| def __init__(self, message: str, status_code: int = 400) -> None: | |
| super().__init__(message) | |
| self.message = message | |
| self.status_code = status_code | |
| def _default_scopes(plan: Dict[str, Any]) -> List[str]: | |
| """Derive the recommended default scopes for a plan.""" | |
| pn = plan.get("production_notes") or {} | |
| dflt = pn.get("default_scope_request") | |
| if dflt: | |
| return [s.strip() for s in dflt.split() if s.strip()] | |
| recs = plan.get("recommended_combinations") or {} | |
| if recs: | |
| first = next(iter(recs.values())) | |
| return list(first) | |
| return [] | |
| def _allowed_scopes(plan: Dict[str, Any]) -> frozenset: | |
| """Set of scope identifiers (URI or short name) accepted for a plan.""" | |
| allowed = set(_OIDC_SHORT_NAMES) | |
| for scope in plan.get("scopes") or []: | |
| if scope.get("uri"): | |
| allowed.add(scope["uri"]) | |
| if scope.get("name"): | |
| allowed.add(scope["name"]) | |
| return frozenset(allowed) | |
| class GoogleScopeService: | |
| """Serves the Google per-API scope catalog entirely from memory. | |
| The catalog is a Python object structure (``google_scopes_data.GOOGLE_APIS``) | |
| and the currently selected scopes are held as an in-memory dict of objects. | |
| No JSON files are read or written. | |
| """ | |
| def __init__(self) -> None: | |
| self._plans: Dict[str, Dict[str, Any]] = dict(GOOGLE_APIS) | |
| self._current_scopes: Optional[Dict[str, List[str]]] = None | |
| def catalog(self) -> Dict[str, Dict[str, Any]]: | |
| return self._plans | |
| def current_scopes(self) -> Dict[str, List[str]]: | |
| return self._state() | |
| def _state(self) -> Dict[str, List[str]]: | |
| if self._current_scopes is None: | |
| self._current_scopes = { | |
| pid: _default_scopes(plan) for pid, plan in self._plans.items() | |
| } | |
| return self._current_scopes | |
| async def load_all_plans(self) -> Dict[str, Dict[str, Any]]: | |
| return self._plans | |
| async def load_state(self) -> Dict[str, List[str]]: | |
| return self._state() | |
| async def set_current_scopes( | |
| self, updates: Dict[str, List[str]], reset: bool = False | |
| ) -> Dict[str, List[str]]: | |
| """Validate and store the requested current scopes in memory. | |
| Raises GoogleScopeError with a 400 status for unknown API ids or scopes | |
| that are not defined in the catalog. | |
| """ | |
| if reset: | |
| self._current_scopes = { | |
| pid: _default_scopes(plan) for pid, plan in self._plans.items() | |
| } | |
| return self._current_scopes | |
| current = self._state() | |
| errors: List[str] = [] | |
| for pid, scopes in (updates or {}).items(): | |
| if pid not in self._plans: | |
| errors.append(f"Unknown API id '{pid}'. Valid ids: {', '.join(sorted(self._plans))}") | |
| continue | |
| allowed = _allowed_scopes(self._plans[pid]) | |
| cleaned = [str(s).strip() for s in scopes if str(s).strip()] | |
| bad = [s for s in cleaned if s not in allowed] | |
| if bad: | |
| errors.append(f"Invalid scopes for '{pid}': {', '.join(bad)}") | |
| continue | |
| current[pid] = cleaned | |
| if errors: | |
| raise GoogleScopeError("; ".join(errors), status_code=400) | |
| self._current_scopes = current | |
| return current | |
| def _scope_text(scope: Dict[str, Any]) -> str: | |
| return " ".join( | |
| str(scope.get(k) or "") for k in ("name", "uri", "description") | |
| ).lower() | |
| def _scope_matches( | |
| scope: Dict[str, Any], | |
| *, | |
| scope_name: Optional[str] = None, | |
| permission_level: Optional[List[str]] = None, | |
| required: Optional[bool] = None, | |
| recommended: Optional[bool] = None, | |
| ) -> bool: | |
| if scope_name: | |
| needle = scope_name.lower() | |
| if needle not in (str(scope.get("name") or "").lower()) and needle not in ( | |
| str(scope.get("uri") or "").lower() | |
| ): | |
| return False | |
| if permission_level: | |
| level = str(scope.get("permission_level") or "").lower() | |
| if level not in {p.lower() for p in permission_level}: | |
| return False | |
| if required is not None and bool(scope.get("required")) is not required: | |
| return False | |
| if recommended is not None and bool(scope.get("recommended")) is not recommended: | |
| return False | |
| return True | |
| def filter_data( | |
| cls, | |
| data: List[Dict[str, Any]], | |
| *, | |
| api_name: Optional[str] = None, | |
| oauth_only: bool = False, | |
| no_scope: bool = False, | |
| search: Optional[str] = None, | |
| scope_name: Optional[str] = None, | |
| permission_level: Optional[List[str]] = None, | |
| required: Optional[bool] = None, | |
| recommended: Optional[bool] = None, | |
| scope_name_only: bool = False, | |
| ) -> List[Dict[str, Any]]: | |
| """Apply discoverability filters to the built catalog entries. | |
| - ``api_name`` keeps a single API (case-insensitive id match). | |
| - ``oauth_only`` / ``no_scope`` filter by whether scopes are required. | |
| - ``search`` free-text matches the API id/name and scope fields. | |
| - ``scope_name``, ``permission_level``, ``required``, ``recommended`` | |
| filter the ``scopes`` array of each entry; entries left with zero | |
| matching scopes are dropped unless ``api_name`` was requested. | |
| - ``scope_name_only`` replaces each ``scopes`` array with the list of | |
| scope names (falling back to the URI). | |
| """ | |
| api_name_l = api_name.strip().lower() if api_name else None | |
| scope_filters_active = any( | |
| [scope_name, permission_level, required is not None, recommended is not None] | |
| ) | |
| needle = search.strip().lower() if search else None | |
| levels = [p.strip().lower() for p in (permission_level or []) if p.strip()] | |
| filtered: List[Dict[str, Any]] = [] | |
| for entry in data: | |
| if api_name_l and str(entry.get("id") or "").lower() != api_name_l: | |
| continue | |
| if oauth_only and not entry.get("scope_required"): | |
| continue | |
| if no_scope and entry.get("scope_required"): | |
| continue | |
| haystack = ( | |
| f"{entry.get('id') or ''} {entry.get('api') or ''} " | |
| + " ".join(cls._scope_text(s) for s in entry.get("scopes") or []) | |
| ).lower() | |
| if needle and needle not in haystack: | |
| continue | |
| scopes = entry.get("scopes") or [] | |
| if scope_filters_active: | |
| scopes = [ | |
| s for s in scopes | |
| if cls._scope_matches( | |
| s, | |
| scope_name=scope_name, | |
| permission_level=levels or None, | |
| required=required, | |
| recommended=recommended, | |
| ) | |
| ] | |
| if not scopes and not api_name_l: | |
| continue | |
| item = dict(entry) | |
| if scope_name_only: | |
| item["scopes"] = [str(s.get("name") or s.get("uri")) for s in scopes] | |
| else: | |
| item["scopes"] = scopes | |
| filtered.append(item) | |
| return filtered | |
| def build_data( | |
| plans: Dict[str, Dict[str, Any]], | |
| state: Dict[str, List[str]], | |
| ) -> List[Dict[str, Any]]: | |
| """Merge catalog plans with current scopes into the API response payload.""" | |
| data: List[Dict[str, Any]] = [] | |
| for pid in sorted(plans): | |
| plan = plans[pid] | |
| data.append( | |
| { | |
| "id": pid, | |
| "api": plan.get("api"), | |
| "version": plan.get("version"), | |
| "base_url": plan.get("base_url"), | |
| "batch_endpoint": plan.get("batch_endpoint"), | |
| "auth_model": plan.get("auth_model"), | |
| "scope_required": plan.get("scope_required", False), | |
| "docs_url": plan.get("docs_url"), | |
| "info": plan.get("info"), | |
| "current_scopes": state.get(pid, []), | |
| "default_scopes": _default_scopes(plan), | |
| "scopes": plan.get("scopes", []), | |
| "recommended_combinations": plan.get("recommended_combinations", {}), | |
| } | |
| ) | |
| return data | |