tPj%UdZddlmZddlmZmZddlmZmZddlm Z ddl m Z ddl m Z mZmZerddlmZe dd Zeed <d efd ZGd dZddded dfdZddded dfdZy)a Accumulates gateway request counts (SGR) recorded at the ASGI edge and commits them to ``LiteLLM_DailyGatewayRequests``. Unlike the spend queues this keeps no per-request item. A count is a pure aggregate, so requests fold into an in-memory map as they finish. Every dimension of the key is server-chosen and drawn from a fixed set: the date, the category, and a route that the classifier maps to one of a closed list of strings rather than passing the raw path through. Nothing a caller sends can add a key, so the fold and the table it commits to are bounded by (days x routes) however much traffic arrives, and the response path carries no unbounded queue that would block once full. )asdict)datetimetimezone) TYPE_CHECKINGFinal)verbose_proxy_logger)BillableCategory)GatewayRequestCountsGatewayRequestKeyGatewayRequestSnapshot) PrismaClientsuccessful_requestsfailed_requests_EMPTYreturncftjtjj dS)Nz%Y-%m-%d)rnowrutcstrftimen/home/oem/CyberBrain26-unified/.venv/lib/python3.12/site-packages/litellm/proxy/db/gateway_request_tracking.py _utc_dater!s << % . .z ::rcLeZdZdZd dZdedededdfdZde fd Z d e ddfd Z y) GatewayRequestAccumulatorzMSink for the request-metrics middleware. ``record`` is sync and never awaits.rNci|_yN_counts)selfs r__init__z"GatewayRequestAccumulator.__init__(s FH rcategoryroute status_codectt|j|}|jj |t j d|cxkxrdknc|j|<y)Ndater#r$i,) succeeded)r rvaluer getrplus)r!r#r$r%keys rrecordz GatewayRequestAccumulator.record+sP&IK(..X]^ LL,,S&9>>P[Ia^aIa>b Src,|j}i|_|Srr)r!draineds rdrainzGatewayRequestAccumulator.drain/s rsnapshotc|jD]m\}}|jj|t}t |j |j z|j |j z|j|<oy)a Merge un-committed counts back so the next flush retries them. A dropped flush would silently undercount the metric the dashboard now treats as the source of truth. Merging cannot grow without bound: keys collapse on collision, so the fold stays bounded by (date x category x route) however long the database is unreachable. This buys at-least-once, not exactly-once, and the cost is worth stating. The batch commits inside its context manager's ``__aexit__``, so a failure raised after the transaction committed (a connection dropped while reading the acknowledgement) restores counts that are already persisted, and the next flush increments them a second time. Exactly-once would need a dedup key the upserts could ignore on replay. For a traffic-volume metric a rare overcount on a dropped acknowledgement beats losing a whole interval to every database blip, so the trade is deliberate. rN)itemsr r,rr rr)r!r3r.countsexistings rrestorez!GatewayRequestAccumulator.restore4so$$>>+ KC||''V4H 4$,$@$@6C]C]$] ( 8 86;Q;Q Q!DLL  r)rN) __name__ __module__ __qualname____doc__r"r strintr/r r2r8rrrrr%sTWIc"2c3cScUYc-  64rr prisma_clientr r3Nc K|syt|jd}|jj4d{}|D]h\}}t |}|j j d|ii||j|jdd|jid|jiddjdddd{tjd t|y77&#1d{7swY6xYww) z8Upsert one incrementing row per (date, category, route).NcZ|dj|dj|djfS)Nrr')items rz/commit_gateway_requests_to_db..Ws-Q dSTgN^N^`def`g`m`m?nr)r.date_category_router increment)createupdate)wheredataz7Gateway request tracking - committed %d aggregated rows) sortedr5dbbatch_rlitellm_dailygatewayrequestsupsertrrrdebuglen)r?r3orderedbatcherr.r6columnss rcommit_gateway_requests_to_dbrTNs  HNN,2noG &&(G" KCSkG  0 0 7 7,g6!/5/I/I+1+A+A 1C;C"C;A.C&2 C;=C$>%C;$C;&C8,C/ -C84C; accumulatorcK|j} t||d{y7#t$r5|j|t j dt |dYywxYww)a Scheduler entrypoint. Never raises: a metering failure must not kill the job. ``CancelledError`` is deliberately not caught, so a flush cancelled during shutdown drops its snapshot rather than restoring counts onto an accumulator the process is about to discard. )r?r3NzOGateway request tracking - failed to commit %d rows, retrying on the next flushT)exc_info)r2rT Exceptionr8rwarningrP)r?rUr3s rflush_gateway_requestsrZqsc"'')H +-RZ[[[  H%$$ ] M  s1A-,*,A-,;A*'A-)A**A-)r< dataclassesrrrtypingrrlitellm._loggingrrbs ''1Y 0$ANN;3;&&R h! h% h hF ! *   r