Berack/upo-app-agents
0
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

Refactor news API methods to use 'limit' parameter instead of 'total' for consistency across wrappers

Browse Source
This commit is contained in:
Giacomo Bertolazzi
2025-10-01 10:26:21 +02:00
parent 99ebb420fa
commit e4e7023c17
12 changed files with 74 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/app/news/__init__.py
View File

@@ -47,10 +47,10 @@ class NewsAPIsTool(NewsWrapper, Toolkit):
# TODO Pensare se ha senso restituire gli articoli da TUTTI i wrapper o solo dal primo che funziona
# la modifica è banale, basta usare try_call_all invece di try_call
def get_top_headlines(self, total: int = 100) -> list[Article]:
return self.wrapper_handler.try_call(lambda w: w.get_top_headlines(total))
def get_latest_news(self, query: str, total: int = 100) -> list[Article]:
return self.wrapper_handler.try_call(lambda w: w.get_latest_news(query, total))
def get_top_headlines(self, limit: int = 100) -> list[Article]:
return self.wrapper_handler.try_call(lambda w: w.get_top_headlines(limit))
def get_latest_news(self, query: str, limit: int = 100) -> list[Article]:
return self.wrapper_handler.try_call(lambda w: w.get_latest_news(query, limit))
NEWS_INSTRUCTIONS = """

10
src/app/news/base.py
View File

@@ -12,22 +12,22 @@ class NewsWrapper:
All news API wrappers should inherit from this class and implement the methods.
"""
def get_top_headlines(self, total: int = 100) -> list[Article]:
def get_top_headlines(self, limit: int = 100) -> list[Article]:
"""
Get top headlines, optionally limited by total.
Get top headlines, optionally limited by limit.
Args:
total (int): The maximum number of articles to return.
limit (int): The maximum number of articles to return.
Returns:
list[Article]: A list of Article objects.
"""
raise NotImplementedError("This method should be overridden by subclasses")
def get_latest_news(self, query: str, total: int = 100) -> list[Article]:
def get_latest_news(self, query: str, limit: int = 100) -> list[Article]:
"""
Get latest news based on a query.
Args:
query (str): The search query.
total (int): The maximum number of articles to return.
limit (int): The maximum number of articles to return.
Returns:
list[Article]: A list of Article objects.
"""

8
src/app/news/cryptopanic_api.py
View File

@@ -62,10 +62,10 @@ class CryptoPanicWrapper(NewsWrapper):
def set_filter(self, filter: CryptoPanicFilter):
self.filter = filter
def get_top_headlines(self, total: int = 100) -> list[Article]:
return self.get_latest_news("", total) # same endpoint so just call the other method
def get_top_headlines(self, limit: int = 100) -> list[Article]:
return self.get_latest_news("", limit) # same endpoint so just call the other method
def get_latest_news(self, query: str, total: int = 100) -> list[Article]:
def get_latest_news(self, query: str, limit: int = 100) -> list[Article]:
params = self.get_base_params()
params['currencies'] = query
@@ -74,4 +74,4 @@ class CryptoPanicWrapper(NewsWrapper):
json_response = response.json()
articles = get_articles(json_response)
return articles[:total]
return articles[:limit]

8
src/app/news/duckduckgo.py
View File

@@ -20,13 +20,13 @@ class DuckDuckGoWrapper(NewsWrapper):
self.tool = DuckDuckGoTools()
self.query = "crypto"
def get_top_headlines(self, total: int = 100) -> list[Article]:
results = self.tool.duckduckgo_news(self.query, max_results=total)
def get_top_headlines(self, limit: int = 100) -> list[Article]:
results = self.tool.duckduckgo_news(self.query, max_results=limit)
json_results = json.loads(results)
return [create_article(result) for result in json_results]
def get_latest_news(self, query: str, total: int = 100) -> list[Article]:
results = self.tool.duckduckgo_news(query or self.query, max_results=total)
def get_latest_news(self, query: str, limit: int = 100) -> list[Article]:
results = self.tool.duckduckgo_news(query or self.query, max_results=limit)
json_results = json.loads(results)
return [create_article(result) for result in json_results]

