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
|
"""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.oauth2 import errors as oautherrors
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."""
current_app.logger.error("Handling 404 errors", exc_info=True)
return jsonify(add_trace(pnf, {
"error": pnf.name,
"error_description": pnf.description
})), 404
def internal_server_error(pnf):
"""Generic 404 handler."""
current_app.logger.error("Handling internal server errors", exc_info=True)
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."""
current_app.logger.error("Handling url server errors", exc_info=True)
return jsonify(add_trace(pnf, {
"error": f"URLLib Error no: {pnf.reason.errno}",
"error_description": pnf.reason.strerror,
})), 500
def handle_authorisation_error(exc: AuthorisationError):
"""Handle AuthorisationError if not handled anywhere else."""
current_app.logger.error("Handling external auth errors", exc_info=True)
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("Handling external oauth2 errors", exc_info=True)
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("Handling sqlite3 errors", exc_info=True)
return jsonify({
"error": "DatabaseError",
"error_description": exc.args[0],
}), 500
def handle_sparql_errors(exc):
"""Handle sparql/virtuoso errors if not handled anywhere else."""
current_app.logger.error("Handling sparql errors", exc_info=True)
code = {
"EndPointInternalError": 500,
"EndPointNotFound": 404,
"QueryBadFormed": 400,
"Unauthorized": 401,
"URITooLong": 414
}
return jsonify({
"error": exc.msg,
}), code.get(exc.__class__.__name__)
def handle_generic(exc: Exception) -> Response:
"""Handle generic exception."""
current_app.logger.error("Handling generic errors", exc_info=True)
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_local_authorisation_errors(exc: oautherrors.AuthorisationError):
"""Handle errors relating to authorisation that are raised locally."""
current_app.logger.error("Handling local auth errors", exc_info=True)
return jsonify(add_trace(exc, {
"error": type(exc).__name__,
"error_description": " ".join(exc.args)
})), 400
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)
|
