The retrying_async library is not Python 3.10 ready, so we are providing
a 3.10 compatible callback for it to use instead of the default one
Change-Id: I6e98f6d7ebc2fe134b0e3fe37d180e383044b30b
Signed-off-by: Mark Beierl <mark.beierl@canonical.com>
else "--skip-repos",
)
try:
else "--skip-repos",
)
try:
self._local_async_exec(command=command, raise_exception_on_error=False)
)
except Exception as e:
self._local_async_exec(command=command, raise_exception_on_error=False)
)
except Exception as e:
RBAC_LABEL_KEY_NAME = "rbac-id"
RBAC_LABEL_KEY_NAME = "rbac-id"
+@asyncio.coroutine
+def retry_callback(attempt, exc, args, kwargs, delay=0.5, *, loop):
+ # Specifically overridden from upstream implementation so it can
+ # continue to work with Python 3.10
+ yield from asyncio.sleep(attempt * delay)
+ return retry
+
+
class Libjuju:
def __init__(
self,
class Libjuju:
def __init__(
self,
if controller:
await controller.disconnect()
if controller:
await controller.disconnect()
- @retry(attempts=3, delay=5, timeout=None)
+ @retry(attempts=3, delay=5, timeout=None, callback=retry_callback)
async def add_model(self, model_name: str, cloud: VcaCloud):
"""
Create model
async def add_model(self, model_name: str, cloud: VcaCloud):
"""
Create model
await self.disconnect_controller(controller)
return application_configs
await self.disconnect_controller(controller)
return application_configs
- @retry(attempts=3, delay=5)
+ @retry(attempts=3, delay=5, callback=retry_callback)
async def get_model(self, controller: Controller, model_name: str) -> Model:
"""
Get model from controller
async def get_model(self, controller: Controller, model_name: str) -> Model:
"""
Get model from controller
finally:
await self.disconnect_controller(controller)
finally:
await self.disconnect_controller(controller)
- @retry(attempts=20, delay=5, fallback=JujuLeaderUnitNotFound())
+ @retry(
+ attempts=20, delay=5, fallback=JujuLeaderUnitNotFound(), callback=retry_callback
+ )
async def _get_leader_unit(self, application: Application) -> Unit:
unit = None
for u in application.units:
async def _get_leader_unit(self, application: Application) -> Unit:
unit = None
for u in application.units: