apapi.transactional

  1"""
  2apapi.transactional
  3
  4Child of Basic Connection class, responsible for Transactional API capabilities.
  5"""
  6from __future__ import annotations
  7
  8import json
  9from typing import Iterable
 10
 11from requests import Response
 12
 13from .basic_connection import BasicConnection
 14from .utils import ENCODING_GZIP, PAGING_LIMIT, ExportType, MIMEType
 15
 16
 17class TransactionalConnection(BasicConnection):
 18    """Anaplan connection with Transactional API functions."""
 19
 20    # Users
 21    def get_users(self) -> Response:
 22        """Get info about all users in the tenant."""
 23        return self.request("GET", f"{self._api_main_url}/users")
 24
 25    def get_me(self) -> Response:
 26        """Get info about me."""
 27        return self.request("GET", f"{self._api_main_url}/users/me")
 28
 29    def get_user(self, user_id: str) -> Response:
 30        """Get info about a specified user."""
 31        return self.request("GET", f"{self._api_main_url}/users/{user_id}")
 32
 33    def get_workspace_users(self, workspace_id: str) -> Response:
 34        """Get info about users with access to a specified workspace."""
 35        return self.request(
 36            "GET",
 37            f"{self._api_main_url}/workspaces/{workspace_id}/users",
 38        )
 39
 40    def get_workspace_admins(self, workspace_id: str) -> Response:
 41        """Get info about all administrators of a specified workspace."""
 42        return self.request(
 43            "GET",
 44            f"{self._api_main_url}/workspaces/{workspace_id}/admins",
 45            {"limit": PAGING_LIMIT},  # needed for this endpoint, as default is 20
 46        )
 47
 48    def get_model_users(self, model_id: str) -> Response:
 49        """Get info about users with access to a specified model."""
 50        return self.request("GET", f"{self._api_main_url}/models/{model_id}/users")
 51
 52    # Workspaces
 53    def get_workspaces(self, details: bool = None) -> Response:
 54        """Get info about all workspaces in the tenant."""
 55        return self.request(
 56            "GET",
 57            f"{self._api_main_url}/workspaces",
 58            {"tenantDetails": self.details if details is None else details},
 59        )
 60
 61    def get_workspace(self, workspace_id: str, details: bool = None) -> Response:
 62        """Get info about all a specified workspace."""
 63        return self.request(
 64            "GET",
 65            f"{self._api_main_url}/workspaces/{workspace_id}",
 66            {"tenantDetails": self.details if details is None else details},
 67        )
 68
 69    def get_user_workspaces(self, user_id: str) -> Response:
 70        """Get info about all workspaces to which a specified user has access."""
 71        return self.request("GET", f"{self._api_main_url}/users/{user_id}/workspaces")
 72
 73    # Models
 74    def get_models(self, details: bool = None) -> Response:
 75        """Get info about all models in the tenant."""
 76        return self.request(
 77            "GET",
 78            f"{self._api_main_url}/models",
 79            {"modelDetails": self.details if details is None else details},
 80        )
 81
 82    def get_workspace_models(self, workspace_id: str, details: bool = None) -> Response:
 83        """Get info about all models within a specified workspace."""
 84        return self.request(
 85            "GET",
 86            f"{self._api_main_url}/workspaces/{workspace_id}/models",
 87            {"modelDetails": self.details if details is None else details},
 88        )
 89
 90    def get_model(self, model_id: str, details: bool = None) -> Response:
 91        """Get info about a specified model."""
 92        return self.request(
 93            "GET",
 94            f"{self._api_main_url}/models/{model_id}",
 95            {"modelDetails": self.details if details is None else details},
 96        )
 97
 98    def get_user_models(self, user_id: str) -> Response:
 99        """Get info about all models to which a specified user has access."""
