tPjE 2UdZddlZddlmZmZddlmZddlmZddl m Z ddl m Z m Z e dZeezezdzZeeefZed d Gd d Ze ed deddeddeddeddeddd dZe eeefed<dZe ed<dZe ed <d!Ze ed"<e ieDcic]}|d#c}eDcic]}|d$c}Ze eeefed%<d&eed'efd(Zd)ed'efd*Zd)ed'efd+Zd)ed'efd,Zd-ed.ed'e ed/ffd0Z!d1eed'efd2Z"d-ed3eed'e e e ed/fefd/ffd4Z#d-ed5e ed/fd.ed'e ed/ffd6Z$d-ed'e ed/ffd7Z%d-ed8ee e ed/fefd'e ee ed/fffd9Z&ycc}wcc}w):aOne multi-row ``INSERT ... ON CONFLICT DO UPDATE`` per batch of daily spend rows. Emitting a statement per aggregated key put every replica's flush on the database as hundreds of separate statements against the same handful of hot rows, each holding its row locks for the rest of the enclosing batch transaction. Folding a batch into a single statement keeps the aggregation identical while collapsing both the statement count and the window in which those locks are held. N)MappingSequence) dataclass)groupby)MappingProxyType)FinalLiteral)userteamorgtagend_useragentT)frozenslotsc4eZdZUdZeed<eed<dZeed<y)DailySpendTablez4The physical table behind one entity's daily rollup.nameentity_id_columnFcarries_request_idN)__name__ __module__ __qualname____doc__str__annotations__rboolm/home/oem/CyberBrain26-unified/.venv/lib/python3.12/site-packages/litellm/proxy/db/daily_spend_bulk_upsert.pyrrs> I$$rrLiteLLM_DailyUserSpenduser_id)rrLiteLLM_DailyTeamSpendteam_idLiteLLM_DailyOrganizationSpendorganization_idLiteLLM_DailyEndUserSpend end_user_idLiteLLM_DailyAgentSpendagent_idLiteLLM_DailyTagSpendr )rrr)r r r rrr DAILY_SPEND_TABLES)dateapi_keymodelcustom_llm_providermcp_namespaced_tool_nameendpoint _KEY_COLUMNS) prompt_tokenscompletion_tokens api_requestssuccessful_requestsfailed_requestscache_read_input_tokenscache_creation_input_tokenscompression_saved_tokens_COUNTER_COLUMNS)spendcompression_savings_spendprompt_caching_savings_spend&gateway_injected_caching_savings_spendautorouter_savings_spend_SPEND_COLUMNSbigintzdouble precision_CASTScolumnsreturnc2djd|DS)N, c3(K|] }d|d yw)"Nr).0columns r z_quoted..Ns9vq]9s)join)rEs r _quotedrOMs 9999 99rvaluec |dSt|S)N)rrPs r _as_textrTQs2.CJ.rcHt|ttfr t|SdS)Nr isinstanceintfloatrSs r _as_intrZUs#EC<83u:?a?rcHt|ttfr t|SdS)NgrVrSs r _as_floatr\Ys%ec5\:5<CCrtable transaction.cPtfd|jgtDS)zJThe tuple the database arbitrates the upsert on, normalized free of NULLs.c3RK|]}tj| ywN)rTgetrKrLr^s r rMzconflict_key.._siv+//&12i$')tuplerr3)r]r^s `r conflict_keyrf]s# i%BXBXAh[gAhi iirgroupc t|dk(r|dSi|dtDcic]tfd|Dc}tDcic]tfd|Dc}Scc}wcc}w)Nrc3RK|]}t|j ywrarZrbrKrowrLs r rMz_merge..gsACwswwv/Ardc3RK|]}t|j ywrar\rbrls r rMz_merge..hsCcy1Crd)lenr<sumrB)rgrLs `r _mergerrbs} 5zQQx  ( P` af63A5AA A a Sa a63CUCC C a  a as A, A1 transactionscdt|fd}tdt|fdDS)aBatch entries keyed by the conflict tuple, in a deterministic order. The queue keys transactions by their raw field values, so two entries differing only in a NULL versus an empty member reach the writer separately while arbitrating to the same row. Postgres rejects a statement whose values touch one row twice, so they are summed here into the single row they were always destined to become. Ordering by the key keeps concurrent writers taking row locks in the same sequence. ct|Srarf)r^r]s r z'merge_by_conflict_key..xs,uVaBbr)keyc3NK|]\}}|tt|fywra)rrre)rKrxrgs r rMz(merge_by_conflict_key..ys"ve#veEl+,vs#%ct|Srarv)tr]s r rwz'merge_by_conflict_key..ys^jkprs^tr)sortedrer)r]rsordereds` r merge_by_conflict_keyr~ls+L.bcG vwwTt?uv vvrrxcDjd}ttjg|jddnt jdfdt Dfdt D|jr|dfSt |fSdS)N request_id model_groupc3RK|]}tj| ywrarkrcs r rMz_row_params..s Jv'+//&) * Jrdc3RK|]}tj| ywrarorcs r rMz_row_params..s J)KOOF+ , Jrdr)rbruuiduuid4rTr<rBr)r]rxr^rs ` r _row_paramsr|s $ 5J DJJL    .6H[__UbEc J   FKE]E]:%4 A   ,4J+? A   df  rcld|jgtdtt|jrdSdS)Nidr)rr)rr3r<rBr)r]s r _insert_columnsrsa              #55/ <> rbatchc tdjddjfdtt |D}djfdgt t D}jrddnd}dd td |d tjgtd ||d }|tfd|DfS)zMThe single statement writing one merged batch, plus its positional arguments.rJrHc3tK|].ddjfdtDzdz0yw)(rHc3K|]5\}}dtz|zdzdtj|d7yw)$riz::textN)rprDrb)rKoffsetrLrE row_indexs r rMz.build_bulk_upsert...sJ  CL(61A56bFF9S8T U s;>z, (NOW() AT TIME ZONE 'UTC'))N)rN enumerate)rKrrEs @r rMz$build_bulk_upsert..sH  )) "+G"4   *  *s48c 3<K|]}d|dd|d|d yw)rJz" = z."z" + EXCLUDED."Nr)rKrL quoted_tables r rMz$build_bulk_upsert..s4"  F84 ~Rx~fXQG"sz1, "request_id" = COALESCE(EXCLUDED."request_id", z."request_id")rRz INSERT INTO z (z, "updated_at") VALUES z ON CONFLICT (z) DO UPDATE SET z-, "updated_at" = (NOW() AT TIME ZONE 'UTC')c3LK|]\}}t||D]}|ywra)r)rKrxr^rPr]s r rMz$build_bulk_upsert..s.l 0[{SXZ]_jGkleelels!$) rrrNrangerpr<rBrrOrr3re)r]rrows incrementsrequest_id_updatesqlrErs` @@r build_bulk_upsertrs %U+Gejj\+L))s5z*D ":(:>:"J  # # X  |nBww'7&89!7!7 G, GHIJ L*+,8 9 lEll llr)'rrcollections.abcrr dataclassesr itertoolsrtypesrtypingrr DailySpendEntityrrXrYSqlValueobjectSpendRowrr,rr3r<rBrDrOrTrZr\rerfrrr~rrr)rLs0r rs -!"!LM 9u t # 3;  $d#%%$%IY%=PYZ%=PYZ$DWhi#)DWde &?R\]$;ehlm  IE'"2O"CDE r eq % $4*: ;68  ;4B C&6% % C$gc3h :Xc]:s:/F/s/@6@c@DVDDjjhj5c?j (8$ w  w8$ w 5sCx(* +S 01 w    sCx   8S=  ? uS#X "m "m E%S/834 5"m 3hm$ $%"mc < Cs  F  F