nessi
58eab24258
Add FastAPI-based monitoring dashboard with support for Codex CLI and Devin API v3 Enterprise consumption tracking. Includes Docker deployment, systemd service configuration, kiosk mode launcher, comprehensive documentation, and demo mode for testing without credentials.
53 lines
2.0 KiB
Python
53 lines
2.0 KiB
Python
import asyncio
|
|
import time
|
|
from abc import ABC, abstractmethod
|
|
from datetime import datetime, timezone
|
|
|
|
from app.models import ProviderUsage
|
|
|
|
|
|
GENERIC_ERROR = 'Unable to retrieve usage information'
|
|
|
|
|
|
def utc_now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
class UsageProvider(ABC):
|
|
def __init__(self, provider: str, cache_ttl: float, source: str | None = None):
|
|
self.provider = provider
|
|
self.cache_ttl = cache_ttl
|
|
self.source = source
|
|
self.last_successful_update: datetime | None = None
|
|
self._cached: ProviderUsage | None = None
|
|
self._cache_time = 0.0
|
|
self._lock = asyncio.Lock()
|
|
|
|
async def get_usage(self) -> ProviderUsage:
|
|
if self._cached is not None and time.monotonic() - self._cache_time < self.cache_ttl:
|
|
return self._cached.model_copy(deep=True)
|
|
async with self._lock:
|
|
if self._cached is not None and time.monotonic() - self._cache_time < self.cache_ttl:
|
|
return self._cached.model_copy(deep=True)
|
|
try:
|
|
result = await self._fetch_usage()
|
|
if result.status == 'ok':
|
|
self.last_successful_update = result.last_successful_update or utc_now()
|
|
result.last_successful_update = self.last_successful_update
|
|
self._cached = result.model_copy(deep=True)
|
|
self._cache_time = time.monotonic()
|
|
return result
|
|
result.last_successful_update = self.last_successful_update
|
|
result.error = GENERIC_ERROR
|
|
return result
|
|
except Exception:
|
|
return ProviderUsage(provider=self.provider, status='error', error=GENERIC_ERROR,
|
|
last_successful_update=self.last_successful_update, source=self.source)
|
|
|
|
@abstractmethod
|
|
async def _fetch_usage(self) -> ProviderUsage:
|
|
raise NotImplementedError
|
|
|
|
async def aclose(self) -> None:
|
|
return None
|