def get_members()

in enableDetective.py [0:0]

• 21 lines of code
• 12 McCabe index (conditional complexity)

def get_members(d_client: botocore.client.BaseClient, graphs: typing.List[str]) ->\
        (typing.Dict[str, typing.Set[str]], typing.Dict[str, typing.Set[str]]):
    """
    Get member accounts for all behaviour graphs in a region.

    Args:
        - d_client: Detective boto3 client generated from the master session.
        - graphs: List of graphs arns

    Returns:
        Two dictionaries: one with all account ids, other with the ones pending to accept
        the invitation.
    """
    try:
        # itertools.tee creates two independent iterators from a single one. This way
        # we can iterate the iterator twice: one to return all elements and other to return
        # the ones pending to be invited.
        ####
        # check the value of NextToken in the response. if it is non-null, pass it back into a subsequent list_members call (and keep doing this until a null token is returned)
        def _master_memberList(g: str) -> typing.List[typing.Dict]:
            # create a list to append the member accounts
            memberAccounts = []
            # create a dictionary for the nextToken from each call
            tokenTracker = {}
            # loop through list_members call results and take action for each returned result
            while True:
                # list_members of graph "g" and return the first 100 results
                members = d_client.list_members(
                    GraphArn=g, MaxResults=100, **tokenTracker)
                # add the returned list members to the list
                memberAccounts.extend(members['MemberDetails'])
                # if the returned results have a "NextToken" key then use it to query again
                if 'NextToken' in members:
                    tokenTracker['NextToken'] = members['NextToken']
                # if the returned results do not have a "NextToken" key then exit the loop
                else:
                    break
            # return members list.
            # The return statement doesn't need ()
            return memberAccounts
    except Exception as e:
        logging.exception(f'exception when getting memebers: {e}')
        
    # iterate through each list and return results
    all_ac, pending = itertools.tee((g, _master_memberList(g))
                                    for g in graphs)
    
    return ({g: {x['AccountId'] for x in v} for g, v in all_ac},
            {g: {x['AccountId'] for x in v if x['Status'] == 'INVITED'} for g, v in pending})