From 73b5922c97592b907fd52e6abbf3d1da1b006f58 Mon Sep 17 00:00:00 2001 From: Noloquideus Date: Sun, 28 Jun 2026 14:12:23 +0300 Subject: [PATCH] fix: harden wallet balances and service startup --- src/infrastructure/config/settings.py | 2 + src/infrastructure/wallet_balances/service.py | 46 +++++++++++++++---- src/main.py | 11 +++-- src/presentation/dependencies/commands.py | 8 +++- 4 files changed, 55 insertions(+), 12 deletions(-) diff --git a/src/infrastructure/config/settings.py b/src/infrastructure/config/settings.py index 91322aa..ba5810d 100644 --- a/src/infrastructure/config/settings.py +++ b/src/infrastructure/config/settings.py @@ -65,6 +65,8 @@ class Settings(BaseSettings): CSRF_COOKIE_PATH: str = '/' CSRF_COOKIE_DOMAIN: str | None = None + CORS_ALLOW_ORIGIN_REGEX: str = r'https?://([a-z0-9-]+\.)*elcsa\.ru(:\d+)?$' + DOCS_USERNAME: str = 'admin' DOCS_PASSWORD: str = 'admin' diff --git a/src/infrastructure/wallet_balances/service.py b/src/infrastructure/wallet_balances/service.py index e8f4c23..63300f9 100644 --- a/src/infrastructure/wallet_balances/service.py +++ b/src/infrastructure/wallet_balances/service.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio import hashlib +import time from dataclasses import dataclass from decimal import Decimal, InvalidOperation from typing import Any @@ -19,6 +20,8 @@ from src.infrastructure.wallet_balances.token_registry import ( ) TIMEOUT_SECONDS = 15 +RETRY_ATTEMPTS = 2 +PRICE_CACHE_SECONDS = 30 BLOCKSTREAM_URL = 'https://blockstream.info/api' TRONGRID_URL = 'https://api.trongrid.io' COINGECKO_URL = 'https://api.coingecko.com/api/v3/simple/price' @@ -49,6 +52,10 @@ class WalletBalance: class WalletBalanceService: + def __init__(self) -> None: + self._prices_cache: dict[tuple[tuple[str,str],...],tuple[float,dict[str,Decimal | None]]] = {} + + async def get_wallet_balances(self, wallets) -> list[WalletBalance]: prices = await self._get_prices(wallets) results = await asyncio.gather( @@ -193,10 +200,17 @@ class WalletBalanceService: for token in self._tokens_for_chain(chain): coin_id = get_coingecko_id(chain, token['symbol']) if coin_id: - pairs[f'{chain}:{token['symbol']}'] = coin_id + symbol = token['symbol'] + pairs[f'{chain}:{symbol}'] = coin_id if not pairs: return {} + cache_key = tuple(sorted(pairs.items())) + cached = self._prices_cache.get(cache_key) + now = time.monotonic() + if cached is not None and now - cached[0] < PRICE_CACHE_SECONDS: + return dict(cached[1]) + try: response = await self._get_json( COINGECKO_URL, @@ -210,10 +224,12 @@ class WalletBalanceService: except Exception: return {key: None for key in pairs} - return { + prices = { key: self._decimal_or_none((response.get(coin_id) or {}).get('usd')) for key, coin_id in pairs.items() } + self._prices_cache[cache_key] = (now,prices) + return prices def _token_amounts( self, @@ -287,9 +303,16 @@ class WalletBalanceService: timeout: int = TIMEOUT_SECONDS, ) -> Any: async with httpx.AsyncClient(timeout=timeout) as client: - response = await client.get(url, params=params, headers=headers) - response.raise_for_status() - return response.json() + for attempt in range(RETRY_ATTEMPTS + 1): + try: + response = await client.get(url, params=params, headers=headers) + response.raise_for_status() + return response.json() + except (httpx.RequestError,httpx.HTTPStatusError,ValueError): + if attempt >= RETRY_ATTEMPTS: + raise + await asyncio.sleep(0.2 * (attempt + 1)) + raise RuntimeError('HTTP GET failed') async def _post_json( self, @@ -300,9 +323,16 @@ class WalletBalanceService: timeout: int = TIMEOUT_SECONDS, ) -> Any: async with httpx.AsyncClient(timeout=timeout) as client: - response = await client.post(url, json=payload, headers=headers) - response.raise_for_status() - return response.json() + for attempt in range(RETRY_ATTEMPTS + 1): + try: + response = await client.post(url, json=payload, headers=headers) + response.raise_for_status() + return response.json() + except (httpx.RequestError,httpx.HTTPStatusError,ValueError): + if attempt >= RETRY_ATTEMPTS: + raise + await asyncio.sleep(0.2 * (attempt + 1)) + raise RuntimeError('HTTP POST failed') def _tokens_for_chain(self, chain: str) -> list[dict]: if chain in {'ETH', 'BSC'}: diff --git a/src/main.py b/src/main.py index 4b172d4..65d6c58 100644 --- a/src/main.py +++ b/src/main.py @@ -74,8 +74,13 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: path=settings.VAULT_CRYPTO_MASTER_KEY_PATH, ) logger.info('Crypto master key loaded from Vault') - yield - logger.info(f'B2B service instance ended with id {instance_id}') + try: + yield + finally: + sched = getattr(app.state,'jwt_keys_scheduler',None) + if sched: + sched.shutdown(wait=False) + logger.info(f'B2B service instance ended with id {instance_id}') app: FastAPI = FastAPI( @@ -111,7 +116,7 @@ app.add_middleware( app.add_middleware( CORSMiddleware, allow_origins=[], - allow_origin_regex='.*', + allow_origin_regex=settings.CORS_ALLOW_ORIGIN_REGEX, allow_credentials=True, allow_methods=['*'], allow_headers=['*'], diff --git a/src/presentation/dependencies/commands.py b/src/presentation/dependencies/commands.py index 23d1af7..97ff51c 100644 --- a/src/presentation/dependencies/commands.py +++ b/src/presentation/dependencies/commands.py @@ -1,3 +1,4 @@ +from functools import lru_cache from fastapi import Depends from src.application.abstractions import IUnitOfWork from src.application.commands import ( @@ -54,13 +55,18 @@ def get_list_organization_wallets_command( return ListOrganizationWalletsCommand(unit_of_work=unit_of_work, logger=logger) +@lru_cache(maxsize=1) +def _wallet_balance_service() -> WalletBalanceService: + return WalletBalanceService() + + def get_get_organization_wallet_balances_command( logger: ILogger = Depends(get_logger), unit_of_work: IUnitOfWork = Depends(get_unit_of_work), ) -> GetOrganizationWalletBalancesCommand: return GetOrganizationWalletBalancesCommand( unit_of_work=unit_of_work, - balance_service=WalletBalanceService(), + balance_service=_wallet_balance_service(), logger=logger, )