Code Example - Model Copier

import requests
from akumen_api import progress
import akumen_api
import time

def akumen(src_akumen_url, src_api_token, src_model, dest_model, update_model_files, base_dest_study_name, overwrite_existing_scenario_inputs, **kwargs):
    """
    !! This akumen() function must exist in the execution file!

    Parameters:
        !! These lines define parameters, and a line must exist per input (or output).

        - Input: src_akumen_url [string]
        - Input: src_api_token [string]
        - Input: src_model [string]
        - Input: dest_model [string]
        - Input: update_model_files [boolean]
        - Input: base_dest_study_name [string]
        - Input: overwrite_existing_scenario_inputs [boolean]

    """
    if src_api_token is None or src_api_token == '':
        # No api token passed in, so we assume we're using the existing tenancy
        src_api_token = akumen_api.API_KEY
        
    if src_akumen_url is None or src_akumen_url == '':
        # No url is passed in, so we assume we're using the existing tenancy
        src_akumen_url = akumen_api.AKUMEN_URL
        
    src_headers = { 'Authorization': src_api_token }
    dest_headers = { 'Authorization': akumen_api.API_KEY }

    # --- Get a copy of the source and destination models.  If either don't exist, abort operation and throw an Exception

    # Validate the source url
    response = requests.get(f"{src_akumen_url}/api/v1/models/{src_model}", headers=src_headers, json={ 'include_studies': True, 'include_parameters': False })
    # --- Call raise for status after every requests call to stop execution, otherwise the model will keep running, but the response will have error data in it, rather than real data
    response.raise_for_status()
    
    src_model_json = response.json()
    
    progress('Validated source url')
    
    # Validate the destination url, the model must exist
    response = requests.get(f"{akumen_api.AKUMEN_API_URL}models/{dest_model}", headers=dest_headers, json={ 'include_studies': True, 'include_parameters': False })
    response.raise_for_status()
    
    dest_model_json = response.json()
    
    progress('Validated destination model')
    
    if update_model_files:
        progress('Commencing update model files')
        # --- Gets a copy of the file names
        response = requests.get(f"{src_akumen_url}/api/v1/modelfiles/{src_model}/files", headers=src_headers)
        response.raise_for_status()
        model_files = response.json()
        for model_file_name in model_files:
            # --- Gets the individual file, including the code
            response = requests.get(f"{src_akumen_url}/api/v1/modelfiles/{src_model}/files/{model_file_name['filepath']}", headers=src_headers)
            response.raise_for_status()
            file = response.json()
            # --- Posts the file to the destination
            response = requests.post(f"{akumen_api.AKUMEN_API_URL}modelfiles/{dest_model}/files", headers=dest_headers, json=file)
            response.raise_for_status()
            progress(f"Finished writing model file: {file['filepath']}")
        
        progress('Finished update model files')
        
    # --- Loops through each study in the order they appear in the source
    for study in sorted(src_model_json['studies'], key=lambda study:study['ordinal']):
        progress(f"Commencing study {study['name']}")
        # First clone the destination base study if it doesn't already exist - if it does, we'll reuse it and just merge in the scenarios
        # --- Check if the study exists by name
        existing_study = next((x for x in dest_model_json['studies'] if x['name'] == study['name']), None)

        if existing_study is None:
            # --- Existing study does not exist, so create a new one
            existing_base_study = next((x for x in dest_model_json['studies'] if x['name'] == base_dest_study_name), None)
            if existing_base_study is None:
                raise Exception('Could not find baseline study in destination to create a clone from')
                
            # --- Clone the study defined in the inputs
            response = requests.post(f"{akumen_api.AKUMEN_API_URL}models/{dest_model}/{existing_base_study['name']}/clone", headers=dest_headers, json=dest_model_json['parameters'])
            response.raise_for_status()
            
            # --- Rename the cloned study to match that coming in from the source
            new_study = response.json()
            response = requests.post(f"{akumen_api.AKUMEN_API_URL}models/{dest_model}/{new_study['name']}/rename", headers=dest_headers, json={ 'new_name': study['name'], 'description': study['description'] })
            response.raise_for_status()
            new_study['name'] = study['name']
            existing_study = new_study
            
        # --- Loops through the scenarios in the order they appear in the study
        for scenario in sorted(study['scenarios'], key=lambda scenario: scenario['ordinal']):
            progress(f"Commencing scenario: {scenario['name']}")
            # We don't want to overwrite any scenarios, only add new ones, so if it exists, ignore it
            existing_scenario = next((x for x in existing_study['scenarios'] if x['name'] == scenario['name']), None)
            
            has_existing_scenario = existing_scenario is not None
            
            if existing_scenario is None:
                # --- Clone the baseline scenario from the existing study - we'll use this to update the inputs (if any)
                response = requests.post(f"{akumen_api.AKUMEN_API_URL}models/{dest_model}/{existing_study['name']}/{existing_study['baseline_scenario_name']}/clone", headers=dest_headers, json=dest_model_json['parameters'])
                response.raise_for_status()
                new_scenario = response.json()
                # --- Rename the cloned scenario
                response = requests.post(f"{akumen_api.AKUMEN_API_URL}models/{dest_model}/{existing_study['name']}/{new_scenario['name']}/rename", headers=dest_headers, json={ 'new_name': scenario['name'], 'description': scenario['description'] })
                response.raise_for_status()
                new_scenario['name'] = scenario['name']
                existing_scenario = new_scenario
                
            if has_existing_scenario and not overwrite_existing_scenario_inputs:
                # --- The scenario already exists and we don't want to overwrite, so skip it
                progress(f"Scenario: {scenario['name']} skipped")
                continue

            # --- Fetch the inputs from the source scenario
            response = requests.get(f"{src_akumen_url}/api/v1/models/{src_model}/{study['name']}/{scenario['name']}/input_parameters", headers=src_headers)
            response.raise_for_status()
            src_parameters = response.json()
            # --- Post the scenarios from the source to the destination.  Note that we can just post them as is, because everything is referenced by name, Akumen will attach the inputs to the new scenario
            response = requests.post(f"{akumen_api.AKUMEN_API_URL}models/{dest_model}/{existing_study['name']}/{existing_scenario['name']}/SaveInputParameters", headers=dest_headers, json={ 'input_parameters': src_parameters })
            response.raise_for_status()
            
            # This can be a little expensive, so give Akumen some time to catchup with the cloning process
            time.sleep(0.1)
            progress(f"Finished scenario: {scenario['name']}")
            
        progress(f"Finished study: {study['name']}")

    # The akumen() function must return a dictionary including keys relating to outputs.
    return {

    }


if __name__ == '__main__':
    """
    Any local test code can be used in an import guard
    when developing on a local machine, you can put code here that won't
    get run by Akumen.
    """
    print('Running local tests...')
    assert (akumen(1, 2)['first_result'] == 3)
    assert (akumen(3, 4)['second_result'] == -1)
    print('Tests completed!')