Skip to content

CosmoTech_Acceleration_Library.Accelerators.cosmo_api

cosmo_api

get_current_scenario_data()

Uses environment vars to find the current scenario data from the cosmotech api :return: a dict containing the data of the scenario from the API or None in another context

Source code in CosmoTech_Acceleration_Library/Accelerators/cosmo_api.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def get_current_scenario_data():
    """
    Uses environment vars to find the current scenario data from the cosmotech api
    :return: a dict containing the data of the scenario from the API or None in another context
    """
    organization_id = os.environ.get("CSM_ORGANIZATION_ID")
    workspace_id = os.environ.get("CSM_WORKSPACE_ID")
    scenario_id = os.environ.get("CSM_SCENARIO_ID")

    if not all([organization_id, workspace_id, scenario_id]):
        return None

    with cosmotech_api.ApiClient(__get_configuration()) as api_client:
        api_instance = ScenarioApi(api_client)
        scenario_data = api_instance.find_scenario_by_id(organization_id=organization_id,
                                                         workspace_id=workspace_id,
                                                         scenario_id=scenario_id)
    return scenario_data

send_dataframe_to_api(dataframe, file_name)

Send a dataframe to the API

Source code in CosmoTech_Acceleration_Library/Accelerators/cosmo_api.py 
44
45
46
47
48
49
50
def send_dataframe_to_api(dataframe, file_name: str):
    """Send a dataframe to the API"""
    file_content = io.StringIO()
    dataframe.to_csv(file_content, index=False)
    file_content.seek(0)
    file_content.name = file_name.split('/')[-1]
    send_file_to_api(file_content, file_name)

send_file_to_api(file_content, file_name)

Send a file to the api

Source code in CosmoTech_Acceleration_Library/Accelerators/cosmo_api.py 
30
31
32
33
34
35
36
37
38
39
40
41
def send_file_to_api(file_content, file_name: str):
    """Send a file to the api"""
    organization_id = os.environ.get("CSM_ORGANIZATION_ID")
    workspace_id = os.environ.get("CSM_WORKSPACE_ID")

    with cosmotech_api.ApiClient(__get_configuration()) as api_client:
        api_ws = WorkspaceApi(api_client)
        api_ws.upload_workspace_file(organization_id=organization_id,
                                     workspace_id=workspace_id,
                                     file=file_content,
                                     overwrite=True,
                                     destination=file_name)