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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
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
144
145
146
147
148
149
150
151
152
153
154
155
156
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
186
187
188
|
from typing import Optional
from urllib.parse import urljoin
import requests
from requests import HTTPError
from girok.api.entity import APIResponse
from girok.config.auth_handler import AuthHandler
from girok.constants import BASE_URL
def get_all_categories() -> APIResponse:
access_token = AuthHandler.get_access_token()
resp = requests.get(
url=urljoin(BASE_URL, "categories"),
headers={"Authorization": "Bearer " + access_token},
)
try:
resp.raise_for_status()
return APIResponse(is_success=True, body=resp.json())
except HTTPError as e:
try:
error_body = resp.json()
error_message = error_body["message"]
except:
error_message = "Failed to get categories"
return APIResponse(is_success=False, error_message=error_message)
def create_category(category_path: str, color: str) -> APIResponse:
access_token = AuthHandler.get_access_token()
category_path_list = category_path.split("/")
new_category_name = category_path_list[-1]
# Resolve parent category's id
parent_category_id_resp = get_category_id_by_path(category_path_list[:-1])
if not parent_category_id_resp.is_success:
return APIResponse(is_success=False, error_message=parent_category_id_resp.error_message)
parent_category_id = parent_category_id_resp.body["categoryId"]
resp = requests.post(
url=urljoin(BASE_URL, "categories"),
headers={"Authorization": "Bearer " + access_token},
json={"parentId": parent_category_id, "name": new_category_name, "color": color},
)
try:
resp.raise_for_status()
return APIResponse(is_success=True, body=resp.json())
except HTTPError:
try:
error_body = resp.json()
error_code = error_body["errorCode"]
error_message = error_body["message"]
if error_code == "DUPLICATE_CATEGORY":
parent_category_path_str = (
"/" if not category_path_list[:-1] else "/".join(category_path_list[:-1]) + "/"
)
error_message = f"Duplicate Category: '{parent_category_path_str}' already has '{new_category_name}'"
except:
error_message = "Failed to create a new category"
return APIResponse(is_success=False, error_message=error_message)
def remove_category(category_path: str) -> APIResponse:
access_token = AuthHandler.get_access_token()
category_path_list = category_path.split("/")
category_id_resp = get_category_id_by_path(category_path_list)
if not category_id_resp.is_success:
return APIResponse(is_success=False, error_message=category_id_resp.error_message)
category_id = category_id_resp.body["categoryId"]
resp = requests.delete(
url=urljoin(BASE_URL, f"categories/{category_id}"),
headers={"Authorization": "Bearer " + access_token},
)
try:
resp.raise_for_status()
return APIResponse(is_success=True)
except HTTPError:
try:
error_body = resp.json()
error_message = error_body["message"]
except:
error_message = "Failed to remove a category"
return APIResponse(is_success=False, error_message=error_message)
def update_category(category_path: str, new_name: Optional[str] = None, new_color: Optional[str] = None) -> APIResponse:
access_token = AuthHandler.get_access_token()
category_path_list = category_path.split("/")
category_id_resp = get_category_id_by_path(category_path_list)
if not category_id_resp.is_success:
return category_id_resp
category_id = category_id_resp.body["categoryId"]
body = {}
if new_name:
body["newName"] = new_name
if new_color:
body["color"] = new_color
resp = requests.patch(
url=urljoin(BASE_URL, f"categories/{category_id}"),
headers={"Authorization": "Bearer " + access_token},
json=body,
)
try:
resp.raise_for_status()
return APIResponse(is_success=True)
except HTTPError:
try:
error_body = resp.json()
error_message = error_body["message"]
except:
error_message = "Failed to rename a category"
return APIResponse(is_success=False, error_message=error_message)
def move_category(path: str, new_parent_path: str) -> APIResponse:
# girok mvcat A/B/C D/E
access_token = AuthHandler.get_access_token()
path_list = path.split("/")
new_parent_path_list = new_parent_path.split("/") if new_parent_path else []
# 1. Get the category id
resp = get_category_id_by_path(path_list)
if not resp.is_success:
return resp
category_id = resp.body["categoryId"]
# 2. Get target parent's id
resp = get_category_id_by_path(new_parent_path_list)
if not resp.is_success:
return resp
new_parent_category_id = resp.body["categoryId"]
resp = requests.patch(
url=urljoin(BASE_URL, f"categories/{category_id}/parent"),
headers={"Authorization": "Bearer " + access_token},
json={"newParentId": new_parent_category_id},
)
try:
resp.raise_for_status()
return APIResponse(is_success=True)
except HTTPError:
try:
error_body = resp.json()
error_message = error_body["message"]
except:
error_message = "Failed to move a category"
return APIResponse(is_success=False, error_message=error_message)
def get_category_id_by_path(path_list: list[str]) -> APIResponse:
if len(path_list) == 0:
return APIResponse(is_success=True, body={"categoryId": None})
access_token = AuthHandler.get_access_token()
resp = requests.get(
url=urljoin(BASE_URL, "categories/id-by-path"),
headers={"Authorization": "Bearer " + access_token},
params={"path": path_list},
)
try:
resp.raise_for_status()
return APIResponse(is_success=True, body=resp.json())
except HTTPError:
try:
error_body = resp.json()
error_message = error_body["message"]
except:
error_message = f"Failed to get a category id of '{path_list}'"
return APIResponse(is_success=False, error_message=error_message)
|
