python script to generate/update cost catagories for unit group and buissness unit
requirements
authentication
main.py [csv file path] [unit group cc name] [bu cc name]
| from csv import reader | |
| from os import getenv | |
| from sys import argv | |
| from re import sub | |
| import json | |
| from requests import put, post, get | |
| class AWSAccount: | |
| def __init__( | |
| self, | |
| payer: str, | |
| name: str, | |
| number: str, | |
| unit_group: str, | |
| unit_group_owner: str, | |
| bu: str, | |
| ): | |
| self.payer = payer | |
| self.name = name | |
| self.number = number | |
| self.unit_group = unit_group | |
| self.unit_group_owner = unit_group_owner | |
| self.bu = bu | |
| def unformat(guid: str) -> str: | |
| return guid.replace("-", "") | |
| def get_all_cc() -> list: | |
| # get all the cc in an account | |
| resp = get( | |
| "https://app.harness.io/ccm/api/business-mapping", | |
| params={ | |
| "accountIdentifier": getenv("HARNESS_ACCOUNT_ID"), | |
| }, | |
| headers={ | |
| "Content-Type": "application/json", | |
| "x-api-key": getenv("HARNESS_PLATFORM_API_KEY"), | |
| }, | |
| ) | |
| resp.raise_for_status() | |
| return resp.json().get("resource", {}).get("businessMappings", []) | |
| def get_cc(name: str) -> dict: | |
| # get a cc from the name, case sensitive | |
| return [x for x in get_all_cc() if x["name"] == name].pop() | |
| def update_cc(payload: dict): | |
| # update a cc | |
| resp = put( | |
| "https://app.harness.io/gateway/ccm/api/business-mapping", | |
| params={ | |
| "accountIdentifier": getenv("HARNESS_ACCOUNT_ID"), | |
| # "routingId": getenv("HARNESS_ACCOUNT_ID"), | |
| }, | |
| headers={ | |
| "Content-Type": "application/json", | |
| "x-api-key": getenv("HARNESS_PLATFORM_API_KEY"), | |
| }, | |
| json=payload, | |
| ) | |
| resp.raise_for_status() | |
| return resp.json() | |
| if __name__ == "__main__": | |
| if len(argv) < 4: | |
| print("usage: main.py [csv file path] [unit group cc name] [bu cc name]") | |
| exit(1) | |
| filename = argv[1] | |
| unit_group_cc_name = argv[2] | |
| bu_cc_name = argv[3] | |
| unit_group_uuid = get_cc(unit_group_cc_name).get("uuid") | |
| unit_group_payload = { | |
| "accountId": getenv("HARNESS_ACCOUNT_ID"), | |
| "name": unit_group_cc_name, | |
| "uuid": unit_group_uuid, | |
| "costTargets": [], | |
| "unallocatedCost": { | |
| "strategy": "DISPLAY_NAME", | |
| "label": "Unattributed", | |
| "sharingStrategy": None, | |
| "splits": None, | |
| }, | |
| "dataSources": ["AWS"], | |
| } | |
| bu_uuid = get_cc(bu_cc_name).get("uuid") | |
| bu_payload = { | |
| "accountId": getenv("HARNESS_ACCOUNT_ID"), | |
| "name": bu_cc_name, | |
| "uuid": bu_uuid, | |
| "costTargets": [], | |
| "sharedCosts": [], | |
| "unallocatedCost": { | |
| "strategy": "DISPLAY_NAME", | |
| "label": "Unattributed", | |
| "sharingStrategy": None, | |
| "splits": None, | |
| }, | |
| "dataSources": ["AWS"], | |
| } | |
| accounts = [] | |
| with open(filename, "r") as cc_data: | |
| datareader = reader(cc_data) | |
| next(datareader) | |
| for row in datareader: | |
| # create instance of account with given data | |
| new = AWSAccount( | |
| unformat(row[1]), | |
| row[5], | |
| unformat(row[3]), | |
| row[6], | |
| row[7], | |
| sub(r"[^a-zA-Z0-9() ]", " ", row[8]), | |
| ) | |
| # unit group: find existing instance of cc in payload or create new | |
| if to_update := [ | |
| unit_group_payload["costTargets"].index(x) | |
| for x in unit_group_payload["costTargets"] | |
| if x["name"] == new.unit_group | |
| ]: | |
| unit_group_payload["costTargets"][to_update.pop()]["rules"][0][ | |
| "viewConditions" | |
| ][0]["values"].append(new.number) | |
| else: | |
| unit_group_payload["costTargets"].append( | |
| { | |
| "name": new.unit_group, | |
| "rules": [ | |
| { | |
| "viewConditions": [ | |
| { | |
| "type": "VIEW_ID_CONDITION", | |
| "viewField": { | |
| "fieldId": "awsUsageaccountid", | |
| "fieldName": "Account", | |
| "identifier": "AWS", | |
| "identifierName": "AWS", | |
| }, | |
| "viewOperator": "IN", | |
| "values": [new.number], | |
| } | |
| ] | |
| } | |
| ], | |
| } | |
| ) | |
| # bu: find existing instance of cc in payload or create new | |
| if to_update := [ | |
| bu_payload["costTargets"].index(x) | |
| for x in bu_payload["costTargets"] | |
| if x["name"] == new.bu | |
| ]: | |
| bu_payload["costTargets"][to_update.pop()]["rules"][0][ | |
| "viewConditions" | |
| ][0]["values"].append(new.number) | |
| else: | |
| bu_payload["costTargets"].append( | |
| { | |
| "name": new.bu, | |
| "rules": [ | |
| { | |
| "viewConditions": [ | |
| { | |
| "type": "VIEW_ID_CONDITION", | |
| "viewField": { | |
| "fieldId": "awsUsageaccountid", | |
| "fieldName": "Account", | |
| "identifier": "AWS", | |
| "identifierName": "AWS", | |
| }, | |
| "viewOperator": "IN", | |
| "values": [new.number], | |
| } | |
| ] | |
| } | |
| ], | |
| } | |
| ) | |
| print( | |
| update_cc(unit_group_payload).get( | |
| "resource", "unknown return status for unit group" | |
| ) | |
| ) | |
| print(update_cc(bu_payload).get("resource", "unknown return status for unit group")) |
| requests |