100        return self.request("GET", f"{self._api_main_url}/users/{user_id}/models")
101
102    def delete_models(self, workspace_id: str, models_ids: [str]) -> Response:
103        """Delete models from a workspace.
104
105        **WARNING**: This query is destructive - use with caution! It will only work on
106        closed models - you can close a model in Anaplan web interface via Manage Tasks.
107        """
108        return self.request(
109            "POST",
110            f"{self._api_main_url}/workspaces/{workspace_id}/bulkDeleteModels",
111            data=json.dumps({"modelIdsToDelete": models_ids}),
112        )
113
114    # Calendar
115    def get_fiscal_year(self, model_id: str):
116        """Get current fiscal year setting for a specified model."""
117        return self.request(
118            "GET", f"{self._api_main_url}/models/{model_id}/modelCalendar"
119        )
120
121    def set_fiscal_year(self, model_id: str, data: str):
122        """Set current fiscal year setting."""
123        return self.request(
124            "PUT",
125            f"{self._api_main_url}/models/{model_id}/modelCalendar/fiscalYear",
126            data=json.dumps({"year": data}),
127        )
128
129    def get_current_period(self, model_id: str):
130        """Get current period setting a specified model."""
131        return self.request(
132            "GET", f"{self._api_main_url}/models/{model_id}/currentPeriod"
133        )
134
135    def set_current_period(self, model_id: str, data: str):
136        """Set current period setting (data should have YYYY-MM-DD format)."""
137        return self.request(
138            "PUT",
139            f"{self._api_main_url}/models/{model_id}/currentPeriod",
140            data=json.dumps({"date": data}),
141        )
142
143    # Versions
144    def get_versions(self, model_id: str):
145        """Get versions information and settings."""
146        return self.request("GET", f"{self._api_main_url}/models/{model_id}/versions")
147
148    def set_version_switchover(self, model_id: str, version_id: str, data: str):
149        """Set version's switchover (data should have YYYY-MM-DD format)."""
150        return self.request(
151            "PUT",
152            f"{self._api_main_url}/models/{model_id}/versions/{version_id}/switchover",
153            data=json.dumps({"date": data}),
154        )
155
156    # Lists
157    def get_lists(self, model_id: str) -> Response:
158        """Get lists information for a specified model."""
159        return self.request("GET", f"{self._api_main_url}/models/{model_id}/lists")
160
161    def get_list(self, model_id: str, list_id: str) -> Response:
162        """Get information about a specified list."""
163        return self.request(
164            "GET", f"{self._api_main_url}/models/{model_id}/lists/{list_id}"
165        )
166
167    def get_list_items(
168        self, model_id: str, list_id: str, details: bool = None, accept: MIMEType = None
169    ) -> Response:
170        """Get list's items, up to a million lines.
171
172        **WARNING**: This query can be used to retrieve information for smaller lists.
173        For larger lists use large_list_read functions.
174
175        Extra details (selective access, subsets and properties) are available.
176        Returned data can be either in JSON (default) or CSV format.
177        """
178        return self.request(
179            "GET",
180            f"{self._api_main_url}/models/{model_id}/lists/{list_id}/items",
181            {"includeAll": self.details if details is None else details},
182            headers={"Accept": accept.value} if accept else None,
183        )
184
185    def start_large_list_read(self, model_id: str, list_id: str) -> Response:
186        """Start large list read, which allows getting unlimited list's items."""
187        return self.request(
188            "POST",
189            f"{self._api_main_url}/models/{model_id}/lists/{list_id}/readRequests",
190        )
191
192    def get_large_list_read_status(
193        self, model_id: str, list_id: str, request_id: str
194    ) -> Response:
195        """Get status of a specified large list read request.
196
197        Tip: some pages might be available even if process is still in progress.
198        """
199        return self.request(
200            "GET",
201            f"{self._api_main_url}/models/{model_id}/lists/{list_id}/readRequests/{request_id}",
202        )
203
204    def get_large_list_read_data(
205        self,
206        model_id: str,
207        list_id: str,
208        request_id: str,
209        page: str,
210        compress: bool = None,
211    ) -> Response:
212        """Get a page of data of a specified large list read request.
213
214        You can get information about haw many pages are already available using
215        TransactionalConnection.get_large_list_read_status().
216        If there are i.e. 10 available pages, it means that pages from 0 to 9 are ready.
217        You can clean after read using TransactionalConnection.delete_large_list_read().
218        """
219        headers = {"Accept": MIMEType.TEXT_CSV.value}
220        if compress or (compress is None and self.compress):
221            headers["Accept-Encoding"] = ENCODING_GZIP
222        return self.request(
223            "GET",
224            f"{self._api_main_url}/models/{model_id}/lists/{list_id}/readRequests/{request_id}/pages/{page}",
225            headers=headers,
226        )
227
228    def delete_large_list_read(
229        self, model_id: str, list_id: str, request_id: str
230    ) -> Response:
231        """Delete specified large list read request.
232
233        You can call this endpoint to free up space if you don't need the data anymore.
234        If not accessed, data will be cleared after 30 minutes automatically.
235        """
236        return self.request(
237            "DELETE",
238            f"{self._api_main_url}/models/{model_id}/lists/{list_id}/readRequests/{request_id}",
239        )
240
241    def add_list_items(self, model_id: str, list_id: str, data: list[dict]) -> Response:
242        """Add specified items to a list.
243
244        Array of items definitions is expected as documented in the official Anaplan
245        [API documentation](https://anaplanbulkapi20.docs.apiary.io/#AddListItems).
246        """
247        return self.request(
248            "POST",
249            f"{self._api_main_url}/models/{model_id}/lists/{list_id}/items",
250            {"action": "add"},
251            json.dumps({"items": data}),
252        )
253
254    def update_list_items(
255        self, model_id: str, list_id: str, data: list[dict]
256    ) -> Response:
257        """Update a specified items of a list.
258
259        Array of items definitions is expected as documented in the official Anaplan
260        [API documentation](https://anaplanbulkapi20.docs.apiary.io/#UpdateListItems).
261        """
262        return self.request(
263            "PUT",
264            f"{self._api_main_url}/models/{model_id}/lists/{list_id}/items",
265            data=json.dumps({"items": data}),
266        )
267
268    def delete_list_items(
269        self, model_id: str, list_id: str, data: list[dict]
270    ) -> Response:
271        """Delete specified items from a list.
272
273        Data parameter should consist of dictionaries identifying items by id or code.
274        """
275        return self.request(
276            "POST",
277            f"{self._api_main_url}/models/{model_id}/lists/{list_id}/items",
278            {"action": "delete"},
279            json.dumps({"items": data}),
280        )
281
282    def reset_list_index(self, model_id: str, list_id: str) -> Response:
283        """Reset index of a specified numbered list.
284
285        **WARNING**: This action works only for numbered lists that are empty.
286        """
287        return self.request(
288            "POST", f"{self._api_main_url}/models/{model_id}/lists/{list_id}/resetIndex"
289        )
290
291    # Modules
292    def get_modules(self, model_id: str) -> Response:
293        """Get modules information for a specified model."""
294        return self.request("GET", f"{self._api_main_url}/models/{model_id}/modules")
295
296    # Lineitems
297    def get_lineitems(self, model_id: str, details: bool = None) -> Response:
298        """Get all lineitems information for a specified model."""
299        return self.request(
300            "GET",
301            f"{self._api_main_url}/models/{model_id}/lineItems",
302            {"includeAll": self.details if details is None else details},
303        )
304
305    def get_module_lineitems(
306        self, model_id: str, module_id: str, details: bool = None
307    ) -> Response:
308        """Get lineitems information for a specified module."""
309        return self.request(
310            "GET",
311            f"{self._api_main_url}/models/{model_id}/modules/{module_id}/lineItems",
312            {"includeAll": self.details if details is None else details},
313        )
314
315    # Views
316    def get_views(self, model_id: str, details: bool = None) -> Response:
317        """Get all views information for a specified model."""
318        return self.request(
319            "GET",
320            f"{self._api_main_url}/models/{model_id}/views",
321            {"includesubsidiaryviews": self.details if details is None else details},
322        )
323
324    def get_module_views(
325        self, model_id: str, module_id: str, details: bool = None
326    ) -> Response:
327        """Get views information for a specified module."""
328        return self.request(
329            "GET",
330            f"{self._api_main_url}/models/{model_id}/modules/{module_id}/views",
331            {"includesubsidiaryviews": self.details if details is None else details},
332        )
333
334    def get_view_dimensions(self, model_id: str, view_id: str) -> Response:
335        """Get dimensions information for a specified view."""
336        return self.request(
337            "GET", f"{self._api_main_url}/models/{model_id}/views/{view_id}"
338        )
339
340    # Dimensions
341    def get_dimension_items(self, model_id: str, dimension_id: str) -> Response:
342        """Get list of all items within a specified dimension."""
343        return self.request(
344            "GET",
345            f"{self._api_main_url}/models/{model_id}/dimensions/{dimension_id}/items",
346        )
347
348    def get_lineitem_dimensions(self, model_id: str, lineitem_id: str) -> Response:
349        """Get dimensions information for a specified lineitem."""
350        return self.request(
351            "GET",
352            f"{self._api_main_url}/models/{model_id}/lineItems/{lineitem_id}/dimensions",
353        )
354
355    def get_lineitem_dimension_items(
356        self, model_id: str, lineitem_id: str, dimension_id: str
357    ) -> Response:
358        """Get list of items for a specified dimension in the context of a lineitem."""
359        return self.request(
360            "GET",
361            f"{self._api_main_url}/models/{model_id}/lineItems/{lineitem_id}/dimensions/{dimension_id}/items",
362        )
363
364    def get_view_dimension_items(
365        self, model_id: str, view_id: str, dimension_id: str
366    ) -> Response:
367        """Get list of items for a specified dimension in the context of a view."""
368        return self.request(
369            "GET",
370            f"{self._api_main_url}/models/{model_id}/views/{view_id}/dimensions/{dimension_id}/items",
371        )
372
373    def check_dimension_items_id(
374        self, model_id: str, dimension_id: str, data: dict
375    ) -> Response:
376        """Get ids of dimension's items specified by names or codes."""
377        return self.request(
378            "POST",
379            f"{self._api_main_url}/models/{model_id}/dimensions/{dimension_id}/items",
380            data=json.dumps(data),
381        )
382
383    # Cells
384    def get_cell_data(
385        self,
386        model_id: str,
387        view_id: str,
388        accept: MIMEType = None,
389        pages: Iterable[str] = None,
390    ) -> Response:
391        """Get cells values for a specified view, up to a million cells.
392
393        **WARNING**: This query can be used to retrieve information for smaller views.
394        For larger views use large_cell_read functions.
395
396        Pages argument should be iterable collection of dimensionId & itemId strings
397        in a form of key-value pairs joined by ":" (colon),
398        i.e. ["101000000026:330000000028","20000000012:587000000000"]
399        Returned data can be either in JSON (default), CSV or CSV-escaped format.
400        More information about this endpoint can be found in the official Anaplan
401        [API documentation](https://anaplanbulkapi20.docs.apiary.io/#RetrieveCellDataView).
402        """
403        headers = self._session.headers.copy()
404        if accept:
405            headers["Accept"] = accept.value
406        params = {}
407        if pages:
408            params["pages"] = ",".join(pages)
409        if headers["Accept"] == MIMEType.APP_JSON.value:
410            params["format"] = "v1"
411        return self.request(
412            "GET",
413            f"{self._api_main_url}/models/{model_id}/views/{view_id}/data",
414            params=params,
415            headers=headers,
416        )
417
418    def start_large_cell_read(
419        self, model_id: str, view_id: str, mode: ExportType
420    ) -> Response:
421        """Start large cell read, which allows getting unlimited cells for a view.
422
423        Returned data grid mode should be set using apapi.utils.ExportType.
424        """
425        return self.request(
426            "POST",
427            f"{self._api_main_url}/models/{model_id}/views/{view_id}/readRequests",
428            data=json.dumps({"exportType": mode.value}),
429        )
430
431    def get_large_cell_read_status(
432        self, model_id: str, view_id: str, request_id: str
433    ) -> Response:
434        """Get status of a specified large cell read request.
435
436        Tip: some pages might be available even if process is still in progress.
437        """
438        return self.request(
439            "GET",
440            f"{self._api_main_url}/models/{model_id}/views/{view_id}/readRequests/{request_id}",
441        )
442
443    def get_large_cell_read_data(
444        self,
445        model_id: str,
446        view_id: str,
447        request_id: str,
448        page: str,
449        compress: bool = None,
450    ) -> Response:
451        """Get a page of data of a specified large cell read request.
452
453        You can get information about haw many pages are already available using
454        TransactionalConnection.get_large_cell_read_status().
455        If there are i.e. 10 available pages, it means that pages from 0 to 9 are ready.
456        You can clean after read using TransactionalConnection.delete_large_cell_read().
457        """
458        headers = {"Accept": MIMEType.TEXT_CSV.value}
459        if compress or (compress is None and self.compress):
460            headers["Accept-Encoding"] = ENCODING_GZIP
461        return self.request(
462            "GET",
463            f"{self._api_main_url}/models/{model_id}/views/{view_id}/readRequests/{request_id}/pages/{page}",
464            headers=headers,
465        )
466
467    def delete_large_cell_read(
468        self, model_id: str, view_id: str, request_id: str
469    ) -> Response:
470        """Delete specified large cell read request.
471
472        You can call this endpoint to free up space if you don't need the data anymore.
473        If not accessed, data will be cleared after 30 minutes automatically.
474        """
475        return self.request(
476            "DELETE",
477            f"{self._api_main_url}/models/{model_id}/views/{view_id}/readRequests/{request_id}",
478        )
479
480    def post_cell_data(
481        self, model_id: str, module_id: str, data: list[dict]
482    ) -> Response:
483        """Update value of specific cells in a module.
484
485        Data argument should be a list of dictionaries containing lineItemId,
486        dimensions definition (ids of dimensions and items), and new value.
487        More information about this endpoint can be found in the official Anaplan
488        [API documentation](https://anaplanbulkapi20.docs.apiary.io/#WriteCellDataByCoordinateModule).
489        """
490        return self.request(
491            "POST",
492            f"{self._api_main_url}/models/{model_id}/modules/{module_id}/data",
493            data=json.dumps(data),
494        )