5
5
import base64
6
6
import dataclasses
7
7
import datetime
8
+ import json
8
9
from http import HTTPStatus
9
10
from zoneinfo import ZoneInfo
10
11
11
- from flask import Flask , jsonify , request
12
+ from flask import Flask , Response , request
12
13
from pydantic import BaseSettings
13
14
14
15
from mock_vws .database import VuforiaDatabase
@@ -31,7 +32,7 @@ class TargetManagerSettings(BaseSettings):
31
32
"/databases/<string:database_name>" ,
32
33
methods = ["DELETE" ],
33
34
)
34
- def delete_database (database_name : str ) -> tuple [ str , int ] :
35
+ def delete_database (database_name : str ) -> Response :
35
36
"""
36
37
Delete a database.
37
38
@@ -44,23 +45,26 @@ def delete_database(database_name: str) -> tuple[str, int]:
44
45
if database_name == database .database_name
45
46
}
46
47
except ValueError :
47
- return "" , HTTPStatus .NOT_FOUND
48
+ return Response ( response = "" , status = HTTPStatus .NOT_FOUND )
48
49
49
50
TARGET_MANAGER .remove_database (database = matching_database )
50
- return "" , HTTPStatus .OK
51
+ return Response ( response = "" , status = HTTPStatus .OK )
51
52
52
53
53
54
@TARGET_MANAGER_FLASK_APP .route ("/databases" , methods = ["GET" ])
54
- def get_databases () -> tuple [ str , int ] :
55
+ def get_databases () -> Response :
55
56
"""
56
57
Return a list of all databases.
57
58
"""
58
59
databases = [database .to_dict () for database in TARGET_MANAGER .databases ]
59
- return jsonify (databases ), HTTPStatus .OK
60
+ return Response (
61
+ response = json .dumps (obj = databases ),
62
+ status = HTTPStatus .OK ,
63
+ )
60
64
61
65
62
66
@TARGET_MANAGER_FLASK_APP .route ("/databases" , methods = ["POST" ])
63
- def create_database () -> tuple [ str , int ] :
67
+ def create_database () -> Response :
64
68
"""
65
69
Create a new database.
66
70
@@ -91,27 +95,28 @@ def create_database() -> tuple[str, int]:
91
95
:status 201: The database has been successfully created.
92
96
"""
93
97
random_database = VuforiaDatabase ()
94
- server_access_key = request .json .get (
98
+ request_json = json .loads (request .data )
99
+ server_access_key = request_json .get (
95
100
"server_access_key" ,
96
101
random_database .server_access_key ,
97
102
)
98
- server_secret_key = request . json .get (
103
+ server_secret_key = request_json .get (
99
104
"server_secret_key" ,
100
105
random_database .server_secret_key ,
101
106
)
102
- client_access_key = request . json .get (
107
+ client_access_key = request_json .get (
103
108
"client_access_key" ,
104
109
random_database .client_access_key ,
105
110
)
106
- client_secret_key = request . json .get (
111
+ client_secret_key = request_json .get (
107
112
"client_secret_key" ,
108
113
random_database .client_secret_key ,
109
114
)
110
- database_name = request . json .get (
115
+ database_name = request_json .get (
111
116
"database_name" ,
112
117
random_database .database_name ,
113
118
)
114
- state_name = request . json .get (
119
+ state_name = request_json .get (
115
120
"state_name" ,
116
121
random_database .state .name ,
117
122
)
@@ -129,16 +134,22 @@ def create_database() -> tuple[str, int]:
129
134
try :
130
135
TARGET_MANAGER .add_database (database = database )
131
136
except ValueError as exc :
132
- return str (exc ), HTTPStatus .CONFLICT
133
-
134
- return jsonify (database .to_dict ()), HTTPStatus .CREATED
137
+ return Response (
138
+ response = str (exc ),
139
+ status = HTTPStatus .CONFLICT ,
140
+ )
141
+
142
+ return Response (
143
+ response = json .dumps (database .to_dict ()),
144
+ status = HTTPStatus .CREATED ,
145
+ )
135
146
136
147
137
148
@TARGET_MANAGER_FLASK_APP .route (
138
149
"/databases/<string:database_name>/targets" ,
139
150
methods = ["POST" ],
140
151
)
141
- def create_target (database_name : str ) -> tuple [ str , int ] :
152
+ def create_target (database_name : str ) -> Response :
142
153
"""
143
154
Create a new target in a given database.
144
155
"""
@@ -147,27 +158,31 @@ def create_target(database_name: str) -> tuple[str, int]:
147
158
for database in TARGET_MANAGER .databases
148
159
if database .database_name == database_name
149
160
]
150
- image_base64 = request .json ["image_base64" ]
161
+ request_json = json .loads (request .data )
162
+ image_base64 = request_json ["image_base64" ]
151
163
image_bytes = base64 .b64decode (s = image_base64 )
152
164
target = Target (
153
- name = request . json ["name" ],
154
- width = request . json ["width" ],
165
+ name = request_json ["name" ],
166
+ width = request_json ["width" ],
155
167
image_value = image_bytes ,
156
- active_flag = request . json ["active_flag" ],
157
- processing_time_seconds = request . json ["processing_time_seconds" ],
158
- application_metadata = request . json ["application_metadata" ],
159
- target_id = request . json ["target_id" ],
168
+ active_flag = request_json ["active_flag" ],
169
+ processing_time_seconds = request_json ["processing_time_seconds" ],
170
+ application_metadata = request_json ["application_metadata" ],
171
+ target_id = request_json ["target_id" ],
160
172
)
161
173
database .targets .add (target )
162
174
163
- return jsonify (target .to_dict ()), HTTPStatus .CREATED
175
+ return Response (
176
+ response = json .dumps (target .to_dict ()),
177
+ status = HTTPStatus .CREATED ,
178
+ )
164
179
165
180
166
181
@TARGET_MANAGER_FLASK_APP .route (
167
182
"/databases/<string:database_name>/targets/<string:target_id>" ,
168
183
methods = ["DELETE" ],
169
184
)
170
- def delete_target (database_name : str , target_id : str ) -> tuple [ str , int ] :
185
+ def delete_target (database_name : str , target_id : str ) -> Response :
171
186
"""
172
187
Delete a target.
173
188
"""
@@ -181,14 +196,17 @@ def delete_target(database_name: str, target_id: str) -> tuple[str, int]:
181
196
new_target = dataclasses .replace (target , delete_date = now )
182
197
database .targets .remove (target )
183
198
database .targets .add (new_target )
184
- return jsonify (new_target .to_dict ()), HTTPStatus .OK
199
+ return Response (
200
+ response = json .dumps (new_target .to_dict ()),
201
+ status = HTTPStatus .OK ,
202
+ )
185
203
186
204
187
205
@TARGET_MANAGER_FLASK_APP .route (
188
206
"/databases/<string:database_name>/targets/<string:target_id>" ,
189
207
methods = ["PUT" ],
190
208
)
191
- def update_target (database_name : str , target_id : str ) -> tuple [ str , int ] :
209
+ def update_target (database_name : str , target_id : str ) -> Response :
192
210
"""
193
211
Update a target.
194
212
"""
@@ -199,17 +217,19 @@ def update_target(database_name: str, target_id: str) -> tuple[str, int]:
199
217
]
200
218
target = database .get_target (target_id = target_id )
201
219
202
- width = request .json .get ("width" , target .width )
203
- name = request .json .get ("name" , target .name )
204
- active_flag = request .json .get ("active_flag" , target .active_flag )
205
- application_metadata = request .json .get (
220
+ request_json = json .loads (request .data )
221
+ width = request_json .get ("width" , target .width )
222
+ name = request_json .get ("name" , target .name )
223
+ active_flag = request_json .get ("active_flag" , target .active_flag )
224
+ application_metadata = request_json .get (
206
225
"application_metadata" ,
207
226
target .application_metadata ,
208
227
)
209
228
210
229
image_value = target .image_value
211
- if "image" in request .json :
212
- image_value = base64 .b64decode (s = request .json ["image" ])
230
+ request_json = json .loads (request .data )
231
+ if "image" in request_json :
232
+ image_value = base64 .b64decode (s = request_json ["image" ])
213
233
214
234
gmt = ZoneInfo ("GMT" )
215
235
last_modified_date = datetime .datetime .now (tz = gmt )
@@ -227,7 +247,10 @@ def update_target(database_name: str, target_id: str) -> tuple[str, int]:
227
247
database .targets .remove (target )
228
248
database .targets .add (new_target )
229
249
230
- return jsonify (new_target .to_dict ()), HTTPStatus .OK
250
+ return Response (
251
+ response = json .dumps (new_target .to_dict ()),
252
+ status = HTTPStatus .OK ,
253
+ )
231
254
232
255
233
256
if __name__ == "__main__" : # pragma: no cover
0 commit comments