in src/services/report_service.py [0:0]
def average_statistics(*iterables: list[StatisticsItem]
) -> Generator[dict, None, None]:
remapped = {} # (policy region) to items
for i in chain(*iterables):
remapped.setdefault((i['policy'], i['region']), []).append(i)
for key, items in remapped.items():
total_api_calls = {}
executions = []
failed_invocations = 0
scanned, failed = [], []
for item in items:
for k, v in (item.get('api_calls') or {}).items():
if k not in total_api_calls:
total_api_calls[k] = v
else:
total_api_calls[k] += v
executions.append(item['end_time'] - item['start_time'])
if item.get('scanned_resources'):
scanned.append(item['scanned_resources'])
if item.get('failed_resources'):
failed.append(item['failed_resources'])
if item.get('error_type'):
failed_invocations += 1
scanned = scanned or [0]
failed = failed or [0]
yield {
'policy': key[0],
'region': key[1],
'invocations': len(items),
'succeeded_invocations': len(items) - failed_invocations,
'failed_invocations': failed_invocations,
'total_api_calls': total_api_calls,
'min_exec': min(executions),
'max_exec': max(executions),
'total_exec': sum(executions),
'average_exec': statistics.mean(executions),
'resources_failed': sum(failed),
'resources_scanned': sum(scanned),
'average_resources_scanned': statistics.mean(scanned),
'average_resources_failed': statistics.mean(failed),
}