Coverage for src/kimi_code_usage/providers/anthropic.py: 100%

1import aiohttp 

2from typing import List, Optional 

3from . import ProviderUsage 

4 

5async def fetch_anthropic_usage(api_key: str, base_url: str, management_key: Optional[str] = None) -> List[ProviderUsage]: 

6 if api_key.startswith("sk-ant-"): 

7 return [ 

8 ProviderUsage( 

9 provider="anthropic", 

10 label="API Plan", 

11 used=0.0, 

12 limit=None, 

13 remaining=None, 

14 percent=None, 

15 reset_at=None, 

16 unit="text" 

17 ) 

18 ] 

19 

20 url = f"{base_url.rstrip('/')}/api/oauth/usage" 

21 headers = {"Authorization": f"Bearer {api_key}"} 

22 

23 async with aiohttp.ClientSession() as session: 

24 async with session.get(url, headers=headers) as resp: 

25 if resp.status in (401, 403): 

26 return [ 

27 ProviderUsage( 

28 provider="anthropic", 

29 label="API Plan (No usage API)", 

30 used=0.0, 

31 limit=None, 

32 remaining=None, 

33 percent=None, 

34 reset_at=None, 

35 unit="text" 

36 ) 

37 ] 

38 if resp.status != 200: 

39 text = await resp.text() 

40 raise Exception(f"Anthropic API Error {resp.status}: {text}") 

41 data = await resp.json() 

42 

43 five_hour = data.get("five_hour", {}) 

44 seven_day = data.get("seven_day", {}) 

45 five_hour_pct = float(five_hour.get("utilization", 0.0)) * 100.0 if isinstance(five_hour, dict) else 0.0 

46 seven_day_pct = float(seven_day.get("utilization", 0.0)) * 100.0 if isinstance(seven_day, dict) else 0.0 

47 

48 return [ 

49 ProviderUsage( 

50 provider="anthropic", 

51 label="5 Hours", 

52 used=five_hour_pct, 

53 limit=100.0, 

54 remaining=100.0 - five_hour_pct, 

55 percent=five_hour_pct, 

56 reset_at=None, 

57 unit="%" 

58 ), 

59 ProviderUsage( 

60 provider="anthropic", 

61 label="7 Days", 

62 used=seven_day_pct, 

63 limit=100.0, 

64 remaining=100.0 - seven_day_pct, 

65 percent=seven_day_pct, 

66 reset_at=None, 

67 unit="%" 

68 ) 

69 ]