Skip to content

ChoreService

ChoreService(rest)

Bases: ObjectService

Service to handle Object Updates for TM1 Chores

Source code in TM1py/Services/ChoreService.py 
68
69
def __init__(self, rest: RestService):
    super().__init__(rest)

activate(chore_name, **kwargs)

activate chore on TM1 Server

Parameters:

Name Type Description Default
chore_name str
required

Returns:

Type Description
Response

response
Source code in TM1py/Services/ChoreService.py 
193
194
195
196
197
198
199
def activate(self, chore_name: str, **kwargs) -> Response:
    """activate chore on TM1 Server
    :param chore_name:
    :return: response
    """
    url = format_url("/Chores('{}')/tm1.Activate", chore_name)
    return self._rest.POST(url, "", **kwargs)

create(chore, **kwargs)

create a chore

Parameters:

Name Type Description Default
chore Chore

instance of TM1py.Chore

 required

Returns:

Type Description
Response
Source code in TM1py/Services/ChoreService.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def create(self, chore: Chore, **kwargs) -> Response:
    """create a chore
    :param chore: instance of TM1py.Chore
    :return:
    """
    url = "/Chores"
    response = self._rest.POST(url=url, data=chore.body, **kwargs)

    if chore.dst_sensitivity:
        self.set_local_start_time(chore.name, chore.start_time.datetime)

    if chore.active:
        self.activate(chore.name)
    return response

deactivate(chore_name, **kwargs)

deactivate chore on TM1 Server

Parameters:

Name Type Description Default
chore_name str
required

Returns:

Type Description
Response

response
Source code in TM1py/Services/ChoreService.py 
201
202
203
204
205
206
207
def deactivate(self, chore_name: str, **kwargs) -> Response:
    """deactivate chore on TM1 Server
    :param chore_name:
    :return: response
    """
    url = format_url("/Chores('{}')/tm1.Deactivate", chore_name)
    return self._rest.POST(url, "", **kwargs)

delete(chore_name, **kwargs)

delete chore in TM1

Parameters:

Name Type Description Default
chore_name str
required

Returns:

Type Description
Response

response
Source code in TM1py/Services/ChoreService.py 
111
112
113
114
115
116
117
118
def delete(self, chore_name: str, **kwargs) -> Response:
    """delete chore in TM1
    :param chore_name:
    :return: response
    """
    url = format_url("/Chores('{}')", chore_name)
    response = self._rest.DELETE(url, **kwargs)
    return response

execute_chore(chore_name, **kwargs)

Ask TM1 Server to execute a chore

Parameters:

Name Type Description Default
chore_name str

String, name of the chore to be executed

 required

Returns:

Type Description
Response

the response
Source code in TM1py/Services/ChoreService.py 
225
226
227
228
229
230
def execute_chore(self, chore_name: str, **kwargs) -> Response:
    """Ask TM1 Server to execute a chore
    :param chore_name: String, name of the chore to be executed
    :return: the response
    """
    return self._rest.POST(format_url("/Chores('{}')/tm1.Execute", chore_name), "", **kwargs)

exists(chore_name, **kwargs)

Check if Chore exists

Parameters:

Name Type Description Default
chore_name str
required

Returns:

Type Description
bool
Source code in TM1py/Services/ChoreService.py 
120
121
122
123
124
125
126
127
def exists(self, chore_name: str, **kwargs) -> bool:
    """Check if Chore exists

    :param chore_name:
    :return:
    """
    url = format_url("/Chores('{}')", chore_name)
    return self._exists(url, **kwargs)

get(chore_name, **kwargs)

Get a chore from the TM1 Server

Parameters:

Name Type Description Default
chore_name str
required

Returns:

Type Description
Chore

instance of TM1py.Chore
Source code in TM1py/Services/ChoreService.py 
71
72
73
74
75
76
77
78
def get(self, chore_name: str, **kwargs) -> Chore:
    """Get a chore from the TM1 Server
    :param chore_name:
    :return: instance of TM1py.Chore
    """
    url = format_url("/Chores('{}')?$expand=Tasks($expand=*,Process($select=Name),Chore($select=Name))", chore_name)
    response = self._rest.GET(url, **kwargs)
    return Chore.from_dict(response.json())

