-
Notifications
You must be signed in to change notification settings - Fork 1
/
transformation.py
104 lines (94 loc) · 3.99 KB
/
transformation.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
import marketplace_standard_app_api.models.transformation as transformation
from ..utils import check_capability_availability
from .base import _MarketPlaceAppBase
class MarketPlaceTransformationApp(_MarketPlaceAppBase):
@check_capability_availability
def get_transformation_list(
self, limit: int = 100, offset: int = 0
) -> transformation.TransformationListResponse:
return transformation.TransformationListResponse.parse_obj(
self._client.get(
self._proxy_path("getTransformationList"),
params={"limit": limit, "offset": offset},
).json()
)
@check_capability_availability
def new_transformation(
self, new_transformation: transformation.NewTransformationModel
) -> transformation.TransformationCreateResponse:
return transformation.TransformationCreateResponse.parse_obj(
self._client.post(
self._proxy_path("newTransformation"), json=new_transformation
).json()
)
@check_capability_availability
def get_transformation(
self, transformation_id: transformation.TransformationId
) -> transformation.TransformationModel:
return transformation.TransformationModel.parse_obj(
self._client.get(
self._proxy_path("getTransformation"),
params={"transformation_id": transformation_id},
).json()
)
@check_capability_availability
def delete_transformation(self, transformation_id: transformation.TransformationId):
self._client.delete(
self._proxy_path("deleteTransformation"),
params={"transformation_id": transformation_id},
)
@check_capability_availability("update_transformation")
def _update_transformation(
self,
transformation_id: transformation.TransformationId,
update: transformation.TransformationUpdateModel,
) -> transformation.TransformationUpdateResponse:
return transformation.TransformationUpdateResponse.parse_obj(
self._client.patch(
self._proxy_path("updateTransformation"),
params={"transformation_id": transformation_id},
json=update.dict(),
).json()
)
def start_transformation(
self, transformation_id: transformation.TransformationId
) -> transformation.TransformationStateResponse:
update: transformation.TransformationUpdateModel = (
transformation.TransformationUpdateModel(
state=transformation.TransformationState.RUNNING
)
)
update_response: transformation.TransformationUpdateResponse = (
self._update_transformation(
transformation_id=transformation_id, update=update
)
)
return transformation.TransformationStateResponse.parse_obj(
update_response.dict()
)
def stop_transformation(
self, transformation_id: transformation.TransformationId
) -> transformation.TransformationStateResponse:
update: transformation.TransformationUpdateModel = (
transformation.TransformationUpdateModel(
state=transformation.TransformationState.STOPPED
)
)
update_response: transformation.TransformationUpdateResponse = (
self._update_transformation(
transformation_id=transformation_id, update=update
)
)
return transformation.TransformationStateResponse.parse_obj(
update_response.dict()
)
@check_capability_availability
def get_transformation_state(
self, transformation_id: transformation.TransformationId
) -> transformation.TransformationState:
return transformation.TransformationStateResponse.parse_obj(
self._client.get(
self._proxy_path("getTransformationState"),
params={"transformation_id": transformation_id},
).json()
).state