Importing and Updating the Total Annual Costs of Applications

Import and update total annual costs using a Python script.

Overview

In this tutorial, you will learn how to import and update the total annual costs of applications using a Python script. Use this tutorial when completing an initial setup of your workspace or updating values in bulk.

Analyzing total annual costs can help identify inefficiencies in your organization's IT spending, enabling optimizations that can significantly reduce expenses. It also provides a clear view of IT investments, facilitating informed strategic decisions about resource allocation and budget planning.

📘

Note

In the standard Meta Model configuration, the costTotalAnnual attribute is stored in the relation between an Application and an IT Component Fact Sheet.

This tutorial assumes you have basic knowledge of:

Before You Start

Before you start the tutorial, do the following:

  • Obtain an API token by creating a Technical User. For more information, see Create a Technical User.
  • Prepare the total annual cost data to be imported.

Importing Total Annual Costs

When you complete an initial setup of your workspace, you can import Fact Sheets and related cost data using a Python script.

Before you start, prepare the data to be imported. The following table contains example data that we use in this tutorial.

Application NameIT Component NameTotal Annual Costs
HR Management AppStorage System125
HR Management AppWeb Server250
Finance Management AppStorage System500
Finance Management AppWeb Server1000

Example CSV file:

application_name,it_component,total_annual_cost
HR Management App,Storage System,125
HR Management App,Web Server,250
Finance Management App,Storage System,500
Finance Management App,Web Server,1000

Once you have your input data ready, run the following Python script. The script completes the following tasks:

  • Perform authentication to LeanIX services
  • Create Application and IT Component Fact Sheets using data from the input file
  • Create relations between Application and IT Component Fact Sheets and update the costTotalAnnual attribute with values from the input file
  • Export the CSV data including the newly created Fact Sheet IDs as a JSON file

Example script:

import csv
import json
import logging
import os

import requests

logging.basicConfig(level=logging.INFO)

CSV_FILE = os.getenv('CSV_FILE')

LEANIX_API_TOKEN = os.getenv('LEANIX_API_TOKEN')
LEANIX_SUBDOMAIN = os.getenv('LEANIX_SUBDOMAIN')
LEANIX_GRAPHQL_URL = f'https://{LEANIX_SUBDOMAIN}.leanix.net/services/pathfinder/v1/graphql'
LEANIX_OAUTH2_URL = f'https://{LEANIX_SUBDOMAIN}.leanix.net/services/mtm/v1/oauth2/token'
TIMEOUT = 20
IT_COMPOMENT_MUTATION = """
mutation ITCompomentMutation($input: BaseFactSheetInput!) {
    createFactSheet(input: $input) {
        factSheet {
            id
            name
            type
        }
    }
}
"""
APPLICATION_MUTATION = """
mutation ApplicationMutation($input: BaseFactSheetInput!, $patches: [Patch!]) {
    createFactSheet(input: $input, patches: $patches) {
        factSheet {
            id
            name
            type
        }
    }
}
"""

def _obtain_access_token():
    """Obtain a LeanIX Access token using the Technical User generated
    API secret.

    Returns
    -------
        Optional(str): The LeanIX OAuth2 Access Token
    
    """
    if not LEANIX_API_TOKEN:
        raise Exception('A valid token is required')
    response = requests.post(
        LEANIX_OAUTH2_URL,
        auth=('apitoken', LEANIX_API_TOKEN),
        data={'grant_type': 'client_credentials'},
        timeout=TIMEOUT
    )
    response.raise_for_status()
    return response.json().get('access_token')

def make_request(payload: dict):
    """Perform a GraphQL request to the LeanIX GraphQL API endpoint.

    Args:
    ----
        payload (dict): The query or the mutation to perform against the API endpoint.
    
    Returns:
    -------
        Optional(dict): The GraphQL response.
    
    """
    # Fetch the access token and set the Authorization Header
    access_token = _obtain_access_token()
    auth_header = f'Bearer {access_token}'
    # Provide the headers
    headers = {
        'Authorization': auth_header,
    }
    
    response = requests.post(
        LEANIX_GRAPHQL_URL,
        json=payload,
        headers=headers,
        timeout=TIMEOUT
    )
    response.raise_for_status()
    json_response = response.json()
    # GraphQL always returns a 200 response even if errors are included
    # as such we check if `errors` is not empty.
    errors=json_response.get('errors', [])
    if len(errors):
        raise Exception(f'Request {payload} to {LEANIX_GRAPHQL_URL} was not successful: {errors}')
    return json_response