get_all(**kwargs)

get a List of all Chores

Returns:

Type Description
List[Chore]

List of TM1py.Chore
Source code in TM1py/Services/ChoreService.py 
80
81
82
83
84
85
86
def get_all(self, **kwargs) -> List[Chore]:
    """get a List of all Chores
    :return: List of TM1py.Chore
    """
    url = "/Chores?$expand=Tasks($expand=*,Process($select=Name),Chore($select=Name))"
    response = self._rest.GET(url, **kwargs)
    return [Chore.from_dict(chore_as_dict) for chore_as_dict in response.json()["value"]]

get_all_names(**kwargs)

get a List of all Chores

Returns:

Type Description
List[str]

List of TM1py.Chore
Source code in TM1py/Services/ChoreService.py 
88
89
90
91
92
93
94
def get_all_names(self, **kwargs) -> List[str]:
    """get a List of all Chores
    :return: List of TM1py.Chore
    """
    url = "/Chores?$select=Name"
    response = self._rest.GET(url, **kwargs)
    return [chore["Name"] for chore in response.json()["value"]]

search_for_parameter_value(parameter_value, **kwargs)

Return chore details for any/all chores that have a specified value set in the chore parameter settings *this will NOT check the process parameter default, rather the defined parameter value saved in the chore

Parameters:

Name Type Description Default
parameter_value str

string, will search wildcard for string in parameter value using Contains(string)

 required
Source code in TM1py/Services/ChoreService.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def search_for_parameter_value(self, parameter_value: str, **kwargs) -> List[Chore]:
    """Return chore details for any/all chores that have a specified value set in the chore parameter settings
        *this will NOT check the process parameter default, rather the defined parameter value saved in the chore

    :param parameter_value: string, will search wildcard for string in parameter value using Contains(string)
    """
    url = format_url(
        "/Chores?"
        "$filter=Tasks/any(t: t/Parameters/any(p: isof(p/Value, Edm.String) and contains(tolower(p/Value), '{}')))"
        "&$expand=Tasks($expand=*,Process($select=Name),Chore($select=Name))",
        parameter_value.lower(),
    )
    response = self._rest.GET(url, **kwargs)
    return [Chore.from_dict(chore_as_dict) for chore_as_dict in response.json()["value"]]

search_for_process_name(process_name, **kwargs)

Return chore details for any/all chores that contain specified process name in tasks

Parameters:

Name Type Description Default
process_name str

string, a valid ti process name; spaces will be elimniated

 required
Source code in TM1py/Services/ChoreService.py 
129
130
131
132
133
134
135
136
137
138
139
140
def search_for_process_name(self, process_name: str, **kwargs) -> List[Chore]:
    """Return chore details for any/all chores that contain specified process name in tasks

    :param process_name: string, a valid ti process name; spaces will be elimniated
    """
    url = format_url(
        "/Chores?$filter=Tasks/any(t: replace(tolower(t/Process/Name), ' ', '') eq '{}')"
        "&$expand=Tasks($expand=*,Chore($select=Name),Process($select=Name))",
        process_name.lower().replace(" ", ""),
    )
    response = self._rest.GET(url, **kwargs)
    return [Chore.from_dict(chore_as_dict) for chore_as_dict in response.json()["value"]]

set_local_start_time(chore_name, date_time, **kwargs)

Makes Server crash if chore is activated (10.2.2 FP6) :)

Parameters:

Name Type Description Default
chore_name str
required
date_time datetime
required

Returns:

