I run botocove with a input function that raises some custom exceptions.
When the custom exception has the same initializer signature as Exception botocove works as expected.
But one of the exceptions has a different initializer signature. And when it is raised, botocove fails with a TypeError when constructing the main CoveOutput object.
The problem occurs in the dataclass_converter function. See the TypeError in the output below.
I see the intention of the dataclass_converter is to convert all the output types to dicts, but I'm not sure why it needs to do that. A comment in the decorator function suggests it exists for backwards compatibility.
So you may reproduce the result, I have provided some demo code that defines two input functions and two dataclass_converter implementations. Botocove runs once for all possible function-converter pairs and prints the results.
The test also disables tqdm to avoid distractions in the output.
from functools import partial, update_wrapper
from itertools import product, starmap
from pprint import pprint
import sys
import traceback
import boto3
from botocove import cove
import botocove.cove_decorator
class NormalException(Exception):
pass
class FancyException(Exception):
def __init__(self, *, fancy_thing):
self.fancy_thing = fancy_thing
Exception.__init__(self, f"Problem with the {fancy_thing=}")
def normal_fail(session):
raise NormalException("normal error")
def fancy_fail(session):
raise FancyException(fancy_thing="fancy error")
def run_cove_and_return_result_or_exception(target_account_id, func, converter):
botocove.cove_decorator.dataclass_converter = converter
try:
result = cove(func, target_ids=[target_account_id])()
return result
except Exception as ex:
return traceback.format_exc()
dataclass_converter = botocove.cove_decorator.dataclass_converter
def identity(i):
return i
def disable_tqdm():
def tqdm_passthrough(iterable, **kwargs):
return iterable
botocove.cove_sessions.tqdm = tqdm_passthrough
botocove.cove_runner.tqdm = tqdm_passthrough
def present_func_output_for_args(cove, func, conv):
output = cove(func, conv)
case_header = f"Output for case {func.__name__}, {conv.__name__}:"
print(case_header)
print("=" * len(case_header))
print()
print(output)
print()
def main():
account_id = sys.argv[1]
disable_tqdm()
run_cove_in_account = update_wrapper(
partial(run_cove_and_return_result_or_exception, account_id),
run_cove_and_return_result_or_exception
)
cases = product(
(normal_fail, fancy_fail),
(dataclass_converter, identity)
)
for func, conv in cases:
present_func_output_for_args(run_cove_in_account, func, conv)
if __name__ == "__main__":
main()
$ python demo.py 111111111111
Output for case normal_fail, dataclass_converter:
=================================================
{'FailedAssumeRole': [], 'Results': [], 'Exceptions': [{'Id': '111111111111', 'Arn': 'arn:aws:organizations::222222222222:account/o-aaaaaaaaaa/111111111111', 'Email': '[email protected]', 'Name': 'Target 1', 'Status': 'ACTIVE', 'AssumeRoleSuccess': True, 'RoleSessionName': 'OrganizationAccountAccessRole', 'ExceptionDetails': NormalException('normal error')}]}
Output for case normal_fail, identity:
======================================
{'Exceptions': [CoveSessionInformation(Id='111111111111', Arn='arn:aws:organizations::222222222222:account/o-aaaaaaaaaa/111111111111', Email='[email protected]', Name='Target 1', Status='ACTIVE', AssumeRoleSuccess=True, RoleSessionName='OrganizationAccountAccessRole', Policy=None, PolicyArns=None, Result=None, ExceptionDetails=NormalException('normal error'))],
'FailedAssumeRole': [],
'Results': []}
Output for case fancy_fail, dataclass_converter:
================================================
Traceback (most recent call last):
File "demo.py", line 36, in run_cove_and_return_result_or_exception
result = cove(func, target_ids=[target_account_id])()
File "/home/isme/.local/share/virtualenvs/faa89da738127ef/lib/python3.8/site-packages/botocove/cove_decorator.py", line 61, in wrapper
Exceptions=[dataclass_converter(e) for e in output["Exceptions"]],
File "/home/isme/.local/share/virtualenvs/faa89da738127ef/lib/python3.8/site-packages/botocove/cove_decorator.py", line 61, in <listcomp>
Exceptions=[dataclass_converter(e) for e in output["Exceptions"]],
File "/home/isme/.local/share/virtualenvs/faa89da738127ef/lib/python3.8/site-packages/botocove/cove_decorator.py", line 17, in dataclass_converter
return {k: v for k, v in asdict(d).items() if v}
File "/usr/lib/python3.8/dataclasses.py", line 1073, in asdict
return _asdict_inner(obj, dict_factory)
File "/usr/lib/python3.8/dataclasses.py", line 1080, in _asdict_inner
value = _asdict_inner(getattr(obj, f.name), dict_factory)
File "/usr/lib/python3.8/dataclasses.py", line 1114, in _asdict_inner
return copy.deepcopy(obj)
File "/usr/lib/python3.8/copy.py", line 172, in deepcopy
y = _reconstruct(x, memo, *rv)
File "/usr/lib/python3.8/copy.py", line 264, in _reconstruct
y = func(*args)
TypeError: __init__() takes 1 positional argument but 2 were given
Output for case fancy_fail, identity:
=====================================
{'Exceptions': [CoveSessionInformation(Id='111111111111', Arn='arn:aws:organizations::222222222222:account/o-aaaaaaaaaa/111111111111', Email='[email protected]', Name='Target 1', Status='ACTIVE', AssumeRoleSuccess=True, RoleSessionName='OrganizationAccountAccessRole', Policy=None, PolicyArns=None, Result=None, ExceptionDetails=FancyException("Problem with the fancy_thing='fancy error'"))],
'FailedAssumeRole': [],
'Results': []}