CMCD-456-add the code of aioretry in copernicusmarine toolbox

copernicusmarine/aioretry/LICENSE

@@ -0,0 +1,20 @@
+Copyright (c) 2013 kaelzhang <>, contributors
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+The above copyright notice and this permission notice shall be
+included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

copernicusmarine/aioretry/__init__.py

@@ -0,0 +1,5 @@
+from .retry import BeforeRetry  # noqa: F401
+from .retry import RetryInfo  # noqa: F401
+from .retry import RetryPolicy  # noqa: F401
+from .retry import RetryPolicyStrategy  # noqa: F401
+from .retry import retry  # noqa: F401

copernicusmarine/aioretry/retry.py

@@ -0,0 +1,162 @@
+import asyncio
+import inspect
+import warnings
+from datetime import datetime
+from typing import Awaitable, Callable, Optional, Tuple, TypeVar, Union
+
+
+# Copyright from: https://github.com/kaelzhang/python-aioretry
+class RetryInfo:
+    __slots__ = ("fails", "exception", "since")
+
+    fails: int
+    exception: Exception
+    since: datetime
+
+    def __init__(
+        self, fails: int, exception: Exception, since: datetime
+    ) -> None:
+        self.fails = fails
+        self.exception = exception
+        self.since = since
+
+    def update(self, exception: Exception) -> "RetryInfo":
+        """Create a new RetryInfo and update fails and exception
+
+        Why?
+            The object might be collected by user,
+            so we need to create a new object every time it fails.
+        """
+
+        return RetryInfo(self.fails + 1, exception, self.since)
+
+
+RetryPolicyStrategy = Tuple[bool, Union[int, float]]
+
+RetryPolicy = Callable[[RetryInfo], RetryPolicyStrategy]
+BeforeRetry = Callable[[RetryInfo], Optional[Awaitable[None]]]
+
+ParamRetryPolicy = Union[RetryPolicy, str]
+ParamBeforeRetry = Union[BeforeRetry, str]
+
+TargetFunction = Callable[..., Awaitable]
+Exceptions = Tuple[Exception, ...]
+ExceptionsOrException = Union[Exceptions, Exception]
+
+T = TypeVar("T", RetryPolicy, BeforeRetry)
+
+
+async def await_coro(coro):
+    if inspect.isawaitable(coro):
+        return await coro
+
+    return coro
+
+
+def warn(method_name: str, exception: Exception):
+    warnings.warn(
+        f"""[aioretry] {method_name} raises an exception:
+    {exception}
+It is usually a bug that you should fix!""",
+        UserWarning,
+        stacklevel=2,
+    )
+
+
+async def perform(
+    fn: TargetFunction,
+    retry_policy: RetryPolicy,
+    before_retry: Optional[BeforeRetry],
+    *args,
+    **kwargs,
+):
+    info = None
+
+    while True:
+        try:
+            return await fn(*args, **kwargs)
+        except Exception as e:
+            if info is None:
+                info = RetryInfo(1, e, datetime.now())
+            else:
+                info = info.update(e)
+
+            try:
+                abandon, delay = retry_policy(info)
+            except Exception as e2:
+                warn("retry_policy", e2)
+                raise e2
+
+            if abandon:
+                raise e
+
+            if before_retry is not None:
+                try:
+                    await await_coro(before_retry(info))
+                except Exception as e:
+                    warn("before_retry", e)
+                    raise e
+
+            # `delay` could be 0
+            if delay > 0:
+                await asyncio.sleep(delay)
+
+
+def get_method(
+    target: Union[T, str],
+    args: Tuple,
+    name: str,
+) -> T:
+    if not isinstance(target, str):
+        return target
+
+    if len(args) == 0:
+        raise RuntimeError(
+            f"[aioretry] decorator should be used for instance method"
+            f" if {name} as a str '{target}', which should be fixed"
+        )
+
+    self = args[0]
+
+    return getattr(self, target)  # type: ignore
+
+
+def retry(
+    retry_policy: ParamRetryPolicy,
+    before_retry: Optional[ParamBeforeRetry] = None,
+) -> Callable[[TargetFunction], TargetFunction]:
+    """Creates a decorator function
+
+    Args:
+        retry_policy (RetryPolicy, str): the retry policy
+        before_retry (BeforeRetry, str, None): the function to
+            be called after each failure of fn
+            and before the corresponding retry.
+
+    Returns:
+        A wrapped function which accepts the same arguments as
+        fn and returns an Awaitable
+
+    Usage::
+        @retry(retry_policy)
+        async def coro_func():
+            ...
+    """
+
+    def wrapper(fn: TargetFunction) -> TargetFunction:
+        async def wrapped(*args, **kwargs):
+            return await perform(
+                fn,
+                get_method(retry_policy, args, "retry_policy"),
+                (
+                    get_method(before_retry, args, "before_retry")
+                    if before_retry is not None
+                    else None
+                ),
+                *args,
+                **kwargs,
+            )
+
+        return wrapped
+
+    return wrapper

copernicusmarine/catalogue_parser/catalogue_parser.py

@@ -15,10 +15,10 @@
 import nest_asyncio
 import pystac
 from aiohttp import ContentTypeError
-from aioretry import RetryInfo, RetryPolicyStrategy, retry
 from cachier.core import cachier
 from tqdm import tqdm
 
+from copernicusmarine.aioretry import RetryInfo, RetryPolicyStrategy, retry
 from copernicusmarine.command_line_interface.exception_handler import (
     log_exception_debug,
 )
@@ -702,7 +702,8 @@ def _construct_marine_data_store_product(
     )
 
     datasets = map(
-        _construct_marine_data_store_dataset, datacubes_by_id  # type: ignore
+        _construct_marine_data_store_dataset,  # type: ignore
+        datacubes_by_id,  # type: ignore
     )
 
     production_center = [

pyproject.toml

@@ -26,7 +26,6 @@ semver = ">=3.0.2"
 nest-asyncio = ">=1.5.8"
 pystac = ">=1.8.3"
 lxml = ">=4.9.0"
-aioretry = "^5.0.2"
 
 [tool.poetry.dev-dependencies]
 pre-commit = "^2.20.0"