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

1 

2import aiohttp 

3 

4from . import ProviderUsage 

5 

6 

7async def fetch_anthropic_usage(api_key: str, base_url: str, management_key: str | None = None) -> list[ProviderUsage]: 

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

9 return [ 

10 ProviderUsage( 

11 provider="anthropic", 

12 label="API Plan", 

13 used=0.0, 

14 limit=None, 

15 remaining=None, 

16 percent=None, 

17 reset_at=None, 

18 unit="text" 

19 ) 

20 ] 

21 

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

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

24 

25 async with aiohttp.ClientSession() as session, session.get(url, headers=headers) as resp: 

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

27 return [ 

28 ProviderUsage( 

29 provider="anthropic", 

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

31 used=0.0, 

32 limit=None, 

33 remaining=None, 

34 percent=None, 

35 reset_at=None, 

36 unit="text" 

37 ) 

38 ] 

39 if resp.status != 200: 

40 text = await resp.text() 

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

42 data = await resp.json() 

43 

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

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

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

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

48 

49 return [ 

50 ProviderUsage( 

51 provider="anthropic", 

52 label="5 Hours", 

53 used=five_hour_pct, 

54 limit=100.0, 

55 remaining=100.0 - five_hour_pct, 

56 percent=five_hour_pct, 

57 reset_at=None, 

58 unit="%" 

59 ), 

60 ProviderUsage( 

61 provider="anthropic", 

62 label="7 Days", 

63 used=seven_day_pct, 

64 limit=100.0, 

65 remaining=100.0 - seven_day_pct, 

66 percent=seven_day_pct, 

67 reset_at=None, 

68 unit="%" 

69 ) 

70 ]