Spaces:
Sleeping
Sleeping
Download app/services/api_gate.py from SkillArbitrage/UpworkAutomation: direct link, hf CLI and curl.
- Browser
- Download file 14 kB
-
https://huggingface.co/spaces/SkillArbitrage/UpworkAutomation/resolve/main/app/services/api_gate.py
- Command line
-
hf download hf://spaces/SkillArbitrage/UpworkAutomation/app/services/api_gate.py
-
curl -L -o api_gate.py https://huggingface.co/spaces/SkillArbitrage/UpworkAutomation/resolve/main/app/services/api_gate.py
14 kB
| """API Gate. | |
| Validates configuration before downstream screens are enabled. Surfaces the | |
| precise status strings from the build plan. Resolves and validates the | |
| configured text provider β one of OpenAI, Anthropic, Groq, or Gemini β by | |
| checking provider support, API-key presence, and model presence. | |
| Configuration-level checks only: no expensive or live API calls are made here | |
| unless ``live=True`` is requested. A lightweight format-only placeholder check | |
| is applied to the Anthropic key specifically (``sk-ant-`` prefix); the other | |
| providers rely on the optional live reachability check. | |
| :func:`run_interview_capability_test` is the same machinery aimed at the | |
| interview provider. It is deliberately a separate entry point so it can be | |
| run when the user opens the interview and never at app start: the dossier | |
| route must keep working with no interview key configured. | |
| The API key value is never logged, displayed, or included in error output. | |
| """ | |
| from __future__ import annotations | |
| import inspect | |
| from dataclasses import dataclass | |
| from app import config as app_config | |
| from app.config import Settings, get_settings | |
| _SUPPORTED_PROVIDERS = ("anthropic", "openai", "groq", "gemini") | |
| # The interview leg is pinned to one provider. ``app.config`` owns the | |
| # canonical value; the literal fallback keeps this module importable against | |
| # a configuration that predates the constant. | |
| INTERVIEW_PROVIDER: str = ( | |
| str(getattr(app_config, "INTERVIEW_PROVIDER", "groq") or "").strip().lower() | |
| or "groq" | |
| ) | |
| class ApiGateError(str): | |
| """Status strings returned by the gate. Mirrors the build plan exactly.""" | |
| NO_API_ADDED = "NO API ADDED" | |
| INVALID_API_KEY = "INVALID API KEY" | |
| # The key works but may not use this model (HTTP 403): a plan, billing | |
| # or organisation-verification gate, which re-copying the key won't fix. | |
| ACCESS_DENIED = "API ACCESS DENIED" | |
| API_QUOTA_EXCEEDED = "API QUOTA EXCEEDED" | |
| API_CONNECTION_FAILED = "API CONNECTION FAILED" | |
| VISION_MODEL_NOT_AVAILABLE = "VISION MODEL NOT AVAILABLE" | |
| MODEL_NOT_CONFIGURED = "MODEL NOT CONFIGURED" | |
| # A model IS set, but the provider did not accept it (retired or | |
| # misspelt name). Kept apart from MODEL_NOT_CONFIGURED, which means no | |
| # model was set at all. | |
| MODEL_NOT_ACCEPTED = "MODEL NOT ACCEPTED" | |
| API_OK = "API OK" | |
| class CapabilityTestResult: | |
| status: str | |
| model: str | None = None | |
| provider: str | None = None | |
| def ok(self) -> bool: | |
| return self.status == ApiGateError.API_OK | |
| def error(self) -> str | None: | |
| return None if self.ok else self.status | |
| def _resolve_provider(provider: str | None, settings: Settings) -> str: | |
| raw = provider if provider is not None else settings.llm_provider | |
| return (raw or "").strip().lower() | |
| def _resolve_api_key(api_key: str | None, provider: str, settings: Settings) -> str | None: | |
| if api_key is not None: | |
| key = api_key.strip() | |
| return key or None | |
| if provider == "anthropic": | |
| return settings.anthropic_api_key | |
| if provider == "openai": | |
| return settings.openai_api_key | |
| if provider == "groq": | |
| return settings.groq_api_key | |
| if provider == "gemini": | |
| return settings.gemini_api_key | |
| return None | |
| def _resolve_model(model: str | None, provider: str, settings: Settings) -> str | None: | |
| if model is not None: | |
| m = model.strip() | |
| return m or None | |
| if provider == "anthropic": | |
| return settings.anthropic_model | |
| if provider == "openai": | |
| return settings.openai_model | |
| if provider == "groq": | |
| return settings.groq_model | |
| if provider == "gemini": | |
| return settings.gemini_model | |
| return None | |
| def _anthropic_key_format_ok(api_key: str) -> bool: | |
| """Cheap structural check β no network call. | |
| The Anthropic console issues keys prefixed with `sk-ant-`. This is a | |
| format sanity check only; it does not prove the key is live. | |
| """ | |
| key = api_key.strip() | |
| return key.startswith("sk-ant-") and len(key) >= 20 | |
| def run_capability_test( | |
| api_key: str | None = None, | |
| model: str | None = None, | |
| provider: str | None = None, | |
| settings: Settings | None = None, | |
| *, | |
| live: bool = False, | |
| route_provider: bool = False, | |
| task_name: str = "api_check", | |
| ) -> CapabilityTestResult: | |
| """Run configuration-level checks against the resolved provider. | |
| Returns a CapabilityTestResult whose `status` is one of the build-plan | |
| status strings. Never raises on bad input; never echoes the API key. | |
| When ``live=True``, after the offline checks succeed, route through | |
| :func:`app.services.llm_client.call_text_llm` with task ``api_check`` | |
| and the prompt ``"Reply with exactly: API OK"``. The live response | |
| only has to be received β content matching is tolerant β so this | |
| works as a lightweight reachability check. | |
| ``route_provider`` aims that live ping at the resolved provider instead | |
| of whichever one Settings has selected. It defaults to False so the | |
| Setup gate keeps testing exactly what it tested before. | |
| """ | |
| settings = settings if settings is not None else get_settings() | |
| resolved_provider = _resolve_provider(provider, settings) | |
| if not resolved_provider: | |
| return CapabilityTestResult(status=ApiGateError.MODEL_NOT_CONFIGURED) | |
| if resolved_provider not in _SUPPORTED_PROVIDERS: | |
| return CapabilityTestResult( | |
| status=ApiGateError.MODEL_NOT_CONFIGURED, | |
| provider=resolved_provider, | |
| ) | |
| resolved_key = _resolve_api_key(api_key, resolved_provider, settings) | |
| if not resolved_key: | |
| return CapabilityTestResult( | |
| status=ApiGateError.NO_API_ADDED, | |
| provider=resolved_provider, | |
| ) | |
| resolved_model = _resolve_model(model, resolved_provider, settings) | |
| if not resolved_model: | |
| return CapabilityTestResult( | |
| status=ApiGateError.MODEL_NOT_CONFIGURED, | |
| provider=resolved_provider, | |
| ) | |
| if resolved_provider == "anthropic" and not _anthropic_key_format_ok(resolved_key): | |
| return CapabilityTestResult( | |
| status=ApiGateError.INVALID_API_KEY, | |
| provider=resolved_provider, | |
| model=resolved_model, | |
| ) | |
| if live: | |
| return _run_live_check( | |
| api_key=resolved_key, | |
| model=resolved_model, | |
| provider=resolved_provider, | |
| settings=settings, | |
| route_provider=route_provider, | |
| task_name=task_name, | |
| ) | |
| return CapabilityTestResult( | |
| status=ApiGateError.API_OK, | |
| provider=resolved_provider, | |
| model=resolved_model, | |
| ) | |
| # --------------------------------------------------------------------------- | |
| # Live reachability check | |
| # --------------------------------------------------------------------------- | |
| _LIVE_STATUS_MAP: dict[str, str] = { | |
| # llm_client.STATUS_* β ApiGateError.* | |
| "ok": ApiGateError.API_OK, | |
| "no_api": ApiGateError.NO_API_ADDED, | |
| "invalid_key": ApiGateError.INVALID_API_KEY, | |
| "access_denied": ApiGateError.ACCESS_DENIED, | |
| "quota": ApiGateError.API_QUOTA_EXCEEDED, | |
| # A context-window overflow is a prompt-size problem, NOT a billing | |
| # quota; it must never be reported as "quota exceeded". For the tiny | |
| # api_check ping it also proves the key/endpoint work, so the gate | |
| # treats it as reachable. | |
| "context_overflow": ApiGateError.API_OK, | |
| "connection": ApiGateError.API_CONNECTION_FAILED, | |
| # The ping went out with the model that is set and the provider turned | |
| # it down (404 / model_not_found) β not the same as no model being set. | |
| "model_missing": ApiGateError.MODEL_NOT_ACCEPTED, | |
| "unsupported_provider": ApiGateError.MODEL_NOT_CONFIGURED, | |
| # parse_error and generic exception both indicate the call reached | |
| # the provider but didn't come back clean β treat as connection issue. | |
| "parse_error": ApiGateError.API_CONNECTION_FAILED, | |
| # A truncated ping reply still proves the provider is reachable and the | |
| # key works (the api_check call only needs *a* response). | |
| "truncated": ApiGateError.API_OK, | |
| "exception": ApiGateError.API_CONNECTION_FAILED, | |
| "skipped": ApiGateError.API_CONNECTION_FAILED, | |
| } | |
| def _supported_call_overrides() -> frozenset[str]: | |
| """Per-call override arguments ``call_text_llm`` accepts.""" | |
| # Import lazily to avoid a circular import at module load. | |
| from app.services import llm_client | |
| try: | |
| params = inspect.signature(llm_client.call_text_llm).parameters | |
| except (TypeError, ValueError): # defensive: never break the gate | |
| return frozenset() | |
| return frozenset({"provider", "model"} & set(params)) | |
| def _run_live_check( | |
| *, | |
| api_key: str, | |
| model: str, | |
| provider: str, | |
| settings: Settings, | |
| route_provider: bool = False, | |
| task_name: str = "api_check", | |
| ) -> CapabilityTestResult: | |
| """Lightweight real-API check via the central llm_client. | |
| With ``route_provider`` the ping is sent to ``provider`` rather than the | |
| one selected in Settings, so a check for one provider can run while a | |
| different one stays configured for the rest of the app. | |
| """ | |
| # Import lazily to avoid a circular import at module load. | |
| from app.services import llm_client | |
| extra: dict[str, str] = {} | |
| if route_provider: | |
| overrides = _supported_call_overrides() | |
| if "provider" not in overrides: | |
| # Without per-call routing the ping would test the configured | |
| # provider and the answer would be reported as if it came from | |
| # this one β a rejected key elsewhere would wrongly block this | |
| # provider. The offline checks above stand on their own. | |
| return CapabilityTestResult( | |
| status=ApiGateError.API_OK, | |
| provider=provider, | |
| model=model, | |
| ) | |
| extra["provider"] = provider | |
| if "model" in overrides: | |
| # Ping the exact model this check reports on, so a retired or | |
| # misspelt name surfaces here rather than mid-flow. | |
| extra["model"] = model | |
| result = llm_client.call_text_llm( | |
| task_name=task_name, | |
| system_prompt="", | |
| user_prompt="Reply with exactly: API OK", | |
| expected_json=False, | |
| max_tokens=16, | |
| settings=settings, | |
| **extra, | |
| ) | |
| if result.success: | |
| return CapabilityTestResult( | |
| status=ApiGateError.API_OK, | |
| provider=provider, | |
| model=model, | |
| ) | |
| mapped = _LIVE_STATUS_MAP.get(result.status, ApiGateError.API_CONNECTION_FAILED) | |
| if result.status == "model_missing" and not getattr(result, "used_api", True): | |
| # The client found no model to send, so nothing was rejected. | |
| mapped = ApiGateError.MODEL_NOT_CONFIGURED | |
| return CapabilityTestResult( | |
| status=mapped, | |
| provider=provider, | |
| model=model, | |
| ) | |
| # --------------------------------------------------------------------------- | |
| # Interview provider check | |
| # --------------------------------------------------------------------------- | |
| def _resolve_interview_model(model: str | None, settings: Settings) -> str | None: | |
| """Return the model the interview will use. | |
| The dedicated ``interview_model`` setting wins; otherwise the interview | |
| provider's own configured model is used. | |
| """ | |
| if model is not None: | |
| stripped = model.strip() | |
| return stripped or None | |
| configured = getattr(settings, "interview_model", None) | |
| if isinstance(configured, str) and configured.strip(): | |
| return configured.strip() | |
| return _resolve_model(None, INTERVIEW_PROVIDER, settings) | |
| def run_interview_capability_test( | |
| api_key: str | None = None, | |
| model: str | None = None, | |
| settings: Settings | None = None, | |
| *, | |
| live: bool = False, | |
| prefer: str | None = None, | |
| ) -> CapabilityTestResult: | |
| """Check the interview provider, for use on entry to the interview. | |
| Kept apart from :func:`run_capability_test`, which gates the whole app | |
| at Setup. This one must never run at app start: a dossier user with no | |
| interview key configured has to reach every other screen unaffected. | |
| The verdict uses the same vocabulary as the Setup gate, so the two | |
| failures that read alike to a user stay apart β an exhausted free tier | |
| comes back as ``API_QUOTA_EXCEEDED`` while a rejected key comes back as | |
| ``INVALID_API_KEY``. With ``live=False`` (the default) only the offline | |
| checks run: provider supported, key present, model set. | |
| ``api_key`` overrides the offline key-presence check only; as on the | |
| Setup check, the live ping resolves the key from Settings. | |
| """ | |
| settings = settings if settings is not None else get_settings() | |
| # Probe whatever the interview will actually run on: a Groq key if one | |
| # is set, otherwise the provider from Setup (see | |
| # ``llm_client.interview_route``). With neither, the Groq slot is | |
| # checked, which reports the missing key. | |
| from app.services import llm_client # lazy: avoids an import cycle | |
| route = llm_client.interview_route(settings, prefer=prefer) | |
| if route is None: | |
| provider = INTERVIEW_PROVIDER | |
| resolved_model = _resolve_interview_model(model, settings) | |
| else: | |
| provider, route_model = route | |
| explicit = (model or "").strip() if model is not None else "" | |
| resolved_model = ( | |
| explicit or route_model or _resolve_model(None, provider, settings) | |
| ) | |
| return run_capability_test( | |
| api_key=api_key, | |
| model=resolved_model, | |
| provider=provider, | |
| settings=settings, | |
| live=live, | |
| route_provider=True, | |
| # Logged apart from the Setup check, so a failed interview key is | |
| # never reported as a problem with the main API. | |
| task_name="interview_check", | |
| ) | |