cleanup

openbb_platform/core/openbb_core/provider/utils/client.py

@@ -1,21 +1,22 @@
 """Aiohttp client."""
 # pylint: disable=protected-access
+import asyncio
 import random
+import re
 import zlib
 from typing import Any, Dict, Union
 
 import aiohttp
 
-FILTER_QUERY_PARAMS = [
-    "apikey",
-    "apiKey",
-    "api_key",
-    "token",
-    "key",
-    "auth_token",
-    "access_token",
-    "c",
-]
+FILTER_QUERY_REGEX = r".*key.*|.*token.*|.*auth.*|(c$)"
+
+
+def obfuscate(params: dict) -> dict:
+    """Obfuscate sensitive information."""
+    return {
+        param: "********" if re.match(FILTER_QUERY_REGEX, param, re.IGNORECASE) else val
+        for param, val in params.items()
+    }
 
 
 def get_user_agent() -> str:
@@ -36,6 +37,21 @@ def get_user_agent() -> str:
 class ClientResponse(aiohttp.ClientResponse):
     """Client response class."""
 
+    def __init__(self, *args, **kwargs):
+        kwargs["request_info"] = self.obfuscate_request_info(kwargs["request_info"])
+        super().__init__(*args, **kwargs)
+
+    @classmethod
+    def obfuscate_request_info(
+        cls, request_info: aiohttp.RequestInfo
+    ) -> aiohttp.RequestInfo:
+        """Remove sensitive information from request info."""
+        query = obfuscate(request_info.url.query.copy())
+        headers = obfuscate(request_info.headers.copy())
+        url = request_info.url.with_query(query)
+
+        return aiohttp.RequestInfo(url, request_info.method, headers, url)
+
     async def get_one(self) -> Dict[str, Any]:
         """Return the first item in the response."""
         data = await self.json()
@@ -44,6 +60,10 @@ class ClientResponse(aiohttp.ClientResponse):
 
         return data
 
+    async def json(self, **kwargs) -> Union[dict, list]:
+        """Return the json response."""
+        return await super().json(**kwargs)
+
 
 class ClientSession(aiohttp.ClientSession):
     _response_class: ClientResponse
@@ -58,6 +78,29 @@ class ClientSession(aiohttp.ClientSession):
 
         super().__init__(*args, **kwargs)
 
+    def __del__(self):
+        """Close the session."""
+        if not self.closed:
+            asyncio.create_task(self.close())
+
+    async def get(self, url: str, **kwargs) -> ClientResponse:
+        """Send GET request."""
+        return await self.request("GET", url, **kwargs)
+
+    async def post(self, url: str, **kwargs) -> ClientResponse:
+        """Send POST request."""
+        return await self.request("POST", url, **kwargs)
+
+    async def get_json(self, url: str, **kwargs) -> Union[dict, list]:
+        """Send GET request and return json."""
+        response = await self.request("GET", url, **kwargs)
+        return await response.json()
+
+    async def get_one(self, url: str, **kwargs) -> Dict[str, Any]:
+        """Send GET request and return first item in json if list."""
+        response = await self.request("GET", url, **kwargs)
+        return await response.get_one()
+
     async def request(
         self, *args, raise_for_status: bool = False, **kwargs
     ) -> ClientResponse:
@@ -76,26 +119,9 @@ class ClientSession(aiohttp.ClientSession):
             kwargs["headers"]["User-Agent"] = get_user_agent()
 
         response = await super().request(*args, **kwargs)
-        if raise_for_status and not response.ok:
-            query = response.request_info.url.query.copy()
 
-            for param in FILTER_QUERY_PARAMS:
-                if param in query:
-                    query[param] = "*****"
-
-            url = response.request_info.url.with_query(query)
-            raise aiohttp.ClientResponseError(
-                aiohttp.RequestInfo(
-                    url,
-                    response.request_info.method,
-                    response.request_info.headers,
-                    url,
-                ),
-                response.history,
-                status=response.status,
-                message=response.reason,
-                headers=response.headers,
-            )
+        if raise_for_status:
+            response.raise_for_status()
 
         encoding = response.headers.get("Content-Encoding", "")
         if encoding in ("gzip", "deflate") and not self.auto_decompress:
