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
|
"""Handle application level errors."""
import traceback
from http.client import RemoteDisconnected
from urllib.error import URLError
from sqlite3 import OperationalError
from SPARQLWrapper.SPARQLExceptions import (
EndPointInternalError,
EndPointNotFound,
QueryBadFormed,
URITooLong,
Unauthorized
)
from werkzeug.exceptions import NotFound
from authlib.oauth2.rfc6749.errors import OAuth2Error
from flask import Flask, jsonify, Response, current_app
from gn3.auth.authorisation.errors import AuthorisationError
from gn3.llms.errors import LLMError
def add_trace(exc: Exception, jsonmsg: dict) -> dict:
"""Add the traceback to the error handling object."""
return {
**jsonmsg,
"error-trace": "".join(traceback.format_exception(exc))
}
def page_not_found(pnf):
"""Generic 404 handler."""
return jsonify(add_trace(pnf, {
"error": pnf.name,
"error_description": pnf.description
})), 404
def internal_server_error(pnf):
"""Generic 404 handler."""
return jsonify(add_trace(pnf, {
"error": pnf.name,
"error_description": pnf.description
})), 500
def url_server_error(pnf):
"""Handler for an exception with a url connection."""
return jsonify(add_trace(pnf, {
"error": f"URLLib Error no: {pnf.reason.errno}",
"error_description": pnf.reason.strerror,
}))
def handle_authorisation_error(exc: AuthorisationError):
"""Handle AuthorisationError if not handled anywhere else."""
current_app.logger.error(exc)
return jsonify(add_trace(exc, {
"error": type(exc).__name__,
"error_description": " :: ".join(exc.args)
})), exc.error_code
def handle_oauth2_errors(exc: OAuth2Error):
"""Handle OAuth2Error if not handled anywhere else."""
current_app.logger.error(exc)
return jsonify(add_trace(exc, {
"error": exc.error,
"error_description": exc.description,
})), exc.status_code
def handle_sqlite3_errors(exc: OperationalError):
"""Handle sqlite3 errors if not handled anywhere else."""
current_app.logger.error(exc)
return jsonify({
"error": "DatabaseError",
"error_description": exc.args[0],
}), 500
def handle_sparql_errors(exc):
"""Handle sqlite3 errors if not handled anywhere else."""
current_app.logger.error(exc)
__code = {
EndPointInternalError: 500,
EndPointNotFound: 400,
QueryBadFormed: 400,
Unauthorized: 401,
URITooLong: 414,
}
return jsonify({
"error": exc.msg,
"error_description": str(exc),
}), __code.get(exc)
def handle_generic(exc: Exception) -> Response:
"""Handle generic exception."""
current_app.logger.error(exc)
resp = jsonify({
"error": type(exc).__name__,
"error_description": (
exc.args[0] if bool(exc.args) else "Generic Exception"),
"trace": traceback.format_exc()
})
resp.status_code = 500
return resp
def handle_llm_error(exc: Exception) -> Response:
""" Handle llm erros if not handled anywhere else. """
current_app.logger.error(exc)
resp = jsonify({
"query": exc.args[1],
"error_type": type(exc).__name__,
"error": (
exc.args[0] if bool(exc.args) else "Fahamu gnqa error occurred"
),
"trace": traceback.format_exc()
})
resp.status_code = 500
return resp
def register_error_handlers(app: Flask):
"""Register application-level error handlers."""
app.register_error_handler(NotFound, page_not_found)
app.register_error_handler(Exception, handle_generic)
app.register_error_handler(OAuth2Error, handle_oauth2_errors)
app.register_error_handler(OperationalError, handle_sqlite3_errors)
app.register_error_handler(AuthorisationError, handle_authorisation_error)
app.register_error_handler(RemoteDisconnected, internal_server_error)
app.register_error_handler(URLError, url_server_error)
app.register_error_handler(LLMError, handle_llm_error)
for exc in (
EndPointInternalError,
EndPointNotFound,
QueryBadFormed,
URITooLong,
Unauthorized
):
app.register_error_handler(exc, handle_sparql_errors)
|