Type Description
Response
Source code in TM1py/Services/ChoreService.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
@deactivate_activate
def set_local_start_time(self, chore_name: str, date_time: datetime, **kwargs) -> Response:
    """Makes Server crash if chore is activated (10.2.2 FP6) :)
    :param chore_name:
    :param date_time:
    :return:
    """
    url = format_url("/Chores('{}')/tm1.SetServerLocalStartTime", chore_name)
    data = {
        "StartDate": "{}-{}-{}".format(date_time.year, date_time.month, date_time.day),
        "StartTime": "{}:{}:{}".format(
            self.zfill_two(date_time.hour), self.zfill_two(date_time.minute), self.zfill_two(date_time.second)
        ),
    }
    return self._rest.POST(url, json.dumps(data), **kwargs)

update(chore, **kwargs)

update chore on TM1 Server does not update: DST Sensitivity!

Parameters:

Name Type Description Default
chore Chore
required

Returns:

Type Description
Source code in TM1py/Services/ChoreService.py 
157
158
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
@deactivate_activate
def update(self, chore: Chore, **kwargs):
    """update chore on TM1 Server
    does not update: DST Sensitivity!
    :param chore:
    :return:
    """
    # Update StartTime, ExecutionMode, Frequency
    url = format_url("/Chores('{}')", chore.name)
    # Remove Tasks from Body. Tasks to be managed individually
    chore_dict_without_tasks = chore.body_as_dict
    chore_dict_without_tasks.pop("Tasks")
    self._rest.PATCH(url, json.dumps(chore_dict_without_tasks), **kwargs)

    # Update Tasks individually
    task_old_count = self._get_tasks_count(chore.name)
    for i, task_new in enumerate(chore.tasks):
        if i >= task_old_count:
            self._add_task(chore.name, task_new, **kwargs)
        else:
            task_old = self._get_task(chore.name, i)
            if task_new != task_old:
                task_new._step = i
                self._update_task(chore.name, task_new, **kwargs)
    for j in range(i + 1, task_old_count):
        self._delete_task(chore.name, i + 1, **kwargs)

    if chore.dst_sensitivity:
        self.set_local_start_time(chore.name, chore.start_time.datetime)

update_or_create(chore, **kwargs)

Source code in TM1py/Services/ChoreService.py 
187
188
189
190
191
def update_or_create(self, chore: Chore, **kwargs) -> Response:
    if self.exists(chore_name=chore.name, **kwargs):
        return self.update(chore, **kwargs)

    return self.create(chore=chore, **kwargs)

zfill_two(number) staticmethod

Pad an int with zeros on the left two create two digit string

Parameters:

Name Type Description Default
number int
required

Returns:

Type Description
str
Source code in TM1py/Services/ChoreService.py 
291
292
293
294
295
296
297
298
@staticmethod
def zfill_two(number: int) -> str:
    """Pad an int with zeros on the left two create two digit string

    :param number:
    :return:
    """
    return str(number).zfill(2)

deactivate_activate(func)

Higher Order function to handle activation and deactivation of chores before updating them

Parameters:

Name Type Description Default
func
required

Returns:

Type Description
Source code in TM1py/Services/ChoreService.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@decohints
def deactivate_activate(func):
    """Higher Order function to handle activation and deactivation of chores before updating them

    :param func:
    :return:
    """

    @functools.wraps(func)
    def wrapper(self, *args, **kwargs):
        # chore (type Chore) or chore_name (type str) is passed as first arg or kwargs
        if "chore" in kwargs:
            chore = kwargs.get("chore").name
        elif "chore_name" in kwargs:
            chore = kwargs.get("chore_name")
        else:
            chore = args[0]

        reactivate = None
        if isinstance(chore, Chore):
            chore_name = chore.name
            reactivate = chore.active
        elif isinstance(chore, str):
            chore_name = chore
        else:
            raise ValueError("Argument must be of type 'Chore' or 'str'")

        # Get current chore
        chore_old = self.get(chore_name)
        if reactivate is None:
            reactivate = chore_old.active

        # Deactivate
        if chore_old.active:
            self.deactivate(chore_name)
        # Do stuff
        try:
            response = func(self, *args, **kwargs)
        except Exception as e:
            raise e
        # Activate if necessary
        finally:
            if reactivate:
                self.activate(chore_name)
        return response

    return wrapper