@@ -106,12 +132,3 @@ class ClientSession(aiohttp.ClientSession):
             )  # pylint: disable=protected-access
 
         return response
-
-    async def get(self, url: str, **kwargs) -> ClientResponse:
-        """Send GET request."""
-        return await self.request("GET", url, **kwargs)
-
-    async def get_json(self, url: str, **kwargs) -> Union[dict, list]:
-        """Send GET request and return json."""
-        response = await self.request("GET", url, **kwargs)
-        return await response.json()

openbb_platform/core/openbb_core/provider/utils/helpers.py

@@ -73,8 +73,10 @@ async def async_request(
         HTTP method to use.  Can be "GET" or "POST", by default "GET"
     timeout : int, optional
         Timeout in seconds, by default 10.  Can be overwritten by user setting, request_timeout
-    response_callback : Callable[[aiohttp.ClientResponse], Awaitable[Union[dict, List[dict]]]], optional
-        Callback to run on the response, by default None
+    response_callback : Callable[[ClientResponse, ClientSession], Awaitable[Union[dict, List[dict]]]], optional
+        Async callback with response and session as arguments that returns the json, by default None
+    session : ClientSession, optional
+        Custom session to use for requests, by default None
 
 
     Returns
@@ -82,36 +84,49 @@ async def async_request(
     Union[dict, List[dict]]
         Response json
     """
+    if method.upper() not in ["GET", "POST"]:
+        raise ValueError("Method must be GET or POST")
+
+    kwargs["timeout"] = kwargs.pop("preferences", {}).get("request_timeout", timeout)
+
     response_callback = response_callback or (
         lambda r, _: asyncio.ensure_future(r.json())
     )
 
-    kwargs["timeout"] = kwargs.pop("preferences", {}).get("request_timeout", timeout)
-    with_session = kwargs.pop("with_session", False)
+    with_session = kwargs.pop("with_session", "session" in kwargs)
     session: ClientSession = kwargs.pop("session", ClientSession())
 
-    response = await session.request(method, url, **kwargs)
-    data = await response_callback(response, session)
-
-    if not with_session:
-        await session.close()
-
-    return data
+    try:
+        response = await session.request(method, url, **kwargs)
+        return await response_callback(response, session)
+    finally:
+        if not with_session:
+            await session.close()
 
 
-async def async_requests(urls: List[str], **kwargs) -> Union[dict, List[dict]]:
+async def async_requests(
+    urls: Union[str, List[str]],
+    response_callback: Optional[
+        Callable[[ClientResponse, ClientSession], Awaitable[Union[dict, List[dict]]]]
+    ] = None,
+    **kwargs,
+):
     """Make multiple requests asynchronously.
 
+
     Parameters
     ----------
-    urls : List[str]
+    urls : Union[str, List[str]]
         List of urls to make requests to
     method : Literal["GET", "POST"], optional
         HTTP method to use.  Can be "GET" or "POST", by default "GET"
     timeout : int, optional
         Timeout in seconds, by default 10.  Can be overwritten by user setting, request_timeout
     response_callback : Callable[[ClientResponse, ClientSession], Awaitable[Union[dict, List[dict]]]], optional
-        Callback to run on the response, by default None
+        Async callback with response and session as arguments that returns the json, by default None
+    session : ClientSession, optional
+        Custom session to use for requests, by default None
+
 
     Returns
     -------
@@ -119,18 +134,21 @@ async def async_requests(urls: List[str], **kwargs) -> Union[dict, List[dict]]:
         Response json
     """
     session: ClientSession = kwargs.pop("session", ClientSession())
-    kwargs["with_session"] = True
+    kwargs["response_callback"] = response_callback
 
-    results = await asyncio.gather(
-        *[async_request(url, session=session, **kwargs) for url in urls]
-    )
+    urls = urls if isinstance(urls, list) else [urls]
 
-    await session.close()
+    try:
+        results = await asyncio.gather(
+            *[async_request(url, session=session, **kwargs) for url in urls]
+        )
 
-    if isinstance(results[0], list):
-        return [item for sublist in results for item in sublist]
+        if isinstance(results[0], list):
+            return [item for sublist in results for item in sublist if item]
 
-    return results
+        return results
+    finally:
+        await session.close()
 
 
 def make_request(

openbb_platform/providers/fmp/openbb_fmp/models/calendar_dividend.py

@@ -22,6 +22,8 @@ class FMPDividendCalendarQueryParams(DividendCalendarQueryParams):
     The maximum time interval between the start and end date can be 3 months.
     """
 
+    __alias_dict__ = {"start_date": "from", "end_date": "to"}
+
 
 class FMPDividendCalendarData(DividendCalendarData):
     """FMP Dividend Calendar Data."""
@@ -88,7 +90,6 @@ class FMPDividendCalendarFetcher(
 
         base_url = "https://financialmodelingprep.com/api/v3"
         query_str = get_querystring(query.model_dump(), [])
-        query_str = query_str.replace("start_date", "from").replace("end_date", "to")
         url = f"{base_url}/stock_dividend_calendar?{query_str}&apikey={api_key}"
 
         return await get_data_many(url, **kwargs)

openbb_platform/providers/fmp/openbb_fmp/models/crypto_historical.py

@@ -23,6 +23,8 @@ class FMPCryptoHistoricalQueryParams(CryptoHistoricalQueryParams):
     https://site.financialmodelingprep.com/developer/docs/cryptocurrency-historical-data-api/#Historical-Daily-Prices
     """
 
+    __alias_dict__ = {"start_date": "from", "end_date": "to"}
+
     timeseries: Optional[NonNegativeInt] = Field(
         default=None, description="Number of days to look back."
     )
@@ -89,11 +91,7 @@ class FMPCryptoHistoricalFetcher(
         api_key = credentials.get("fmp_api_key") if credentials else ""
 
         base_url = "https://financialmodelingprep.com/api/v3"
-        query_str = (
-            get_querystring(query.model_dump(), ["symbol"])
-            .replace("start_date", "from")
-            .replace("end_date", "to")
-        )
+        query_str = get_querystring(query.model_dump(), ["symbol"])
 
         url_params = f"{query.symbol}?{query_str}&apikey={api_key}"
         url = f"{base_url}/historical-chart/{query.interval}/{url_params}"

openbb_platform/providers/fmp/openbb_fmp/models/currency_historical.py

@@ -20,6 +20,8 @@ class FMPCurrencyHistoricalQueryParams(CurrencyHistoricalQueryParams):
     Source: https://site.financialmodelingprep.com/developer/docs/#Historical-Forex-Price
     """
 
+    __alias_dict__ = {"start_date": "from", "end_date": "to"}
+
     interval: Literal[
         "1min", "5min", "15min", "30min", "1hour", "4hour", "1day"
     ] = Field(default="1day", description="Data granularity.")
@@ -82,11 +84,7 @@ class FMPCurrencyHistoricalFetcher(
         api_key = credentials.get("fmp_api_key") if credentials else ""
 
         base_url = "https://financialmodelingprep.com/api/v3"
-        query_str = (
-            get_querystring(query.model_dump(), ["symbol"])
-            .replace("start_date", "from")
-            .replace("end_date", "to")
-        )
+        query_str = get_querystring(query.model_dump(), ["symbol"])
 
         url_params = f"{query.symbol}?{query_str}&apikey={api_key}"
         url = f"{base_url}/historical-chart/{query.interval}/{url_params}"

openbb_platform/providers/fmp/openbb_fmp/models/equity_historical.py

@@ -3,13 +3,13 @@
 from datetime import datetime
 from typing import Any, Dict, List, Literal, Optional
 
-from aiohttp import ClientResponse, ClientSession
 from dateutil.relativedelta import relativedelta
 from openbb_core.provider.abstract.fetcher import Fetcher
 from openbb_core.provider.standard_models.equity_historical import (
     EquityHistoricalData,
     EquityHistoricalQueryParams,
 )
+from openbb_core.provider.utils.client import ClientResponse
 from openbb_core.provider.utils.helpers import async_requests, get_querystring
 from openbb_fmp.utils.helpers import get_interval
 from pydantic import Field, NonNegativeInt
@@ -108,9 +108,7 @@ class FMPEquityHistoricalFetcher(
         if len(query.symbol.split(",")) > 20:
             kwargs.update({"preferences": {"request_timeout": 30}})
 
-        async def response_callback(
-            response: ClientResponse, _: ClientSession
-        ) -> List[Dict]:
+        async def callback(response: ClientResponse, _: Any) -> List[Dict]:
             data: dict = await response.json()
             symbol = response.url.parts[-1]
 
@@ -125,7 +123,7 @@ class FMPEquityHistoricalFetcher(
 
         urls = [get_url_params(symbol) for symbol in query.symbol.split(",")]
 
-        return await async_requests(urls, response_callback=response_callback, **kwargs)
+        return await async_requests(urls, callback, **kwargs)
 
     @staticmethod
     def transform_data(

openbb_platform/providers/fmp/openbb_fmp/models/financial_ratios.py

@@ -74,9 +74,7 @@ class FMPFinancialRatiosFetcher(
 
             # TTM data
             ttm_url = f"{base_url}/ratios-ttm/{symbol}?&apikey={api_key}"
-            if query.with_ttm and (ratios_response := await session.get(ttm_url)):
-                ratios_ttm = await ratios_response.get_one()
-
+            if query.with_ttm and (ratios_ttm := await session.get_one(ttm_url)):
                 results.insert(
                     0,
                     {

openbb_platform/providers/fmp/openbb_fmp/models/key_metrics.py

@@ -74,9 +74,7 @@ class FMPKeyMetricsFetcher(
 
             # TTM data
             ttm_url = f"{base_url}/key-metrics-ttm/{symbol}?&apikey={api_key}"
-            if query.with_ttm and (metrics_response := await session.get(ttm_url)):
-                metrics_ttm = await metrics_response.get_one()
-
+            if query.with_ttm and (metrics_ttm := await session.get_one(ttm_url)):
                 results.insert(
                     0,
                     {

openbb_platform/providers/polygon/openbb_polygon/models/equity_historical.py

@@ -117,15 +117,24 @@ class PolygonEquityHistoricalFetcher(
         """Return the raw data from the Polygon endpoint."""
         api_key = credentials.get("polygon_api_key") if credentials else ""
 
-        async def response_callback(
+        urls = [
+            (
+                "https://api.polygon.io/v2/aggs/ticker/"
+                f"{symbol.upper()}/range/{query._multiplier}/{query._timespan}/"
+                f"{query.start_date}/{query.end_date}?adjusted={query.adjusted}"
+                f"&sort={query.sort}&limit={query.limit}&apiKey={api_key}"
+            )
+            for symbol in query.symbol.split(",")
+        ]
+
+        async def callback(
             response: ClientResponse, session: ClientSession
         ) -> List[Dict]:
-            data: dict = await response.json()
-
-            next_url = data.get("next_url", None)
-            results: list = data.get("results", [])
+            data = await response.json()
 
             symbol = response.url.parts[4]
+            next_url = data.get("next_url", None)
+            results: list = data.get("results", [])
 
             while next_url:
                 url = f"{next_url}&apiKey={api_key}"
@@ -142,17 +151,7 @@ class PolygonEquityHistoricalFetcher(
 
             return results
 
-        urls = [
-            (
-                "https://api.polygon.io/v2/aggs/ticker/"
-                f"{symbol.upper()}/range/{query._multiplier}/{query._timespan}/"
-                f"{query.start_date}/{query.end_date}?adjusted={query.adjusted}"
-                f"&sort={query.sort}&limit={query.limit}&apiKey={api_key}"
-            )
-            for symbol in query.symbol.split(",")
-        ]
-
-        return await async_requests(urls, response_callback=response_callback, **kwargs)
+        return await async_requests(urls, callback, **kwargs)
 
     @staticmethod
     def transform_data(