-
Notifications
You must be signed in to change notification settings - Fork 0
/
api_utils.py
110 lines (78 loc) · 2.77 KB
/
api_utils.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
105
106
107
108
109
110
"""
Genestack Uploader
A HTTP server providing an API and a frontend for easy uploading to Genestack
Copyright (C) 2021, 2022 Genome Research Limited
Author: Michael Grace <[email protected]>
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
"""
import typing as T
Response = T.Tuple[T.Dict[str, T.Any], int]
def create_response(data: T.Any, code: int = 200) -> Response:
"""
Return an API response that complies to the API
Params:
- data: the main body of the response
- code: HTTP code, anything 200-299 is considered OK, anything else is FAIL
Returns:
- Tuple[Dict, int]
Dict: The API response containing status and data as in the spec
int: HTTP Response code
"""
return {
"status": "OK" if code // 100 == 2 else "FAIL",
"data": data
}, code
MISSING_TOKEN = create_response({"error": "missing token"}, 401)
FORBIDDEN = create_response({"error": "forbidden"}, 403)
def internal_server_error(err: Exception):
"""
500 Internal Server Error Response
"""
return create_response({
"error": "internal server error",
"name": err.__class__.__name__,
"detail": err.args
}, 500)
def not_found(err: Exception) -> Response:
"""
404 Not Found Response
"""
return create_response({
"error": "not found",
"name": err.__class__.__name__,
"detail": err.args
}, 404)
class EndpointNotFoundError(Exception):
"""
For default 404 in the API
"""
def __init__(self) -> None:
super().__init__("Not Found")
class SignalNotFoundError(Exception):
"""For when signal not found"""
class MultipleSignalsFoundError(Exception):
"""
When multiple signals are found
given the criteria
"""
class TemplateNotFoundError(Exception):
"""When a template isn't found"""
class StudyNotFoundError(Exception):
"""When a study isn't found"""
class JobIDNotFound(KeyError):
"""when a job ID isn't found.
this could be because it expired"""
def __init__(self, *args: object) -> None:
super().__init__(
"Job ID not found. The job may have expired if it finished more than a week ago",
*args
)