« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Usage and rankings read service over persisted relay request logs.
3Aggregates the ``ai_relay_request_logs`` table into per-user daily usage
4and per-model rankings using only generic SQL, so the service works on
5any backend. Cost is summed as a decimal cast at the database and
6normalised to a trimmed decimal string on the way out.
7"""
9from __future__ import annotations
11from datetime import datetime, timedelta
12from decimal import Decimal
13from typing import TYPE_CHECKING
15from lexigram.contracts.ai.relay import (
16 RelayDailyUsage,
17 RelayModelRank,
18 RelayRequestLogEntry,
19 RelayUsageServiceProtocol,
20)
21from lexigram.primitives import clock
23if TYPE_CHECKING:
24 from lexigram.contracts.data import DatabaseProviderProtocol
26__all__ = ["RelayUsageService"]
28_DAILY_USAGE = """
29SELECT date(created_at) AS day,
30 sum(prompt_tokens) AS prompt_tokens,
31 sum(completion_tokens) AS completion_tokens,
32 sum(cast(cost AS decimal)) AS cost
33FROM ai_relay_request_logs
34WHERE user_id = ? AND date(created_at) >= ?
35GROUP BY date(created_at)
36ORDER BY day
37"""
39_MODEL_RANK = """
40SELECT model,
41 sum(completion_tokens) AS completion_tokens,
42 count(*) AS request_count,
43 sum(cast(cost AS decimal)) AS cost
44FROM ai_relay_request_logs
45WHERE date(created_at) >= ?
46GROUP BY model
47ORDER BY completion_tokens DESC
48LIMIT ?
49"""
51_LIST_REQUESTS = """
52SELECT request_id, user_id, token_id, endpoint_kind, model, channel_name,
53 status, created_at, prompt_tokens, completion_tokens, cost,
54 latency_ms, error_code
55FROM ai_relay_request_logs
56WHERE date(created_at) >= ?
57 AND (? IS NULL OR user_id = ?)
58 AND (? IS NULL OR token_id = ?)
59ORDER BY created_at DESC, request_id DESC
60LIMIT ? OFFSET ?
61"""
64def _cost_text(value: object) -> str:
65 """Trim a summed numeric cost to a short decimal string."""
66 magnitude = round(float(str(value or "0")), 6)
67 return str(Decimal(str(magnitude)))
70class RelayUsageService(RelayUsageServiceProtocol):
71 """Aggregate reads over ``ai_relay_request_logs``.
73 Args:
74 db: A connected
75 :class:`~lexigram.contracts.data.DatabaseProviderProtocol`
76 resolved from the DI container.
77 """
79 def __init__(self, db: DatabaseProviderProtocol) -> None:
80 self._db = db
82 async def daily_usage(self, user_id: str, days: int) -> list[RelayDailyUsage]:
83 """Aggregate tokens and cost per day for *user_id*.
85 Args:
86 user_id: The user to aggregate.
87 days: Include entries from this many days back (inclusive).
89 Returns:
90 Daily aggregates ordered by day ascending.
91 """
92 cutoff = (clock.now().date() - timedelta(days=days - 1)).isoformat()
93 result = await self._db.execute_query(_DAILY_USAGE, [user_id, cutoff])
94 return [
95 RelayDailyUsage(
96 day=row["day"],
97 prompt_tokens=int(row["prompt_tokens"] or 0),
98 completion_tokens=int(row["completion_tokens"] or 0),
99 cost=_cost_text(row["cost"]),
100 )
101 for row in result.rows
102 ]
104 async def model_rank(self, days: int, limit: int) -> list[RelayModelRank]:
105 """Rank models by completion tokens over the window.
107 Args:
108 days: Include entries from this many days back (inclusive).
109 limit: Maximum number of ranked models to return.
111 Returns:
112 Models ordered by completion tokens descending.
113 """
114 cutoff = (clock.now().date() - timedelta(days=days - 1)).isoformat()
115 result = await self._db.execute_query(_MODEL_RANK, [cutoff, limit])
116 return [
117 RelayModelRank(
118 model=row["model"],
119 completion_tokens=int(row["completion_tokens"] or 0),
120 request_count=int(row["request_count"] or 0),
121 cost=_cost_text(row["cost"]),
122 )
123 for row in result.rows
124 ]
126 async def list_requests(
127 self,
128 days: int,
129 page: int,
130 page_size: int,
131 *,
132 user_id: str | None = None,
133 token_id: str | None = None,
134 ) -> list[RelayRequestLogEntry]:
135 """List recent request-log entries, newest first.
137 Args:
138 days: Include entries from this many days back (inclusive).
139 page: One-based result page number.
140 page_size: Maximum entries per page.
141 user_id: Optional user filter; ``None`` matches all users.
142 token_id: Optional token filter; ``None`` matches all tokens.
144 Returns:
145 The matching entries ordered by time descending.
146 """
147 cutoff = (clock.now().date() - timedelta(days=days - 1)).isoformat()
148 result = await self._db.execute_query(
149 _LIST_REQUESTS,
150 [
151 cutoff,
152 user_id,
153 user_id,
154 token_id,
155 token_id,
156 page_size,
157 (page - 1) * page_size,
158 ],
159 )
160 return [
161 RelayRequestLogEntry(
162 request_id=row["request_id"],
163 user_id=row["user_id"],
164 token_id=row["token_id"],
165 endpoint_kind=row["endpoint_kind"],
166 model=row["model"],
167 channel_name=row["channel_name"],
168 status=row["status"],
169 created_at=datetime.fromisoformat(row["created_at"]),
170 prompt_tokens=int(row["prompt_tokens"] or 0),
171 completion_tokens=int(row["completion_tokens"] or 0),
172 cost=row["cost"],
173 latency_ms=int(row["latency_ms"] or 0),
174 error_code=row["error_code"],
175 )
176 for row in result.rows
177 ]