def create_it_component(it_component: str):
    """Create a LeanIX IT component.
    
    Args:
    ----
        it_component (str): An IT component name.
    
    Returns:
    -------
        str: The UUID of the generated IT Component Fact Sheet.
    
    """
    # Create the Fact Sheet and return the `id` in order to assign it to
    # the application.
    # Mind the extra indendation as we will nest the creation of Fact Sheets
    # under one mutation.
    mutation_variables = {
        'input': {
            'name': it_component,
            'type': 'ITComponent'
        }
    }
    logging.info(f'Creating ITComponent {it_component}')
    response = make_request({'query': IT_COMPOMENT_MUTATION, 'variables': mutation_variables})
    # Response contains the IDs of the ITComponents
    it_component_id = response.get('data', {}).get('createFactSheet',{}).get('factSheet', {}).get('id')
    logging.info(f'Succesfully created IT Component: {it_component} with id: {it_component_id}')
    return it_component_id


def create_applications(applications: dict):
    """Create LeanIX Application FactSheets from the provided set of data.

    Args:
    ----
        applications (dict): A dictionary of Applications with the relevant IT Components
    
    Returns:
    -------
        dict: A dictionary with the Application details including the Application
        Fact Sheet UUID.

    """
    for application in applications.values():
        mutation_variables = {
            'input': {
                'name': application.get('name'),
                'type': 'Application'
            },
            'patches': [
                {
                    'op': 'add',
                    'path': '/relApplicationToITComponent/new_1',
                    'value': '{\"factSheetId\": \"%s\", \"costTotalAnnual\": %s}'%(application.get('it_component'), application.get('total_annual_cost'))
                }
            ]
        }
        logging.info(f'Creating Application {application}')
        response = make_request({'query': APPLICATION_MUTATION, 'variables': mutation_variables})
        application['id'] = response.get('data', {}).get('createFactSheet',{}).get('factSheet', {}).get('id')
        logging.info(f'Succesfully created application: {application.get('name')} with id: {application.get('id')}')

    return applications
    

def main(csv_file_path: str):
    """Read a CSV file containing LeanIX Applications and IT components
    and generate the relevant Application FactSheets and IT Components.

    Args:
    ----
        csv_file_path (str): The path to the CSV file containing the Application information
    
    """
    with open(csv_file_path, newline='') as csvfile:
        reader = csv.DictReader(csvfile)
        applications = dict()
        it_components = dict()
        for row in reader:
            application_name = row.get('application_name')
            it_component = row.get('it_component')
            # Validate required fields
            if not application_name or not application_name.strip():
                raise ValueError('Application name is missing or empty.')
            if not it_component or not it_component.strip():
                raise ValueError('IT component name is missing or empty.')
            # Create the IT components in order to fetch the relevant ID's
            if it_component not in it_components:
                it_components[it_component] = create_it_component(it_component)
            # Crete the relevant Application pairs based on the IT Component IDs
            # Create the ID field with a null value as a placeholder for the response
            if application_name not in applications:
                applications[application_name] = {
                    'name': application_name,
                    'total_annual_cost': row.get('total_annual_cost', 0),
                    'it_component': it_components.get(it_component),
                    'id': None
                }
        # Create the application if everything was successful
        if applications:
            applications = create_applications(applications)
            # Create a JSON export of the data with the Fact Sheet information
            with open('export.json', 'w') as f:
                # Use the json.dump method to write the data to the file
                json.dump(applications, f, indent=4)  # indent parameter for readability
            

if __name__ == '__main__':
    main(CSV_FILE)

Updating Total Annual Costs

To update total annual costs, export the current values using GraphQL and then import new values using a Python script.

Step 1: Retrieve Cost Values

To retrieve total annual costs, use the following GraphQL query. You can run the query in the GraphiQL tool in your workspace.

Example query:

{
  allFactSheets(factSheetType: Application) {
    edges {
      node {
        ... on Application {
          name
          id
          relApplicationToITComponent {
            edges {
              node {
                factSheet {
                  name
                  id
                }
                costTotalAnnual
              }
            }
          }
        }
      }
    }
  }
}

Example response:

