$ja$ VUdZddlZddlZddlZddlmZddlmZmZddl m Z m Z ej e ZeGddZGdd ZGd d Zdaedzed <ddedefdZe ddedzdedzdede efdZGddZdS)z Database connection budget management. Limits concurrent database connections per operation to prevent a single operation (e.g., recall with parallel queries) from exhausting the connection pool. N)asynccontextmanager) dataclassfield)Any AsyncIteratorc~eZdZUdZeed<eed<edZe j ed<eddZ eed <d Z d S) OperationBudgetz Tracks connection budget for a single operation. Each operation gets a semaphore limiting its concurrent connections. operation_idmax_connectionsF)init semaphorer)defaultr active_countcBtj|j|_dSN)asyncio Semaphorer r selfs f/home/rurouni/.hermes/hermes-agent/venv/lib/python3.11/site-packages/hindsight_api/engine/db_budget.py __post_init__zOperationBudget.__post_init__ s *4+?@@N) __name__ __module__ __qualname____doc__str__annotations__intrr rrrrrrr r s #(5e#4#4#4Iw 444ae444L#444AAAAArr c reZdZdZd defdZe d dedzdedzded fd Z dede fd Z dS)ConnectionBudgetManageraA Manages per-operation connection budgets. Usage: manager = ConnectionBudgetManager(default_budget=4) # Start an operation async with manager.operation(max_connections=2) as op: # Acquire connections within the budget async with op.acquire(pool) as conn: await conn.fetch(...) # Multiple connections respect the budget async with op.acquire(pool) as conn1, op.acquire(pool) as conn2: # At most 2 concurrent connections for this operation ... default_budgetcR||_i|_tj|_dS)z Initialize the budget manager. Args: default_budget: Default max connections per operation N)r$ _operationsrLock_lock)rr$s r__init__z ConnectionBudgetManager.__init__7s%-79\^^ rNr r returnBudgetedOperationcK|p"dtjjdd}|p|j}|j4d{V||jvrt d|dt|||j|<dddd{Vn#1d{VswxYwY t||WV|j4d{V|j |ddddd{VdS#1d{VswxYwYdS#|j4d{V|j |ddddd{Vw#1d{VswxYwYwxYw)a{ Create a budgeted operation context. Args: max_connections: Max concurrent connections for this operation. Defaults to manager's default_budget. operation_id: Optional custom operation ID. Auto-generated if not provided. Yields: BudgetedOperation context for acquiring connections zop-N Operation z already exists) uuiduuid4hexr$r(r& ValueErrorr r+pop)rr r op_idbudgets r operationz!ConnectionBudgetManager.operationBs("= =djll&6ss&; = = 7D$7: E E E E E E E E((( !De!D!D!DEEE&5eV&D&DD U # E E E E E E E E E E E E E E E E E E E E E E E E E E E  2#D%00 0 0 0 0z 2 2 2 2 2 2 2 2 $$UD111 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2tz 2 2 2 2 2 2 2 2 $$UD111 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2sS5B BBC97C&& C03C09ED5#E5 D? ?ED? Ecd|j|}|std|d|S)z+Get budget for an operation (internal use).r.z not found)r&getr2)rr r5s r _get_budgetz#ConnectionBudgetManager._get_budgetas@!%%l33 DB,BBBCC C rr#)NN) rrrrrr)rrrr6r r9r rrr"r"$s$ $ $s $ $ $ $'+#'22t2Dj2 * + 2222<rr"ceZdZdZdedefdZedefdZ e de de e fdZ de dd fd Zde d ede ee fd Zd S)r+zt A single operation with connection budget. Provides methods to acquire connections within the budget. managerr c"||_||_dSr)_managerr )rr<r s rr)zBudgetedOperation.__init__ps (rr*c@|j|jS)z"Get the budget for this operation.)r>r9r rs rr5zBudgetedOperation.budgetts}(():;;;rpoolcXK|j}|j4d{V|xjdz c_ ddlm}t ||rD|4d{V}|WVdddd{Vn#1d{VswxYwYn\|d{V} |WV||d{Vn #||d{VwxYw|xjdzc_n#|xjdzc_wxYw dddd{VdS#1d{VswxYwYdS)a Acquire a connection within the operation's budget. Blocks if the operation has reached its connection limit. Args: pool: asyncpg connection pool or DatabaseBackend Yields: Database connection NDatabaseBackend)r5r rdb.baserD isinstanceacquirerelease)rr@r5rDconns rrGzBudgetedOperation.acquireys# ) ) ) ) ) ) ) )   1 $   )444444dO441#||~~#######" ###########################"&//////D1" "ll40000000000dll40000000000##q(#####q(####### ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) )sdD0C0A1C01 A; ;C0>A; ?C0B?#C0?CC0D0DD D#&D# BudgetedPoolc"t||S)av Wrap a pool with this operation's budget. The returned BudgetedPool can be passed to functions expecting a pool, and all acquire() calls will be limited by this operation's budget. Args: pool: asyncpg connection pool to wrap Returns: BudgetedPool that limits connections to this operation's budget )rJ)rr@s r wrap_poolzBudgetedOperation.wrap_poolsD$'''rcountc&Kg} t|D]1}|d{V}||2|WV|D]}||d{VdS#|D]}||d{VwxYw)a Acquire multiple connections within the budget. Note: This acquires connections sequentially to respect the budget. For parallel acquisition, use multiple acquire() calls with asyncio.gather(). This method is intended for use with raw asyncpg pools only, not DatabaseBackend. Args: pool: asyncpg connection pool (raw pool only) count: Number of connections to acquire Yields: List of database connections N)rangerGappendrH)rr@rM connections_rIs r acquire_manyzBudgetedOperation.acquire_manys&  )5\\ ) )!\\^^++++++""4((((    # ) )ll4(((((((((( ) )  ) )ll4(((((((((( )s AA.."BN)rrrrr"rr)propertyr r5rrrrGrLrlistrSr rrr+r+is ) 7)s))))<<<<X<)#)-*<))))< (c (n ( ( ( ())) tCy ! ))))))rr+_default_managerr#r$r*c>tt|atS)z Get or create the global budget manager. Args: default_budget: Default max connections per operation Returns: Global ConnectionBudgetManager instance Nr$)rVr"rXs rget_budget_managerrYs!2.QQQ rr r cKt|}|||4d{V}|WVdddd{VdS#1d{VswxYwYdS)a Convenience function to create a budgeted operation. Args: max_connections: Max concurrent connections for this operation operation_id: Optional custom operation ID default_budget: Default budget if manager not yet created Yields: BudgetedOperation context Example: async with budgeted_operation(max_connections=2) as op: async with op.acquire(pool) as conn: await conn.fetch(...) N)rYr6)r r r$r<ops rbudgeted_operationr\s,!00G  ,??2sA AAcdeZdZdZdZdedefdZede efdZ dedd fd Z d Z d S) rJa A pool wrapper that limits concurrent connection acquisitions. This can be passed to functions expecting a pool, and acquire() calls will be limited by the budget semaphore. Usage: async with budgeted_operation(max_connections=4) as op: budgeted_pool = op.wrap_pool(pool) # Pass budgeted_pool to functions that expect a pool await some_function(budgeted_pool, ...) Tr@r6c"||_||_dSr)_pool _operation)rr@r6s rr)zBudgetedPool.__init__s #rr*cK|jj}|jd{V|xjdz c_ ddlm}t|j|rI|j4d{V}|WVdddd{Vn#1d{VswxYwYnk|jd{V} |WV|j |d{Vn%#|j |d{VwxYwn#t$rwxYw|xjdzc_|j dS#|xjdzc_|j wxYw)z Acquire a connection within the budget as an async context manager. The connection is automatically released when the context exits. NrBrC) r`r5r rGrrErDrFr_rH Exception)rr5rDrIs rrGzBudgetedPool.acquiresV'&&(((((((((q  ' 0 0 0 0 0 0$*o66 3:--//4JJJJ"Z//111111113JJJJ*,,T2222222222$*,,T2222222222        1 $     $ $ & & & & &   1 $     $ $ & & & &sY:D9B?D BDB#DC)!D)"D  DE  DE +E4rINcK|jj} |j|d{V|xjdzc_|jdS#|xjdzc_|jwxYw)z9Release a connection back to the pool (legacy path only).NrB)r`r5r_rHrr )rrIr5s rrHzBudgetedPool.release$s' '*$$T** * * * * * * *   1 $     $ $ & & & & &   1 $     $ $ & & & &s A+Bc,t|j|S)z.Proxy other attributes to the underlying pool.)getattrr_)rnames r __getattr__zBudgetedPool.__getattr__-stz4(((r) rrrr_wraps_backendrr+r)rrrGrHrgr rrrJrJs  N$S$->$$$$'}S1''''6'#'$'''')))))rrJr:)NNr#)rrloggingr/ contextlibr dataclassesrrtypingrr getLoggerrloggerr r"r+rVrrrYrr\rJr rrros ******((((((((%%%%%%%%  8 $ $  A A A A A A A  A BBBBBBBBJY)Y)Y)Y)Y)Y)Y)Y)z48)D0777  s 3J     "&#4Z*$% 4;);););););););););)r