Skip to content

Carol API Module

check_task_status(task_id, task_name='')

Check the task status until it complete with success, fail or if is canceled.

Parameters:

Name Type Description Default
task_id str

The task id.

 required
task_name str

The task name. Defaults to "".

 ''

Returns:

Type Description
Dict[str, str]

Dict[str, str]: A dict with the task_status and the detail.
Source code in carol_app_monitor/carol_api.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
def check_task_status(task_id: str, task_name: str = "") -> Dict[str, str]:
    """Check the task status until it complete with success, fail or if is\
    canceled.

    Args:
        task_id (str): The task id.
        task_name (str, optional): The task name. Defaults to "".

    Returns:
        Dict[str, str]: A dict with the `task_status` and the `detail`.
    """
    if task_name:
        task_string = f"{task_name}/{task_id}"
    else:
        task_string = task_id

    while get_task_status(task_id) in ["READY", "RUNNING"]:
        time.sleep(round(12 * random.random() * 5, 2))

    task_status = get_task_status(task_id)

    if task_status == "COMPLETED":
        detail = f"Task has successfully completed: {task_string}"

        return {"task_status": task_status, "detail": detail}
    elif task_status == "FAILED":
        detail = f"Something went wrong while processing: {task_string}"

        return {"task_status": task_status, "detail": detail}
    elif task_status == "CANCELED":
        detail = f"The task has been CANCELED: {task_string}"

        return {"task_status": task_status, "detail": detail}

get_all_carol_app_names(properties)

Get all CarolApp names existing in tenant.

Parameters:

Name Type Description Default
properties Dict[str, Dict[str, Any]]

A dict with properties from CarolApps.

 required

Returns:

Type Description
List[str]

List[str]: A list with all CarolApp names.
Source code in carol_app_monitor/carol_api.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def get_all_carol_app_names(
    properties: Dict[str, Dict[str, Any]]
) -> List[str]:
    """Get all CarolApp names existing in tenant.

    Args:
        properties (Dict[str, Dict[str, Any]]): A dict with properties from\
            CarolApps.

    Returns:
        List[str]: A list with all CarolApp names.
    """
    carol_app_names = list(properties.keys())
    return carol_app_names

get_entity_type(app_name)

Get entity type from CarolApp by app_name.

Parameters:

Name Type Description Default
app_name str | List[str]

CarolApp name. Can be str or a list.

 required

Returns:

Type Description
List[Dict[str, str]]

List[Dict[str, str | None]]: A list of dicts with app_name and the entity_type: BATCH, ONLINE, or None if app doesn't have process infos.
Notes

The entity_type can be: * mdmTenantAppAIProcess: batch * mdmTenantAppAIProcessWorking: online

Source code in carol_app_monitor/carol_api.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def get_entity_type(app_name: str | List[str]) -> List[Dict[str, str]]:
    """Get entity type from CarolApp by `app_name`.

    Args:
        app_name (str | List[str]): CarolApp name. Can be `str` or a `list`.

    Returns:
        List[Dict[str, str | None]]: A list of dicts with `app_name` and the\
            `entity_type`: `BATCH`, `ONLINE`, or `None` if app\
                doesn't have process infos.

    Notes:
        The `entity_type` can be:
        * mdmTenantAppAIProcess: batch
        * mdmTenantAppAIProcessWorking: online
    """
    if isinstance(app_name, list):
        _process_info = [apps.get_processes_info(app) for app in app_name]
        entity_types = [
            process.get("mdmTenantAppAIProcessValues")[0].get(
                "mdmAIProcessType"
            )
            if process != {}
            else None
            for process in _process_info
        ]

        return [
            {"app_name": app, "entity_type": entity_type}
            for app, entity_type in zip(app_name, entity_types)
        ]
    else:
        _process_info = apps.get_processes_info(app_name)
        entity_type = (
            _process_info.get("mdmTenantAppAIProcessValues")[0].get(
                "mdmAIProcessType"
            )
            if _process_info != {}
            else None
        )

        return [{"app_name": app_name, "entity_type": entity_type}]

get_process_name(app_name)

Get process name.

Parameters:

Name Type Description Default
app_name str | List[str]

CarolApp name. Can be str or a list.

 required

Returns:

Type Description
List[Dict[str, str | None]]

