def main()

in src/access-analyzer/main.py [0:0]

• 37 lines of code
• 8 McCabe index (conditional complexity)

def main(self):
    client = bigquery.Client()
    sa_keys = get_keys(client)
    projects = set()
    key_dict = {}
    access_data = []
    for row in sa_keys:
        key_dict[row["key"]] = {
            "project": row["project_id"],
            "principalName": row["principal_name"],
            "keyId": row["key"],
            "keyCreationTime": str(row["valid_after_time"]),
            "keyLastUse": None,
            "requestTime": str(row["request_time"]),
            "recommenderSubtype": None,
            "recommenderDescription": None,
            "recommenderPriority": None,
            "recommenderRevokedIamPermissionsCount": None,
            "associatedRecommendation": None,
        }
        projects.add(row["project_id"])
    for project in projects:
        analysis_data = analyze.get_policy_analyzer_project(project)
        if analysis_data:
            for data in analysis_data:
                if data["keyId"] in key_dict:
                    key_dict[data["keyId"]].update(data)
    access_data = list(key_dict.values())
    if not access_data:
        print("No new rows to add.")
        return {}

    error = client.insert_rows_json(destination_table, access_data)

    if not error:
        print("New rows have been added.")
    else:
        print(f"Encountered errors while inserting rows: {error}")
    return jsonify(error)