import math import time from datetime import datetime from typing import Any from urllib.parse import quote from zoneinfo import ZoneInfo import httpx from pydantic import BaseModel, ConfigDict, Field, StrictBool, StrictInt, StrictStr, ValidationError from app.config import Settings from app.models import ProviderUsage, UsageLimit from app.providers.base import GENERIC_ERROR, UsageProvider, utc_now class Cycle(BaseModel): model_config = ConfigDict(extra='ignore') after: StrictInt before: StrictInt class CyclesResponse(BaseModel): model_config = ConfigDict(extra='ignore') items: list[Cycle] has_next_page: StrictBool = False end_cursor: StrictStr | None = None class DailyResponse(BaseModel): model_config = ConfigDict(extra='ignore') total_acus: Any consumption_by_date: list[Any] = Field(default_factory=list) class OrgResponse(BaseModel): model_config = ConfigDict(extra='ignore') max_cycle_acu_limit: Any = None class SessionSummary(BaseModel): model_config = ConfigDict(extra='ignore') session_id: StrictStr created_at: StrictInt acus_consumed: Any class SessionsResponse(BaseModel): model_config = ConfigDict(extra='ignore') items: list[SessionSummary] has_next_page: StrictBool = False end_cursor: StrictStr | None = None def _acu(value: Any) -> float: if isinstance(value, bool) or not isinstance(value, (int, float)): raise ValueError number = float(value) if not math.isfinite(number) or number < 0: raise ValueError return number class DevinProvider(UsageProvider): def __init__(self, settings: Settings, client: httpx.AsyncClient | None = None): source = 'Devin Organization Sessions API' if settings.devin_mode == 'organization_sessions' else 'Devin API v3' super().__init__('devin', settings.provider_cache_ttl, source) self.settings = settings self._client = client or httpx.AsyncClient(base_url=settings.devin_api_url.rstrip('/'), timeout=settings.devin_timeout, headers={'Authorization': f'Bearer {settings.devin_api_key.get_secret_value()}'}) self._owns_client = client is None async def _fetch_usage(self) -> ProviderUsage: if self.settings.devin_mode == 'organization_sessions': return await self._fetch_organization_sessions() return await self._fetch_enterprise_usage() async def _fetch_enterprise_usage(self) -> ProviderUsage: now = int(time.time()) params = {'first': 100} seen_cursors: set[str] = set() cycle = None for page in range(100): cycles_response = await self._client.get('/v3/enterprise/consumption/cycles', params=params) cycles_response.raise_for_status() try: cycles_page = CyclesResponse.model_validate(cycles_response.json()) except (ValidationError, ValueError, TypeError): raise ValueError cycle = next((item for item in cycles_page.items if item.after <= now < item.before), None) if cycle is not None: break if not cycles_page.has_next_page: break cursor = cycles_page.end_cursor if not cursor or cursor in seen_cursors: raise ValueError seen_cursors.add(cursor) params = {'first': 100, 'after': cursor} else: raise ValueError if cycle is None: raise ValueError user_id = self.settings.devin_user_id.strip() org_id = self.settings.devin_org_id.strip() if user_id: path = f'/v3/enterprise/consumption/daily/users/{quote(user_id, safe="")}' org_scope = False elif org_id: path = f'/v3/enterprise/consumption/daily/organizations/{quote(org_id, safe="")}' org_scope = True else: path = '/v3/enterprise/consumption/daily' org_scope = False daily_response = await self._client.get(path, params={'time_after': cycle.after, 'time_before': min(now, cycle.before)}) daily_response.raise_for_status() try: daily = DailyResponse.model_validate(daily_response.json()) used = _acu(daily.total_acus) except (ValidationError, ValueError, TypeError): raise ValueError limit = remaining = used_percent = remaining_percent = None if org_scope: org_path = f'/v3/enterprise/organizations/{quote(org_id, safe="")}' try: org_response = await self._client.get(org_path) if org_response.is_success: org_data = OrgResponse.model_validate(org_response.json()) if org_data.max_cycle_acu_limit is not None: limit = _acu(org_data.max_cycle_acu_limit) if limit > 0: remaining = max(limit - used, 0) used_percent = min(used / limit * 100, 100) remaining_percent = max(100 - used_percent, 0) else: remaining = 0 used_percent = 100 if used else 0 remaining_percent = 0 if used else 100 except (httpx.HTTPError, ValidationError, ValueError, TypeError): pass updated = utc_now() return ProviderUsage(provider='devin', status='ok', limits=[UsageLimit( id='cycle-acu-usage', name='Cycle ACU Usage', used=used, remaining=remaining, limit=limit, used_percent=used_percent, remaining_percent=remaining_percent, unit='ACUs', reset_at=datetime.fromtimestamp(cycle.before, tz=ZoneInfo(self.settings.timezone)), window='billing cycle', updated_at=updated)], last_successful_update=updated, source=self.source) async def _fetch_organization_sessions(self) -> ProviderUsage: org_id = self.settings.devin_org_id.strip() if not org_id: raise ValueError user_id = self.settings.devin_user_id.strip() now = int(time.time()) days = self.settings.devin_session_window_days params: dict[str, Any] = { 'first': 200, 'created_after': now - days * 86400, 'created_before': now, } if user_id: params['user_ids'] = [user_id] seen_cursors: set[str] = set() sessions: list[SessionSummary] = [] for page in range(100): response = await self._client.get( f'/v3/organizations/{quote(org_id, safe="")}/sessions', params=params) response.raise_for_status() try: sessions_page = SessionsResponse.model_validate(response.json()) except (ValidationError, ValueError, TypeError): raise ValueError sessions.extend(sessions_page.items) if not sessions_page.has_next_page: break cursor = sessions_page.end_cursor if not cursor or cursor in seen_cursors: raise ValueError seen_cursors.add(cursor) params = {'first': 200, 'after': cursor, 'created_after': now - days * 86400, 'created_before': now} if user_id: params['user_ids'] = [user_id] else: raise ValueError total_acus = sum(_acu(session.acus_consumed) for session in sessions) updated = utc_now() window = f'created in the last {days} days' return ProviderUsage(provider='devin', status='ok', limits=[ UsageLimit(id='organization-session-acus', name=f'{days}-Day Session Usage', used=total_acus, unit='ACUs', window=f'sessions {window}', updated_at=updated), UsageLimit(id='organization-session-count', name='Sessions Included', used=float(len(sessions)), unit='sessions', window=window, updated_at=updated), ], last_successful_update=updated, source=self.source) async def aclose(self) -> None: if self._owns_client: await self._client.aclose()