{
  "data": {
    "allFactSheets": {
      "edges": [
        {
          "node": {
            "name": "HR Management App",
            "id": "4d121f64-116b-4ccc-a292-eb4e4f8d1b24",
            "relApplicationToITComponent": {
              "edges": [
                {
                  "node": {
                    "factSheet": {
                      "name": "Storage System",
                      "id": "a8fe4825-42b8-431b-8124-ca12c579c78b"
                    },
                    "costTotalAnnual": 125
                  }
                },
                {
                  "node": {
                    "factSheet": {
                      "name": "Web Server",
                      "id": "ed46809c-998a-4fd6-9185-4b25e4e77d9b"
                    },
                    "costTotalAnnual": 250
                  }
                }
              ]
            }
          }
        },
        {
          "node": {
            "name": "Finance Management App",
            "id": "28fe4aa2-6e46-41a1-a131-72afb3acf256",
            "relApplicationToITComponent": {
              "edges": [
                {
                  "node": {
                    "factSheet": {
                      "name": "Storage System",
                      "id": "a8fe4825-42b8-431b-8124-ca12c579c78b"
                    },
                    "costTotalAnnual": 500
                  }
                },
                {
                  "node": {
                    "factSheet": {
                      "name": "Web Server",
                      "id": "ed46809c-998a-4fd6-9185-4b25e4e77d9b"
                    },
                    "costTotalAnnual": 1000
                  }
                }
              ]
            }
          }
        }
      ]
    }
  }
}

Convert the output in JSON format into a CSV file using your preferred conversion tool. The following table shows the data that we retrieved.

Application NameApplication IDIT Component NameIT Component IDTotal Annual Costs
HR Management App4d121f64-116b-4ccc-a292-eb4e4f8d1b24Storage Systema8fe4825-42b8-431b-8124-ca12c579c78b125
HR Management App4d121f64-116b-4ccc-a292-eb4e4f8d1b24Web Servered46809c-998a-4fd6-9185-4b25e4e77d9b250
Finance Management App28fe4aa2-6e46-41a1-a131-72afb3acf256Storage Systema8fe4825-42b8-431b-8124-ca12c579c78b500
Finance Management App28fe4aa2-6e46-41a1-a131-72afb3acf256Web Servered46809c-998a-4fd6-9185-4b25e4e77d9b1000

Example CSV file:

application_id,it_component_id,total_annual_cost
4d121f64-116b-4ccc-a292-eb4e4f8d1b24,a8fe4825-42b8-431b-8124-ca12c579c78b,125
4d121f64-116b-4ccc-a292-eb4e4f8d1b24,ed46809c-998a-4fd6-9185-4b25e4e77d9b,250
28fe4aa2-6e46-41a1-a131-72afb3acf256,a8fe4825-42b8-431b-8124-ca12c579c78b,500
28fe4aa2-6e46-41a1-a131-72afb3acf256,ed46809c-998a-4fd6-9185-4b25e4e77d9b,1000

Step 2: Run the Script to Update Cost Values

Before proceeding, prepare an input file in CSV format with updated cost values. For details, see the previous step in this tutorial.

Once you have your input data ready, run the following Python script to update cost values. The script completes the following tasks:

  • Perform authentication to LeanIX services
  • Retrieve relations between Application and IT Component Fact Sheets
  • Update the costTotalAnnual attribute with values from the input file

Example script:


import csv
import logging
import os

import requests

logging.basicConfig(level=logging.INFO)

CSV_FILE = os.getenv('CSV_FILE')

LEANIX_API_TOKEN = os.getenv('LEANIX_API_TOKEN')
LEANIX_SUBDOMAIN = os.getenv('LEANIX_SUBDOMAIN')
LEANIX_GRAPHQL_URL = f'https://{LEANIX_SUBDOMAIN}.leanix.net/services/pathfinder/v1/graphql'
LEANIX_OAUTH2_URL = f'https://{LEANIX_SUBDOMAIN}.leanix.net/services/mtm/v1/oauth2/token'
TIMEOUT = 20

QUERY = """
query retrieveAllFactSheets {
    allFactSheets(factSheetType: Application) {
        edges {
            node {
                id
                ... on Application {
                    relApplicationToITComponent {
                        edges {
                            node {
                                id
                                factSheet {
                                    id
                                }
                            }
                        }
                    }
                }
            }
        }
    }
}

"""
APPLICATION_MUTATION = """
mutation UpdateApplication($id: ID!, $patches: [Patch]!) {
    updateFactSheet(id: $id, patches: $patches) {
        factSheet {
            id
            name
            description
            type
        }
    }
}
"""