List[Dict[str, str]]: A list of dicts with app_name and the process_name or None if app doesn't have process infos.
Source code in carol_app_monitor/carol_api.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def get_process_name(app_name: str | List[str]) -> List[Dict[str, str | None]]:
    """Get process name.

    Args:
        app_name (str | List[str]): CarolApp name. Can be `str` or a `list`.

    Returns:
        List[Dict[str, str]]: A list of dicts with `app_name` and the\
            `process_name` or `None` if app doesn't have process infos.
    """
    if isinstance(app_name, list):
        _process_info = [apps.get_processes_info(app) for app in app_name]
        process_names = [
            process.get("mdmTenantAppAIProcessValues")[0].get("mdmName")
            if process != {}
            else None
            for process in _process_info
        ]

        return [
            {"app_name": app, "process_name": process_name}
            for app, process_name in zip(app_name, process_names)
        ]
    else:
        _process_info = apps.get_processes_info(app_name)
        process_name = (
            _process_info.get("mdmTenantAppAIProcessValues")[0].get("mdmName")
            if _process_info != {}
            else None
        )

        return [{"app_name": app_name, "process_name": process_name}]

get_process_status(app_name)

Get CarolApp process status by CarolApp name.

Parameters:

Name Type Description Default
app_name str | List[str]

CarolApp name. Can be str or a list.

 required

Returns:

Type Description
List[Dict[str, str | None]]

List[Dict[str, str]]: A list of dicts containing the app_name and your process_status or None if app doesn't have process infos.
Source code in carol_app_monitor/carol_api.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def get_process_status(
    app_name: str | List[str],
) -> List[Dict[str, str | None]]:
    """Get CarolApp process status by CarolApp name.

    Args:
        app_name (str | List[str]): CarolApp name. Can be `str` or a `list`.

    Returns:
        List[Dict[str, str]]: A list of dicts containing the `app_name` and\
            your `process_status` or `None` if app doesn't have process infos.
    """
    if isinstance(app_name, list):
        _process_info = [apps.get_processes_info(app) for app in app_name]
        status = [
            process.get("mdmTenantAppAIProcessValues")[0].get(
                "mdmRunningState"
            )
            if process != {}
            else None
            for process in _process_info
        ]

        return [
            {"app_name": app, "process_status": status}
            for app, status in zip(app_name, status)
        ]

    else:
        _process_info = apps.get_processes_info(app_name)
        status = (
            _process_info.get("mdmTenantAppAIProcessValues")[0].get(
                "mdmRunningState"
            )
            if _process_info != {}
            else None
        )

        return [{"app_name": app_name, "process_status": status}]

get_properties_from_app()

Get properties from all existing CarolApps in tenant ordered by date created.

Returns:

Type Description
Dict[str, Dict[str, Any]]

Dict[str, Dict[str, Any]]: A dict with all properties from all existing CarolApps.
Source code in carol_app_monitor/carol_api.py 
17
18
19
20
21
22
23
24
25
def get_properties_from_app() -> Dict[str, Dict[str, Any]]:
    """Get properties from all existing CarolApps in tenant ordered by date\
        created.

    Returns:
        Dict[str, Dict[str, Any]]: A dict with all properties from all\
            existing CarolApps.
    """
    return apps.all(entity_space="PRODUCTION", sort_by="mdmCreated")

get_task_status(task_id)

Check task status by task_id.

Parameters:

Name Type Description Default
task_id str

The TaskID.

 required

Returns:

Name Type Description
str str

The task status. Can be: READY, RUNNING, COMPLETED, FAILED or CANCELED.
Source code in carol_app_monitor/carol_api.py 
146
147
148
149
150
151
152
153
154
155
156
def get_task_status(task_id: str) -> str:
    """Check task status by `task_id`.

    Args:
        task_id (str): The TaskID.

    Returns:
        str: The task status. Can be: READY, RUNNING, COMPLETED, FAILED or\
            CANCELED.
    """
    return tasks.get_task(task_id).task_status

start_app_process(app_info)

Start a CarolApp process.

Parameters:

Name Type Description Default
app_info Dict[str, str]

A dict containing the app_name and the process_name.

 required

Raises:

Type Description
TaskError

If task fail, raises: Something wrong while processing the task.

Returns:

Type Description
Dict[str, str]

Dict[str, str]: A dict with the task_id and the success status.
Source code in carol_app_monitor/carol_api.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def start_app_process(app_info: Dict[str, str]) -> Dict[str, str]:
    """Start a CarolApp process.

    Args:
        app_info (Dict[str, str]): A dict containing the `app_name` and the\
            `process_name`.

    Raises:
        TaskError: If task fail, raises: Something wrong while processing the\
            task.

    Returns:
        Dict[str, str]: A dict with the `task_id` and the success status.
    """
    task_information = apps.start_app_process(
        app_info.get("process_name"), app_info.get("app_name")
    )

    task_id = task_information.get("data").get("mdmId")
    task_succeded = task_information.get("success")

    if not task_succeded:
        raise TaskError(task_id)

    return {"task_id": task_id, "success": task_succeded}