def get_all_kx_changesets()

in ManagedkdbInsights/processing_data/managed_kx.py [0:0]

• 29 lines of code
• 9 McCabe index (conditional complexity)

def get_all_kx_changesets(client, databaseName, environmentId:str=None):
    if environmentId is None:
        environmentId = get_kx_environment_id(client)

    # get all changesets
    c_list=list_kx_changesets(client, environmentId=environmentId, databaseName=databaseName)

    # get details on each changeset in the database
    cd_list = []

    for c in c_list:
        c_id = c['changesetId']
        cd = get_kx_changeset(client=client, environmentId=environmentId, databaseName=databaseName, changesetId=c_id)

        cd_list.append(cd)

    # For each changeset, extract/flattend changeRequest
    part_list = []

    for cs in cd_list:
        for cr in cs.get('changeRequests', None):

            # anything?
            if cr is None:
                continue

            item = {}

            # get the core items
            item['status'] = cs['status']
            item['changesetId'] = cs['changesetId']
            item['createdTimestamp'] = cs['createdTimestamp']
            item['activeFromTimestamp'] = cs['activeFromTimestamp']
            item['lastModifiedTimestamp'] = cs['lastModifiedTimestamp']

            # get the request items
            item['changeType'] = cr['changeType']
            item['dbPath'] = cr['dbPath']

            # get the date partition from dbPath
            dt = None
            p = cr.get('dbPath', None)

            # has a specific date to parse
            if p is not None and p[0] == '/' and len(p) == 10:
                dt = datetime.datetime.strptime(p[1:11],"%Y.%m.%d")

            item['date'] = dt

            part_list.append(item)

    return part_list