def _obtain_access_token():
    """Obtain a LeanIX Access token using the Technical User generated
    API secret.

    Returns
    -------
        Optional(str): The LeanIX OAuth2 Access Token
    
    """
    if not LEANIX_API_TOKEN:
        raise Exception('A valid token is required')
    response = requests.post(
        LEANIX_OAUTH2_URL,
        auth=('apitoken', LEANIX_API_TOKEN),
        data={'grant_type': 'client_credentials'},
        timeout=TIMEOUT
    )
    response.raise_for_status()
    return response.json().get('access_token')

def make_request(payload: dict):
    """Perform a GraphQL request to the LeanIX GraphQL API endpoint.

    Args:
    ----
        payload (dict): The query or the mutation to perform against the API endpoint.
    
    Returns:
    -------
        Optional(dict): The GraphQL response.
    
    """
    # Fetch the access token and set the Authorization Header
    access_token = _obtain_access_token()
    auth_header = f'Bearer {access_token}'
    # Provide the headers
    headers = {
        'Authorization': auth_header,
    }
    
    response = requests.post(
        LEANIX_GRAPHQL_URL,
        json=payload,
        headers=headers,
        timeout=TIMEOUT
    )
    response.raise_for_status()
    json_response = response.json()
    # GraphQL always returns a 200 response even if errors are included
    # as such we check if `errors` is not empty.
    errors=json_response.get('errors', [])
    if len(errors):
        raise Exception(f'Request {payload} to {LEANIX_GRAPHQL_URL} was not successful: {errors}')
    return json_response


def _parse_application_fact_sheets(query_response):
    """Loop through the query results generating a dictionary containing the information
    required to update the relevant application costs.

    Args:
    ----
        query_response (dict): The query response data from the GraphQL query

    """
    applications = dict()
    for edge in query_response.get('data',{}).get('allFactSheets',{}).get('edges', []):
        node = edge.get('node', {})
        application_id = node.get('id')
        for relation_edge in node.get('relApplicationToITComponent', {}).get('edges', []):
            # Set an entry only if there is a relation available
            applications[application_id] = dict()
            relation_node = relation_edge.get('node', {})
            relation_id = relation_node.get('id')
            it_component_id = relation_node.get('factSheet',{}).get('id')
            applications[application_id][it_component_id] = relation_id
    return applications

def retrieve_application_fact_sheets():
    """Retrieve the LeanIX Application Fact Sheets, including their relational
    values.
    """
    response = make_request({'query': QUERY})
    return _parse_application_fact_sheets(response)
        
def update_costs(application_id: str, it_component_id: str, relation_id: str, total_annual_costs: int):
    """Update LeanIX Application FactSheets from the provided set of data.

    Args:
    ----
        application_id (str): The UUID of the Application Fact Sheet.
        it_component_id (str): The UUID of the ITComponent Fact Sheet.
        relation_id (str): The UUID of the relation with the ITComponent Fact Sheet.
        total_annual_costs (int): The total annual cost.

    """
    mutation_variables = {
        'id': application_id,
        'patches': [
            {
                'op': 'replace',
                'path': f'/relApplicationToITComponent/{relation_id}',
								'value': '{\"factSheetId\": \"%s\", \"costTotalAnnual\": %s}'%(it_component_id, total_annual_costs)
            }
        ]
    }
    logging.info(f'Updating costs for Application Fact Sheet: {application_id}')
    response = make_request({'query': APPLICATION_MUTATION, 'variables': mutation_variables})
    logging.info(f'Succesfully updated costs: {response}')


def main(csv_file_path: str):
    """Read a CSV file containing LeanIX Applications and IT components
    and generate the relevant Application FactSheets and IT Components.

    Args:
    ----
        csv_file_path (str): The path to the CSV file containing the Application information
    
    """
    logging.info('Fetching Application Fact Sheets with relations')
    applications = retrieve_application_fact_sheets()
    with open(csv_file_path, newline='') as csvfile:
        reader = csv.DictReader(csvfile)
        for row in reader:
            application_id = row.get('application_id')
            it_component_id = row.get('it_component_id')
            relation_id = applications.get(application_id, {}).get(it_component_id, {})
            total_annual_costs = row.get('total_annual_costs')
            # Validate required fields
            if not application_id or not application_id.strip():
                raise ValueError('Application id is missing or empty.')
            if not it_component_id or not it_component_id.strip():
                raise ValueError('IT component id is missing or empty.')
            if not relation_id or not relation_id.strip():
                raise ValueError('Relation id is missing or empty.')
            update_costs(application_id, it_component_id, relation_id, total_annual_costs)

if __name__ == '__main__':
    main(CSV_FILE)

Related Information