8
src/app/news/googlenews.py
View File

@@ -15,8 +15,8 @@ class GoogleNewsWrapper(NewsWrapper):
It does not require an API key and is free to use.
"""
def get_top_headlines(self, total: int = 100) -> list[Article]:
gnews = GNews(language='en', max_results=total, period='7d')
def get_top_headlines(self, limit: int = 100) -> list[Article]:
gnews = GNews(language='en', max_results=limit, period='7d')
results = gnews.get_top_news()
articles = []
@@ -25,8 +25,8 @@ class GoogleNewsWrapper(NewsWrapper):
articles.append(article)
return articles
def get_latest_news(self, query: str, total: int = 100) -> list[Article]:
gnews = GNews(language='en', max_results=total, period='7d')
def get_latest_news(self, query: str, limit: int = 100) -> list[Article]:
gnews = GNews(language='en', max_results=limit, period='7d')
results = gnews.get_news(query)
articles = []

17
src/app/news/news_api.py
View File

@@ -26,22 +26,25 @@ class NewsApiWrapper(NewsWrapper):
self.language = "en" # TODO Only English articles for now?
self.max_page_size = 100
def get_top_headlines(self, total: int = 100) -> list[Article]:
page_size = min(self.max_page_size, total)
pages = (total // page_size) + (1 if total % page_size > 0 else 0)
def __calc_pages(self, limit: int, page_size: int) -> tuple[int, int]:
page_size = min(self.max_page_size, limit)
pages = (limit // page_size) + (1 if limit % page_size > 0 else 0)
return pages, page_size
def get_top_headlines(self, limit: int = 100) -> list[Article]:
pages, page_size = self.__calc_pages(limit, self.max_page_size)
articles = []
for page in range(1, pages + 1):
headlines = self.client.get_top_headlines(q="", category=self.category, language=self.language, page_size=page_size, page=page)
results = [result_to_article(article) for article in headlines.get("articles", [])]
articles.extend(results)
return articles
def get_latest_news(self, query: str, total: int = 100) -> list[Article]:
page_size = min(self.max_page_size, total)
pages = (total // page_size) + (1 if total % page_size > 0 else 0)
def get_latest_news(self, query: str, limit: int = 100) -> list[Article]:
pages, page_size = self.__calc_pages(limit, self.max_page_size)
articles = []
for page in range(1, pages + 1):
everything = self.client.get_everything(q=query, language=self.language, sort_by="publishedAt", page_size=page_size, page=page)
results = [result_to_article(article) for article in everything.get("articles", [])]

2
tests/api/test_cryptopanic_api.py
View File

@@ -15,7 +15,7 @@ class TestCryptoPanicAPI:
def test_crypto_panic_api_get_latest_news(self):
crypto = CryptoPanicWrapper()
articles = crypto.get_latest_news(query="", total=2)
articles = crypto.get_latest_news(query="", limit=2)
assert isinstance(articles, list)
assert len(articles) == 2
for article in articles:

4
tests/api/test_duckduckgo_news.py
View File

@@ -12,7 +12,7 @@ class TestDuckDuckGoNews:
def test_duckduckgo_get_latest_news(self):
news = DuckDuckGoWrapper()
articles = news.get_latest_news(query="crypto", total=2)
articles = news.get_latest_news(query="crypto", limit=2)
assert isinstance(articles, list)
assert len(articles) == 2
for article in articles:
@@ -23,7 +23,7 @@ class TestDuckDuckGoNews:
def test_duckduckgo_get_top_headlines(self):
news = DuckDuckGoWrapper()
articles = news.get_top_headlines(total=2)
articles = news.get_top_headlines(limit=2)
assert isinstance(articles, list)
assert len(articles) == 2
for article in articles:

4
tests/api/test_google_news.py
View File

@@ -12,7 +12,7 @@ class TestGoogleNews:
def test_gnews_api_get_latest_news(self):
gnews_api = GoogleNewsWrapper()
articles = gnews_api.get_latest_news(query="crypto", total=2)
articles = gnews_api.get_latest_news(query="crypto", limit=2)
assert isinstance(articles, list)
assert len(articles) == 2
for article in articles:
@@ -23,7 +23,7 @@ class TestGoogleNews:
def test_gnews_api_get_top_headlines(self):
news_api = GoogleNewsWrapper()
articles = news_api.get_top_headlines(total=2)
articles = news_api.get_top_headlines(limit=2)
assert isinstance(articles, list)
assert len(articles) == 2
for article in articles:

4
tests/api/test_news_api.py
View File

@@ -14,7 +14,7 @@ class TestNewsAPI:
def test_news_api_get_latest_news(self):
news_api = NewsApiWrapper()
articles = news_api.get_latest_news(query="crypto", total=2)
articles = news_api.get_latest_news(query="crypto", limit=2)
assert isinstance(articles, list)
assert len(articles) > 0 # Ensure we got some articles (apparently it doesn't always return the requested number)
for article in articles:
@@ -26,7 +26,7 @@ class TestNewsAPI:
def test_news_api_get_top_headlines(self):
news_api = NewsApiWrapper()
articles = news_api.get_top_headlines(total=2)
articles = news_api.get_top_headlines(limit=2)
assert isinstance(articles, list)
# assert len(articles) > 0 # apparently it doesn't always return SOME articles
for article in articles:

8
tests/tools/test_news_tool.py
View File

@@ -13,7 +13,7 @@ class TestNewsAPITool:
def test_news_api_tool_get_top(self):
tool = NewsAPIsTool()
result = tool.wrapper_handler.try_call(lambda w: w.get_top_headlines(total=2))
result = tool.wrapper_handler.try_call(lambda w: w.get_top_headlines(limit=2))
assert isinstance(result, list)
assert len(result) > 0
for article in result:
@@ -22,7 +22,7 @@ class TestNewsAPITool:
def test_news_api_tool_get_latest(self):
tool = NewsAPIsTool()
result = tool.wrapper_handler.try_call(lambda w: w.get_latest_news(query="crypto", total=2))
result = tool.wrapper_handler.try_call(lambda w: w.get_latest_news(query="crypto", limit=2))
assert isinstance(result, list)
assert len(result) > 0
for article in result:
@@ -31,7 +31,7 @@ class TestNewsAPITool:
def test_news_api_tool_get_top__all_results(self):
tool = NewsAPIsTool()
result = tool.wrapper_handler.try_call_all(lambda w: w.get_top_headlines(total=2))
result = tool.wrapper_handler.try_call_all(lambda w: w.get_top_headlines(limit=2))
assert isinstance(result, dict)
assert len(result.keys()) > 0
print("Results from providers:", result.keys())
@@ -43,7 +43,7 @@ class TestNewsAPITool:
def test_news_api_tool_get_latest__all_results(self):
tool = NewsAPIsTool()
result = tool.wrapper_handler.try_call_all(lambda w: w.get_latest_news(query="crypto", total=2))
result = tool.wrapper_handler.try_call_all(lambda w: w.get_latest_news(query="crypto", limit=2))
assert isinstance(result, dict)
assert len(result.keys()) > 0
print("Results from providers:", result.keys())

32
tests/tools/test_socials_tool.py Normal file
View File

@@ -0,0 +1,32 @@
import pytest
from app.social import SocialAPIsTool
@pytest.mark.tools
@pytest.mark.social
@pytest.mark.api
class TestSocialAPIsTool:
def test_social_api_tool(self):
tool = SocialAPIsTool()
assert tool is not None
def test_social_api_tool_get_top(self):
tool = SocialAPIsTool()
result = tool.wrapper_handler.try_call(lambda w: w.get_top_crypto_posts(limit=2))
assert isinstance(result, list)
assert len(result) > 0
for post in result:
assert post.title is not None
assert post.time is not None
def test_social_api_tool_get_top__all_results(self):
tool = SocialAPIsTool()
result = tool.wrapper_handler.try_call_all(lambda w: w.get_top_crypto_posts(limit=2))
assert isinstance(result, dict)
assert len(result.keys()) > 0
print("Results from providers:", result.keys())
for provider, posts in result.items():
for post in posts:
print(provider, post.title)
assert post.title is not None
assert post.time is not None