Refactor provider filtering to use WrapperHandler helper function
Co-authored-by: Berack96 <31776951+Berack96@users.noreply.github.com>
This commit is contained in:
@@ -32,8 +32,8 @@ models:
|
||||
api:
|
||||
retry_attempts: 3
|
||||
retry_delay_seconds: 2
|
||||
market_providers: [BinanceWrapper, YFinanceWrapper, CoinBaseWrapper, CryptoCompareWrapper]
|
||||
news_providers: [GoogleNewsWrapper, DuckDuckGoWrapper, NewsApiWrapper, CryptoPanicWrapper]
|
||||
market_providers: [YFinanceWrapper, BinanceWrapper, CoinBaseWrapper, CryptoCompareWrapper]
|
||||
news_providers: [DuckDuckGoWrapper, GoogleNewsWrapper, NewsApiWrapper, CryptoPanicWrapper]
|
||||
social_providers: [RedditWrapper, XWrapper, ChanWrapper]
|
||||
|
||||
agents:
|
||||
|
||||
@@ -27,15 +27,12 @@ class MarketAPIsTool(MarketWrapper, Toolkit):
|
||||
"""
|
||||
config = AppConfig()
|
||||
|
||||
# Get wrapper classes based on configuration
|
||||
wrappers: list[type[MarketWrapper]] = []
|
||||
for provider_name in config.api.market_providers:
|
||||
if provider_name in self._WRAPPER_MAP:
|
||||
wrappers.append(self._WRAPPER_MAP[provider_name])
|
||||
|
||||
# Fallback to all wrappers if none configured
|
||||
if not wrappers:
|
||||
wrappers = [BinanceWrapper, YFinanceWrapper, CoinBaseWrapper, CryptoCompareWrapper]
|
||||
# Get wrapper classes based on configuration using the helper function
|
||||
wrappers = WrapperHandler.filter_wrappers_by_config(
|
||||
wrapper_map=self._WRAPPER_MAP,
|
||||
provider_names=config.api.market_providers,
|
||||
fallback_wrappers=[BinanceWrapper, YFinanceWrapper, CoinBaseWrapper, CryptoCompareWrapper]
|
||||
)
|
||||
|
||||
self.handler = WrapperHandler.build_wrappers(
|
||||
wrappers,
|
||||
|
||||
@@ -30,15 +30,12 @@ class NewsAPIsTool(NewsWrapper, Toolkit):
|
||||
"""
|
||||
config = AppConfig()
|
||||
|
||||
# Get wrapper classes based on configuration
|
||||
wrappers: list[type[NewsWrapper]] = []
|
||||
for provider_name in config.api.news_providers:
|
||||
if provider_name in self._WRAPPER_MAP:
|
||||
wrappers.append(self._WRAPPER_MAP[provider_name])
|
||||
|
||||
# Fallback to all wrappers if none configured
|
||||
if not wrappers:
|
||||
wrappers = [GoogleNewsWrapper, DuckDuckGoWrapper, NewsApiWrapper, CryptoPanicWrapper]
|
||||
# Get wrapper classes based on configuration using the helper function
|
||||
wrappers = WrapperHandler.filter_wrappers_by_config(
|
||||
wrapper_map=self._WRAPPER_MAP,
|
||||
provider_names=config.api.news_providers,
|
||||
fallback_wrappers=[GoogleNewsWrapper, DuckDuckGoWrapper, NewsApiWrapper, CryptoPanicWrapper]
|
||||
)
|
||||
|
||||
self.handler = WrapperHandler.build_wrappers(
|
||||
wrappers,
|
||||
|
||||
@@ -30,15 +30,12 @@ class SocialAPIsTool(SocialWrapper, Toolkit):
|
||||
"""
|
||||
config = AppConfig()
|
||||
|
||||
# Get wrapper classes based on configuration
|
||||
wrappers: list[type[SocialWrapper]] = []
|
||||
for provider_name in config.api.social_providers:
|
||||
if provider_name in self._WRAPPER_MAP:
|
||||
wrappers.append(self._WRAPPER_MAP[provider_name])
|
||||
|
||||
# Fallback to all wrappers if none configured
|
||||
if not wrappers:
|
||||
wrappers = [RedditWrapper, XWrapper, ChanWrapper]
|
||||
# Get wrapper classes based on configuration using the helper function
|
||||
wrappers = WrapperHandler.filter_wrappers_by_config(
|
||||
wrapper_map=self._WRAPPER_MAP,
|
||||
provider_names=config.api.social_providers,
|
||||
fallback_wrappers=[RedditWrapper, XWrapper, ChanWrapper]
|
||||
)
|
||||
|
||||
self.handler = WrapperHandler.build_wrappers(
|
||||
wrappers,
|
||||
|
||||
@@ -130,6 +130,34 @@ class WrapperHandler(Generic[WrapperType]):
|
||||
last_frame = traceback.extract_tb(e.__traceback__)[-1]
|
||||
return f"{e} [\"{last_frame.filename}\", line {last_frame.lineno}]"
|
||||
|
||||
@staticmethod
|
||||
def filter_wrappers_by_config(
|
||||
wrapper_map: dict[str, type[WrapperClassType]],
|
||||
provider_names: list[str],
|
||||
fallback_wrappers: list[type[WrapperClassType]] | None = None
|
||||
) -> list[type[WrapperClassType]]:
|
||||
"""
|
||||
Filters wrapper classes based on a list of provider names from configuration.
|
||||
|
||||
Args:
|
||||
wrapper_map (dict[str, type[W]]): Dictionary mapping provider names to wrapper classes.
|
||||
provider_names (list[str]): List of provider names from configuration.
|
||||
fallback_wrappers (list[type[W]] | None): Optional fallback list if no providers configured.
|
||||
|
||||
Returns:
|
||||
list[type[W]]: List of wrapper classes in the order specified by provider_names.
|
||||
"""
|
||||
wrappers: list[type[WrapperClassType]] = []
|
||||
for provider_name in provider_names:
|
||||
if provider_name in wrapper_map:
|
||||
wrappers.append(wrapper_map[provider_name])
|
||||
|
||||
# Fallback to all wrappers if none configured
|
||||
if not wrappers and fallback_wrappers:
|
||||
wrappers = fallback_wrappers
|
||||
|
||||
return wrappers
|
||||
|
||||
@staticmethod
|
||||
def build_wrappers(constructors: list[type[WrapperClassType]], try_per_wrapper: int = 3, retry_delay: int = 2, kwargs: dict[str, Any] | None = None) -> 'WrapperHandler[WrapperClassType]':
|
||||
"""
|
||||
|
||||
Reference in New Issue
Block a user