Update publish-snapshot.yml

.github/workflows/publish-snapshot.yml

@@ -11,7 +11,7 @@ on:
 jobs:
   build-job:
     name: Check and publish snapshot tarball to PyPi
-    uses: th2-net/.github/.github/workflows/compound-python.yml@main
+    uses: th2-net/.github/.github/workflows/compound-python.yml@vulnerability-scan-fix
     with:
       release-type: development
       style-check-enabled: false
@@ -23,3 +23,5 @@ jobs:
 
     secrets:
       pypi-password: ${{ secrets.PYPI_PASSWORD }}
+
+

package_info.json

@@ -1,5 +1,5 @@
 {
   "package_name": "th2-data-services-lwdp",
-  "package_version": "3.1.1.0"
+  "package_version": "3.1.2.0"
 }
 

release_notes.md

@@ -273,3 +273,5 @@ BugFixes without ticket
 
 ## Improvements
 1. [TH2-5260] More details are provided in case an exception occurs in `StatusUpdateManager.update()`.
+2. [TH2-5263] Retries are added to `IHTTPDataSource.check_connect()`.
+3. [TH2-5264] Retries are added to `nest_asyncio.apply()`.

requirements.txt

@@ -8,3 +8,4 @@ orjson>=3.10,<4
 aiohttp>=3.9.5,<4
 nest-asyncio>=1.6.0,<2
 attrs!=24.3.0  # Temporary solution to license problem
+retry>=0.9,<1

th2_data_services/data_source/lwdp/commands/http.py

@@ -56,16 +56,28 @@
     _check_list_or_tuple,
     _check_response_formats,
 )
+from th2_data_services.data_source.lwdp.utils._retry_logger import RetryLogger
 from th2_data_services.data_source.lwdp.utils.iter_status_manager import StatusUpdateManager
 from th2_data_services.data_source.lwdp.utils._misc import (
     get_utc_datetime_now,
     _get_response_format,
 )
 from th2_data_services.utils._json import BufferedJSONProcessor
 from th2_data_services.data_source.lwdp.page import PageNotFound
+from retry.api import retry_call
+from retry import retry
 
 Event = dict
-nest_asyncio.apply()  # This patch allows nested use of asyncio.run() in environments with an existing event loop.
+retry_call(
+    # This patch allows nested use of asyncio.run() in environments with an existing event loop.
+    # This Retry mechanism is required as workaround because we often face
+    #   "Only one usage of each socket address (protocol/network address/port)
+    #   is normally permitted" issue on Windows.
+    nest_asyncio.apply,
+    delay=5,
+    tries=10,
+    logger=RetryLogger(nest_asyncio.apply),
+)
 
 # Available stream formats:
 # 1) str
@@ -627,24 +639,26 @@ def handle(self, data_source: DataSource) -> dict:  # noqa: D102
         else:
             return response.json()
 
-    async def async_handle(self, data_source: DataSource) -> dict:  # noqa: D102
+    async def async_handle(self, data_source: DataSource, session=None) -> dict:  # noqa: D102
         api: API = data_source.source_api
         url = api.get_url_find_event_by_id(self._id)
 
-        async with aiohttp.ClientSession() as session:
-            async with session.get(url) as response:
-                json_response = await response.text()
+        if session is None:
+            session = aiohttp.ClientSession()
 
-                if response.status == 404 and self._stub_status:
-                    stub = data_source.event_stub_builder.build(
-                        {data_source.event_struct.EVENT_ID: self._id}
-                    )
-                    return stub
-                elif response.status == 404:
-                    # LOG             logger.error(f"Unable to find the message. Id: {self._id}")
-                    raise EventNotFound(self._id, "Unable to find the event")
-                else:
-                    return orjson.loads(json_response)
+        async with session.get(url) as response:
+            json_response = await response.text()
+
+            if response.status == 404 and self._stub_status:
+                stub = data_source.event_stub_builder.build(
+                    {data_source.event_struct.EVENT_ID: self._id}
+                )
+                return stub
+            elif response.status == 404:
+                # LOG             logger.error(f"Unable to find the message. Id: {self._id}")
+                raise EventNotFound(self._id, "Unable to find the event")
+            else:
+                return orjson.loads(json_response)
 
 
 class GetEventsById(IHTTPCommand):
@@ -671,6 +685,7 @@ def __init__(self, ids: List[str], use_stub=False):
         self._ids: ids = ids
         self._stub_status = use_stub
 
+    @retry(tries=5, delay=5, logger=RetryLogger("handle"))
     def handle(self, data_source: DataSource) -> List[dict]:  # noqa: D102
         # return self._sync_handle(data_source)
         return asyncio.run(self._async_handle(data_source))
@@ -685,13 +700,16 @@ def _sync_handle(self, data_source: DataSource) -> List[dict]:  # noqa: D102
 
     async def _async_handle(self, data_source: DataSource) -> List[dict]:  # noqa: D102
         coros = []
-        for event_id in self._ids:
-            co_event = GetEventById(event_id, use_stub=self._stub_status).async_handle(data_source)
-            coros.append(co_event)
+        async with aiohttp.ClientSession() as session:
+            for event_id in self._ids:
+                co_event = GetEventById(event_id, use_stub=self._stub_status).async_handle(
+                    data_source, session
+                )
+                coros.append(co_event)
 
-        events = await asyncio.gather(*coros)
+            events = await asyncio.gather(*coros)
 
-        return events
+            return events
 
 
 class GetEventsByPage(IHTTPCommand):

th2_data_services/data_source/lwdp/interfaces/data_source.py

@@ -25,6 +25,8 @@
 from th2_data_services.data_source.lwdp.interfaces.source_api import IHTTPSourceAPI
 from th2_data_services.data_source.lwdp.struct import IEventStruct, IMessageStruct
 from th2_data_services.data_source.lwdp.stub_builder import IEventStub, IMessageStub
+from th2_data_services.data_source.lwdp.utils._retry_logger import RetryLogger
+from retry import retry
 
 CommandT = TypeVar("CommandT", bound="ILwDPCommand")
 EventStructT = TypeVar("EventStructT", bound="IEventStruct")
@@ -110,6 +112,10 @@ class IHTTPDataSource(
     def command(self, cmd):
         """Execute the transmitted HTTP command."""
 
+    @retry(tries=5, delay=5, logger=RetryLogger("check_connect"))
+    # This Retry mechanism is required as workaround because we often face
+    #   "Only one usage of each socket address (protocol/network address/port)
+    #   is normally permitted" issue on Windows
     def check_connect(self, timeout: Tuple[int, float], certification: bool = True) -> None:
         """Checks whether url is working.
 

th2_data_services/data_source/lwdp/utils/_retry_logger.py

@@ -0,0 +1,36 @@
+#  Copyright 2024 Exactpro (Exactpro Systems Limited)
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+
+class RetryLogger:
+    def __init__(self, name):
+        """Initialize the RetryLogger.
+
+        Args:
+            name (Union[str, Callable]): The name of the function being retried.
+                If a function is provided, its __name__ attribute will be used.
+        """
+        if callable(name):
+            name = name.__name__
+        self._name = name
+
+    def warning(self, fmt, error, delay):
+        """Log a warning message when a function execution fails and will be retried.
+
+        Args:
+            fmt: Format string for the warning message (unused).
+            error: The error that caused the function to fail.
+            delay: Number of seconds before the next retry attempt.
+        """
+        print(f"Function {self._name} failed with {error}. Retrying in {delay} seconds.")