Đây là một cách chung để làm những việc mà ít nhất có nghĩa là bạn không phải bao quanh mỗi cuộc gọi requests
với try ... except
:
Phiên bản cơ bản
# see the docs: if you set no timeout the call never times out! A tuple means "max
# connect time" and "max read time"
DEFAULT_REQUESTS_TIMEOUT = [5, 15] # for example
def log_exception[e, verb, url, kwargs]:
# the reason for making this a separate function will become apparent
raw_tb = traceback.extract_stack[]
if 'data' in kwargs and len[kwargs['data']] > 500: # anticipate giant data string
kwargs['data'] = f'{kwargs["data"][:500]}...'
msg = f'BaseException raised: {e.__class__.__module__}.{e.__class__.__qualname__}: {e}\n' \
+ f'verb {verb}, url {url}, kwargs {kwargs}\n\n' \
+ 'Stack trace:\n' + ''.join[traceback.format_list[raw_tb[:-2]]]
logger.error[msg]
def requests_call[verb, url, **kwargs]:
response = None
exception = None
try:
if 'timeout' not in kwargs:
kwargs['timeout'] = DEFAULT_REQUESTS_TIMEOUT
response = requests.request[verb, url, **kwargs]
except BaseException as e:
log_exception[e, verb, url, kwargs]
exception = e
return [response, exception]
NB
- Hãy nhận biết
ConnectionError
là một bản dựng, không liên quan gì đến lớp ________ 8*. Tôi cho rằng cái sau là phổ biến hơn trong bối cảnh này nhưng không có ý tưởng thực sự ... - Khi kiểm tra một ngoại lệ không trả về,
0, siêu lớp của tất cả các ngoại lệsearch_response, exception = utilities.requests_call['get', f'//localhost:9200/my_index/_search?q={search_string}']
requests
[bao gồmrequests.ConnectionError
], không phải là "
3" theo các tài liệu. Có lẽ nó đã thay đổi kể từ câu trả lời được chấp nhận. **search_response, exception = utilities.requests_call['get', f'//localhost:9200/my_index/_search?q={search_string}']
- Rõ ràng điều này giả định một logger đã được cấu hình. Gọi
4 trong khốisearch_response, exception = utilities.requests_call['get', f'//localhost:9200/my_index/_search?q={search_string}']
5 có vẻ là một ý tưởng tốt nhưng điều đó sẽ chỉ đưa ra ngăn xếp trong phương pháp này! Thay vào đó, hãy lấy dấu vết dẫn đến cuộc gọi đến phương thức này. Sau đó đăng nhập [với chi tiết ngoại lệ và cuộc gọi gây ra sự cố]up to the call to this method. Then log [with details of the exception, and of the call which caused the problem]search_response, exception = utilities.requests_call['get', f'//localhost:9200/my_index/_search?q={search_string}']
*Tôi đã xem mã nguồn: requests.ConnectionError
Các lớp con đơn ____10, phân loại lớp đơn
search_response, exception = utilities.requests_call['get',
f'//localhost:9200/my_index/_search?q={search_string}']
8 [tích hợp]** Tuy nhiên, ở cuối trang này, bạn tìm thấy "Yêu cầu. Exexceptions.RequesTexception" Tại thời điểm viết [2022-02] ... nhưng nó liên kết với trang trên: khó hiểu.
Việc sử dụng rất đơn giản:
search_response, exception = utilities.requests_call['get',
f'//localhost:9200/my_index/_search?q={search_string}']
Trước tiên, bạn kiểm tra phản hồi: Nếu đó là None
có điều gì đó buồn cười đã xảy ra và bạn sẽ có một ngoại lệ phải được hành động theo một cách nào đó tùy thuộc vào ngữ cảnh [và ngoại lệ]. Trong các ứng dụng GUI [PYQT5], tôi thường triển khai "nhật ký trực quan" để cung cấp một số đầu ra cho người dùng [và cũng đăng nhập đồng thời vào tệp nhật ký], nhưng các tin nhắn được thêm vào không phải là kỹ thuật. Vì vậy, một cái gì đó như thế này thường có thể làm theo:
if search_response == None:
# you might check here for [e.g.] a requests.Timeout, tailoring the message
# accordingly, as the kind of error anyone might be expected to understand
msg = f'No response searching on |{search_string}|. See log'
MainWindow.the[].visual_log[msg, log_level=logging.ERROR]
return
response_json = search_response.json[]
if search_response.status_code != 200: # NB 201 ["created"] may be acceptable sometimes...
msg = f'Bad response searching on |{search_string}|. See log'
MainWindow.the[].visual_log[msg, log_level=logging.ERROR]
# usually response_json will give full details about the problem
log_msg = f'search on |{search_string}| bad response\n{json.dumps[response_json, indent=4]}'
logger.error[log_msg]
return
# now examine the keys and values in response_json: these may of course
# indicate an error of some kind even though the response returned OK [status 200]...
Cho rằng dấu vết ngăn xếp được ghi lại tự động, bạn thường không cần nhiều hơn thế ...
Phiên bản nâng cao khi đối tượng JSON trở lại
[... có khả năng tiết lộ rất nhiều nồi hơi!]
Để vượt qua TS, khi một đối tượng JSON dự kiến sẽ được trả về:
Nếu, như trên, một ngoại lệ cung cấp cho người dùng phi kỹ thuật của bạn một thông báo "không có phản hồi" và "phản hồi xấu" không phải là 200, tôi đề nghị rằng"No response", and a non-200 status "Bad response", I suggest that
- Một khóa dự kiến còn thiếu trong cấu trúc JSON của phản hồi sẽ làm phát sinh một thông điệp "Phản hồi dị thường""Anomalous response"
- Giá trị ngoài phạm vi hoặc kỳ lạ đối với một tin nhắn "Phản hồi bất ngờ""Unexpected response"
- và sự hiện diện của một khóa như "lỗi" hoặc "lỗi", với giá trị
0 hoặc bất cứ điều gì, với một thông báo "phản hồi lỗi""Error response"if search_response == None: # you might check here for [e.g.] a requests.Timeout, tailoring the message # accordingly, as the kind of error anyone might be expected to understand msg = f'No response searching on |{search_string}|. See log' MainWindow.the[].visual_log[msg, log_level=logging.ERROR] return response_json = search_response.json[] if search_response.status_code != 200: # NB 201 ["created"] may be acceptable sometimes... msg = f'Bad response searching on |{search_string}|. See log' MainWindow.the[].visual_log[msg, log_level=logging.ERROR] # usually response_json will give full details about the problem log_msg = f'search on |{search_string}| bad response\n{json.dumps[response_json, indent=4]}' logger.error[log_msg] return # now examine the keys and values in response_json: these may of course # indicate an error of some kind even though the response returned OK [status 200]...
Những điều này có thể hoặc không thể ngăn mã tiếp tục.
... Và trên thực tế, tôi rất đáng để làm cho quá trình trở nên chung chung hơn. Đối với tôi, các chức năng tiếp theo này thường cắt giảm 20 dòng mã bằng cách sử dụng
if search_response == None:
# you might check here for [e.g.] a requests.Timeout, tailoring the message
# accordingly, as the kind of error anyone might be expected to understand
msg = f'No response searching on |{search_string}|. See log'
MainWindow.the[].visual_log[msg, log_level=logging.ERROR]
return
response_json = search_response.json[]
if search_response.status_code != 200: # NB 201 ["created"] may be acceptable sometimes...
msg = f'Bad response searching on |{search_string}|. See log'
MainWindow.the[].visual_log[msg, log_level=logging.ERROR]
# usually response_json will give full details about the problem
log_msg = f'search on |{search_string}| bad response\n{json.dumps[response_json, indent=4]}'
logger.error[log_msg]
return
# now examine the keys and values in response_json: these may of course
# indicate an error of some kind even though the response returned OK [status 200]...
1 ở trên xuống còn khoảng 3 và thực hiện hầu hết các xử lý của bạn và thông báo nhật ký của bạn được chuẩn hóa. Nhiều hơn một số cuộc gọi requests
trong dự án của bạn và mã trở nên đẹp hơn và ít đầy hơi hơn:def log_response_error[response_type, call_name, deliverable, verb, url, **kwargs]:
# NB this function can also be used independently
if response_type == 'No': # exception was raised [and logged]
if isinstance[deliverable, requests.Timeout]:
MainWindow.the[].visual_log[f'Time out of {call_name} before response received!', logging.ERROR]
return
else:
if isinstance[deliverable, BaseException]:
# NB if response.json[] raises an exception we end up here
log_exception[deliverable, verb, url, kwargs]
else:
# if we get here no exception has been raised, so no stack trace has yet been logged.
# a response has been returned, but is either "Bad" or "Anomalous"
response_json = deliverable.json[]
raw_tb = traceback.extract_stack[]
if 'data' in kwargs and len[kwargs['data']] > 500: # anticipate giant data string
kwargs['data'] = f'{kwargs["data"][:500]}...'
added_message = ''
if hasattr[deliverable, 'added_message']:
added_message = deliverable.added_message + '\n'
del deliverable.added_message
call_and_response_details = f'{response_type} response\n{added_message}' \
+ f'verb {verb}, url {url}, kwargs {kwargs}\nresponse:\n{json.dumps[response_json, indent=4]}'
logger.error[f'{call_and_response_details}\nStack trace: {"".join[traceback.format_list[raw_tb[:-1]]]}']
MainWindow.the[].visual_log[f'{response_type} response {call_name}. See log.', logging.ERROR]
def check_keys[req_dict_structure, response_dict_structure, response]:
# so this function is about checking the keys in the returned json object...
# NB both structures MUST be dicts
if not isinstance[req_dict_structure, dict]:
response.added_message = f'req_dict_structure not dict: {type[req_dict_structure]}\n'
return False
if not isinstance[response_dict_structure, dict]:
response.added_message = f'response_dict_structure not dict: {type[response_dict_structure]}\n'
return False
for dict_key in req_dict_structure.keys[]:
if dict_key not in response_dict_structure:
response.added_message = f'key |{dict_key}| missing\n'
return False
req_value = req_dict_structure[dict_key]
response_value = response_dict_structure[dict_key]
if isinstance[req_value, dict]:
# if the response at this point is a list apply the req_value dict to each element:
# failure in just one such element leads to "Anomalous response"...
if isinstance[response_value, list]:
for resp_list_element in response_value:
if not check_keys[req_value, resp_list_element, response]:
return False
elif not check_keys[req_value, response_value, response]: # any other response value must be a dict [tested in next level of recursion]
return False
elif isinstance[req_value, list]:
if not isinstance[response_value, list]: # if the req_value is a list the reponse must be one
response.added_message = f'key |{dict_key}| not list: {type[response_value]}\n'
return False
# it is OK for the value to be a list, but these must be strings [keys] or dicts
for req_list_element, resp_list_element in zip[req_value, response_value]:
if isinstance[req_list_element, dict]:
if not check_keys[req_list_element, resp_list_element, response]:
return False
if not isinstance[req_list_element, str]:
response.added_message = f'req_list_element not string: {type[req_list_element]}\n'
return False
if req_list_element not in response_value:
response.added_message = f'key |{req_list_element}| missing from response list\n'
return False
# put None as a dummy value [otherwise something like {'my_key'} will be seen as a set, not a dict
elif req_value != None:
response.added_message = f'required value of key |{dict_key}| must be None [dummy], dict or list: {type[req_value]}\n'
return False
return True
def process_json_requests_call[verb, url, **kwargs]:
# "call_name" is a mandatory kwarg
if 'call_name' not in kwargs:
raise Exception['kwarg "call_name" not supplied!']
call_name = kwargs['call_name']
del kwargs['call_name']
required_keys = {}
if 'required_keys' in kwargs:
required_keys = kwargs['required_keys']
del kwargs['required_keys']
acceptable_statuses = [200]
if 'acceptable_statuses' in kwargs:
acceptable_statuses = kwargs['acceptable_statuses']
del kwargs['acceptable_statuses']
exception_handler = log_response_error
if 'exception_handler' in kwargs:
exception_handler = kwargs['exception_handler']
del kwargs['exception_handler']
response, exception = requests_call[verb, url, **kwargs]
if response == None:
exception_handler['No', call_name, exception, verb, url, **kwargs]
return [False, exception]
try:
response_json = response.json[]
except BaseException as e:
logger.error[f'response.status_code {response.status_code} but calling json[] raised exception']
# an exception raised at this point can't truthfully lead to a "No response" message... so say "bad"
exception_handler['Bad', call_name, e, verb, url, **kwargs]
return [False, response]
status_ok = response.status_code in acceptable_statuses
if not status_ok:
response.added_message = f'status code was {response.status_code}'
log_response_error['Bad', call_name, response, verb, url, **kwargs]
return [False, response]
check_result = check_keys[required_keys, response_json, response]
if not check_result:
log_response_error['Anomalous', call_name, response, verb, url, **kwargs]
return [check_result, response]
Gọi ví dụ [NB với phiên bản này, "có thể phân phối" là ngoại lệ hoặc phản hồi cung cấp cấu trúc JSON]:
success, deliverable = utilities.process_json_requests_call['get',
f'{ES_URL}{INDEX_NAME}/_doc/1',
call_name=f'checking index {INDEX_NAME}',
required_keys={'_source':{'status_text': None}}]
if not success: return False
# here, we know the deliverable is a response, not an exception
# we also don't need to check for the keys being present:
# the generic code has checked that all expected keys are present
index_status = deliverable.json[]['_source']['status_text']
if index_status != 'successfully completed':
# ... i.e. an example of a 200 response, but an error nonetheless
msg = f'Error response: ES index {INDEX_NAME} does not seem to have been built OK: cannot search'
MainWindow.the[].visual_log[msg]
logger.error[f'index |{INDEX_NAME}|: deliverable.json[] {json.dumps[deliverable.json[], indent=4]}']
return False
Vì vậy, thông báo "Nhật ký trực quan" được người dùng nhìn thấy trong trường hợp thiếu khóa "status_text", chẳng hạn, sẽ là "chỉ mục kiểm tra phản hồi bất thường XYZ. Xem nhật ký." .
NB
- Kwarg bắt buộc:
3; Kwargs tùy chọn:if search_response == None: # you might check here for [e.g.] a requests.Timeout, tailoring the message # accordingly, as the kind of error anyone might be expected to understand msg = f'No response searching on |{search_string}|. See log' MainWindow.the[].visual_log[msg, log_level=logging.ERROR] return response_json = search_response.json[] if search_response.status_code != 200: # NB 201 ["created"] may be acceptable sometimes... msg = f'Bad response searching on |{search_string}|. See log' MainWindow.the[].visual_log[msg, log_level=logging.ERROR] # usually response_json will give full details about the problem log_msg = f'search on |{search_string}| bad response\n{json.dumps[response_json, indent=4]}' logger.error[log_msg] return # now examine the keys and values in response_json: these may of course # indicate an error of some kind even though the response returned OK [status 200]...
4,if search_response == None: # you might check here for [e.g.] a requests.Timeout, tailoring the message # accordingly, as the kind of error anyone might be expected to understand msg = f'No response searching on |{search_string}|. See log' MainWindow.the[].visual_log[msg, log_level=logging.ERROR] return response_json = search_response.json[] if search_response.status_code != 200: # NB 201 ["created"] may be acceptable sometimes... msg = f'Bad response searching on |{search_string}|. See log' MainWindow.the[].visual_log[msg, log_level=logging.ERROR] # usually response_json will give full details about the problem log_msg = f'search on |{search_string}| bad response\n{json.dumps[response_json, indent=4]}' logger.error[log_msg] return # now examine the keys and values in response_json: these may of course # indicate an error of some kind even though the response returned OK [status 200]...
5,if search_response == None: # you might check here for [e.g.] a requests.Timeout, tailoring the message # accordingly, as the kind of error anyone might be expected to understand msg = f'No response searching on |{search_string}|. See log' MainWindow.the[].visual_log[msg, log_level=logging.ERROR] return response_json = search_response.json[] if search_response.status_code != 200: # NB 201 ["created"] may be acceptable sometimes... msg = f'Bad response searching on |{search_string}|. See log' MainWindow.the[].visual_log[msg, log_level=logging.ERROR] # usually response_json will give full details about the problem log_msg = f'search on |{search_string}| bad response\n{json.dumps[response_json, indent=4]}' logger.error[log_msg] return # now examine the keys and values in response_json: these may of course # indicate an error of some kind even though the response returned OK [status 200]...
6.if search_response == None: # you might check here for [e.g.] a requests.Timeout, tailoring the message # accordingly, as the kind of error anyone might be expected to understand msg = f'No response searching on |{search_string}|. See log' MainWindow.the[].visual_log[msg, log_level=logging.ERROR] return response_json = search_response.json[] if search_response.status_code != 200: # NB 201 ["created"] may be acceptable sometimes... msg = f'Bad response searching on |{search_string}|. See log' MainWindow.the[].visual_log[msg, log_level=logging.ERROR] # usually response_json will give full details about the problem log_msg = f'search on |{search_string}| bad response\n{json.dumps[response_json, indent=4]}' logger.error[log_msg] return # now examine the keys and values in response_json: these may of course # indicate an error of some kind even though the response returned OK [status 200]...
4if search_response == None: # you might check here for [e.g.] a requests.Timeout, tailoring the message # accordingly, as the kind of error anyone might be expected to understand msg = f'No response searching on |{search_string}|. See log' MainWindow.the[].visual_log[msg, log_level=logging.ERROR] return response_json = search_response.json[] if search_response.status_code != 200: # NB 201 ["created"] may be acceptable sometimes... msg = f'Bad response searching on |{search_string}|. See log' MainWindow.the[].visual_log[msg, log_level=logging.ERROR] # usually response_json will give full details about the problem log_msg = f'search on |{search_string}| bad response\n{json.dumps[response_json, indent=4]}' logger.error[log_msg] return # now examine the keys and values in response_json: these may of course # indicate an error of some kind even though the response returned OK [status 200]...
8 có thể được lồng đến bất kỳ độ sâu nàoif search_response == None: # you might check here for [e.g.] a requests.Timeout, tailoring the message # accordingly, as the kind of error anyone might be expected to understand msg = f'No response searching on |{search_string}|. See log' MainWindow.the[].visual_log[msg, log_level=logging.ERROR] return response_json = search_response.json[] if search_response.status_code != 200: # NB 201 ["created"] may be acceptable sometimes... msg = f'Bad response searching on |{search_string}|. See log' MainWindow.the[].visual_log[msg, log_level=logging.ERROR] # usually response_json will give full details about the problem log_msg = f'search on |{search_string}| bad response\n{json.dumps[response_json, indent=4]}' logger.error[log_msg] return # now examine the keys and values in response_json: these may of course # indicate an error of some kind even though the response returned OK [status 200]...
- Có thể xử lý ngoại lệ hạt mịn hơn bằng cách bao gồm một hàm
6 trongif search_response == None: # you might check here for [e.g.] a requests.Timeout, tailoring the message # accordingly, as the kind of error anyone might be expected to understand msg = f'No response searching on |{search_string}|. See log' MainWindow.the[].visual_log[msg, log_level=logging.ERROR] return response_json = search_response.json[] if search_response.status_code != 200: # NB 201 ["created"] may be acceptable sometimes... msg = f'Bad response searching on |{search_string}|. See log' MainWindow.the[].visual_log[msg, log_level=logging.ERROR] # usually response_json will give full details about the problem log_msg = f'search on |{search_string}| bad response\n{json.dumps[response_json, indent=4]}' logger.error[log_msg] return # now examine the keys and values in response_json: these may of course # indicate an error of some kind even though the response returned OK [status 200]...
0 [mặc dù đừng quên rằngdef log_response_error[response_type, call_name, deliverable, verb, url, **kwargs]: # NB this function can also be used independently if response_type == 'No': # exception was raised [and logged] if isinstance[deliverable, requests.Timeout]: MainWindow.the[].visual_log[f'Time out of {call_name} before response received!', logging.ERROR] return else: if isinstance[deliverable, BaseException]: # NB if response.json[] raises an exception we end up here log_exception[deliverable, verb, url, kwargs] else: # if we get here no exception has been raised, so no stack trace has yet been logged. # a response has been returned, but is either "Bad" or "Anomalous" response_json = deliverable.json[] raw_tb = traceback.extract_stack[] if 'data' in kwargs and len[kwargs['data']] > 500: # anticipate giant data string kwargs['data'] = f'{kwargs["data"][:500]}...' added_message = '' if hasattr[deliverable, 'added_message']: added_message = deliverable.added_message + '\n' del deliverable.added_message call_and_response_details = f'{response_type} response\n{added_message}' \ + f'verb {verb}, url {url}, kwargs {kwargs}\nresponse:\n{json.dumps[response_json, indent=4]}' logger.error[f'{call_and_response_details}\nStack trace: {"".join[traceback.format_list[raw_tb[:-1]]]}'] MainWindow.the[].visual_log[f'{response_type} response {call_name}. See log.', logging.ERROR] def check_keys[req_dict_structure, response_dict_structure, response]: # so this function is about checking the keys in the returned json object... # NB both structures MUST be dicts if not isinstance[req_dict_structure, dict]: response.added_message = f'req_dict_structure not dict: {type[req_dict_structure]}\n' return False if not isinstance[response_dict_structure, dict]: response.added_message = f'response_dict_structure not dict: {type[response_dict_structure]}\n' return False for dict_key in req_dict_structure.keys[]: if dict_key not in response_dict_structure: response.added_message = f'key |{dict_key}| missing\n' return False req_value = req_dict_structure[dict_key] response_value = response_dict_structure[dict_key] if isinstance[req_value, dict]: # if the response at this point is a list apply the req_value dict to each element: # failure in just one such element leads to "Anomalous response"... if isinstance[response_value, list]: for resp_list_element in response_value: if not check_keys[req_value, resp_list_element, response]: return False elif not check_keys[req_value, response_value, response]: # any other response value must be a dict [tested in next level of recursion] return False elif isinstance[req_value, list]: if not isinstance[response_value, list]: # if the req_value is a list the reponse must be one response.added_message = f'key |{dict_key}| not list: {type[response_value]}\n' return False # it is OK for the value to be a list, but these must be strings [keys] or dicts for req_list_element, resp_list_element in zip[req_value, response_value]: if isinstance[req_list_element, dict]: if not check_keys[req_list_element, resp_list_element, response]: return False if not isinstance[req_list_element, str]: response.added_message = f'req_list_element not string: {type[req_list_element]}\n' return False if req_list_element not in response_value: response.added_message = f'key |{req_list_element}| missing from response list\n' return False # put None as a dummy value [otherwise something like {'my_key'} will be seen as a set, not a dict elif req_value != None: response.added_message = f'required value of key |{dict_key}| must be None [dummy], dict or list: {type[req_value]}\n' return False return True def process_json_requests_call[verb, url, **kwargs]: # "call_name" is a mandatory kwarg if 'call_name' not in kwargs: raise Exception['kwarg "call_name" not supplied!'] call_name = kwargs['call_name'] del kwargs['call_name'] required_keys = {} if 'required_keys' in kwargs: required_keys = kwargs['required_keys'] del kwargs['required_keys'] acceptable_statuses = [200] if 'acceptable_statuses' in kwargs: acceptable_statuses = kwargs['acceptable_statuses'] del kwargs['acceptable_statuses'] exception_handler = log_response_error if 'exception_handler' in kwargs: exception_handler = kwargs['exception_handler'] del kwargs['exception_handler'] response, exception = requests_call[verb, url, **kwargs] if response == None: exception_handler['No', call_name, exception, verb, url, **kwargs] return [False, exception] try: response_json = response.json[] except BaseException as e: logger.error[f'response.status_code {response.status_code} but calling json[] raised exception'] # an exception raised at this point can't truthfully lead to a "No response" message... so say "bad" exception_handler['Bad', call_name, e, verb, url, **kwargs] return [False, response] status_ok = response.status_code in acceptable_statuses if not status_ok: response.added_message = f'status code was {response.status_code}' log_response_error['Bad', call_name, response, verb, url, **kwargs] return [False, response] check_result = check_keys[required_keys, response_json, response] if not check_result: log_response_error['Anomalous', call_name, response, verb, url, **kwargs] return [check_result, response]
1 sẽ ghi lại các chi tiết cuộc gọi, loại ngoại lệ vàif search_response == None: # you might check here for [e.g.] a requests.Timeout, tailoring the message # accordingly, as the kind of error anyone might be expected to understand msg = f'No response searching on |{search_string}|. See log' MainWindow.the[].visual_log[msg, log_level=logging.ERROR] return response_json = search_response.json[] if search_response.status_code != 200: # NB 201 ["created"] may be acceptable sometimes... msg = f'Bad response searching on |{search_string}|. See log' MainWindow.the[].visual_log[msg, log_level=logging.ERROR] # usually response_json will give full details about the problem log_msg = f'search on |{search_string}| bad response\n{json.dumps[response_json, indent=4]}' logger.error[log_msg] return # now examine the keys and values in response_json: these may of course # indicate an error of some kind even though the response returned OK [status 200]...
2 và dấu vết ngăn xếp].def log_response_error[response_type, call_name, deliverable, verb, url, **kwargs]: # NB this function can also be used independently if response_type == 'No': # exception was raised [and logged] if isinstance[deliverable, requests.Timeout]: MainWindow.the[].visual_log[f'Time out of {call_name} before response received!', logging.ERROR] return else: if isinstance[deliverable, BaseException]: # NB if response.json[] raises an exception we end up here log_exception[deliverable, verb, url, kwargs] else: # if we get here no exception has been raised, so no stack trace has yet been logged. # a response has been returned, but is either "Bad" or "Anomalous" response_json = deliverable.json[] raw_tb = traceback.extract_stack[] if 'data' in kwargs and len[kwargs['data']] > 500: # anticipate giant data string kwargs['data'] = f'{kwargs["data"][:500]}...' added_message = '' if hasattr[deliverable, 'added_message']: added_message = deliverable.added_message + '\n' del deliverable.added_message call_and_response_details = f'{response_type} response\n{added_message}' \ + f'verb {verb}, url {url}, kwargs {kwargs}\nresponse:\n{json.dumps[response_json, indent=4]}' logger.error[f'{call_and_response_details}\nStack trace: {"".join[traceback.format_list[raw_tb[:-1]]]}'] MainWindow.the[].visual_log[f'{response_type} response {call_name}. See log.', logging.ERROR] def check_keys[req_dict_structure, response_dict_structure, response]: # so this function is about checking the keys in the returned json object... # NB both structures MUST be dicts if not isinstance[req_dict_structure, dict]: response.added_message = f'req_dict_structure not dict: {type[req_dict_structure]}\n' return False if not isinstance[response_dict_structure, dict]: response.added_message = f'response_dict_structure not dict: {type[response_dict_structure]}\n' return False for dict_key in req_dict_structure.keys[]: if dict_key not in response_dict_structure: response.added_message = f'key |{dict_key}| missing\n' return False req_value = req_dict_structure[dict_key] response_value = response_dict_structure[dict_key] if isinstance[req_value, dict]: # if the response at this point is a list apply the req_value dict to each element: # failure in just one such element leads to "Anomalous response"... if isinstance[response_value, list]: for resp_list_element in response_value: if not check_keys[req_value, resp_list_element, response]: return False elif not check_keys[req_value, response_value, response]: # any other response value must be a dict [tested in next level of recursion] return False elif isinstance[req_value, list]: if not isinstance[response_value, list]: # if the req_value is a list the reponse must be one response.added_message = f'key |{dict_key}| not list: {type[response_value]}\n' return False # it is OK for the value to be a list, but these must be strings [keys] or dicts for req_list_element, resp_list_element in zip[req_value, response_value]: if isinstance[req_list_element, dict]: if not check_keys[req_list_element, resp_list_element, response]: return False if not isinstance[req_list_element, str]: response.added_message = f'req_list_element not string: {type[req_list_element]}\n' return False if req_list_element not in response_value: response.added_message = f'key |{req_list_element}| missing from response list\n' return False # put None as a dummy value [otherwise something like {'my_key'} will be seen as a set, not a dict elif req_value != None: response.added_message = f'required value of key |{dict_key}| must be None [dummy], dict or list: {type[req_value]}\n' return False return True def process_json_requests_call[verb, url, **kwargs]: # "call_name" is a mandatory kwarg if 'call_name' not in kwargs: raise Exception['kwarg "call_name" not supplied!'] call_name = kwargs['call_name'] del kwargs['call_name'] required_keys = {} if 'required_keys' in kwargs: required_keys = kwargs['required_keys'] del kwargs['required_keys'] acceptable_statuses = [200] if 'acceptable_statuses' in kwargs: acceptable_statuses = kwargs['acceptable_statuses'] del kwargs['acceptable_statuses'] exception_handler = log_response_error if 'exception_handler' in kwargs: exception_handler = kwargs['exception_handler'] del kwargs['exception_handler'] response, exception = requests_call[verb, url, **kwargs] if response == None: exception_handler['No', call_name, exception, verb, url, **kwargs] return [False, exception] try: response_json = response.json[] except BaseException as e: logger.error[f'response.status_code {response.status_code} but calling json[] raised exception'] # an exception raised at this point can't truthfully lead to a "No response" message... so say "bad" exception_handler['Bad', call_name, e, verb, url, **kwargs] return [False, response] status_ok = response.status_code in acceptable_statuses if not status_ok: response.added_message = f'status code was {response.status_code}' log_response_error['Bad', call_name, response, verb, url, **kwargs] return [False, response] check_result = check_keys[required_keys, response_json, response] if not check_result: log_response_error['Anomalous', call_name, response, verb, url, **kwargs] return [check_result, response]
- Ở trên, tôi cũng thực hiện kiểm tra "dữ liệu" khóa trong bất kỳ
0 nào có thể được ghi lại. Điều này là do một hoạt động số lượng lớn [ví dụ: để điền vào một chỉ số trong trường hợp của Elaticsearch] có thể bao gồm các chuỗi khổng lồ. Vì vậy, hạn chế đến 500 ký tự đầu tiên, ví dụ.enormous strings. So curtail to the first 500 characters, for example.def log_response_error[response_type, call_name, deliverable, verb, url, **kwargs]: # NB this function can also be used independently if response_type == 'No': # exception was raised [and logged] if isinstance[deliverable, requests.Timeout]: MainWindow.the[].visual_log[f'Time out of {call_name} before response received!', logging.ERROR] return else: if isinstance[deliverable, BaseException]: # NB if response.json[] raises an exception we end up here log_exception[deliverable, verb, url, kwargs] else: # if we get here no exception has been raised, so no stack trace has yet been logged. # a response has been returned, but is either "Bad" or "Anomalous" response_json = deliverable.json[] raw_tb = traceback.extract_stack[] if 'data' in kwargs and len[kwargs['data']] > 500: # anticipate giant data string kwargs['data'] = f'{kwargs["data"][:500]}...' added_message = '' if hasattr[deliverable, 'added_message']: added_message = deliverable.added_message + '\n' del deliverable.added_message call_and_response_details = f'{response_type} response\n{added_message}' \ + f'verb {verb}, url {url}, kwargs {kwargs}\nresponse:\n{json.dumps[response_json, indent=4]}' logger.error[f'{call_and_response_details}\nStack trace: {"".join[traceback.format_list[raw_tb[:-1]]]}'] MainWindow.the[].visual_log[f'{response_type} response {call_name}. See log.', logging.ERROR] def check_keys[req_dict_structure, response_dict_structure, response]: # so this function is about checking the keys in the returned json object... # NB both structures MUST be dicts if not isinstance[req_dict_structure, dict]: response.added_message = f'req_dict_structure not dict: {type[req_dict_structure]}\n' return False if not isinstance[response_dict_structure, dict]: response.added_message = f'response_dict_structure not dict: {type[response_dict_structure]}\n' return False for dict_key in req_dict_structure.keys[]: if dict_key not in response_dict_structure: response.added_message = f'key |{dict_key}| missing\n' return False req_value = req_dict_structure[dict_key] response_value = response_dict_structure[dict_key] if isinstance[req_value, dict]: # if the response at this point is a list apply the req_value dict to each element: # failure in just one such element leads to "Anomalous response"... if isinstance[response_value, list]: for resp_list_element in response_value: if not check_keys[req_value, resp_list_element, response]: return False elif not check_keys[req_value, response_value, response]: # any other response value must be a dict [tested in next level of recursion] return False elif isinstance[req_value, list]: if not isinstance[response_value, list]: # if the req_value is a list the reponse must be one response.added_message = f'key |{dict_key}| not list: {type[response_value]}\n' return False # it is OK for the value to be a list, but these must be strings [keys] or dicts for req_list_element, resp_list_element in zip[req_value, response_value]: if isinstance[req_list_element, dict]: if not check_keys[req_list_element, resp_list_element, response]: return False if not isinstance[req_list_element, str]: response.added_message = f'req_list_element not string: {type[req_list_element]}\n' return False if req_list_element not in response_value: response.added_message = f'key |{req_list_element}| missing from response list\n' return False # put None as a dummy value [otherwise something like {'my_key'} will be seen as a set, not a dict elif req_value != None: response.added_message = f'required value of key |{dict_key}| must be None [dummy], dict or list: {type[req_value]}\n' return False return True def process_json_requests_call[verb, url, **kwargs]: # "call_name" is a mandatory kwarg if 'call_name' not in kwargs: raise Exception['kwarg "call_name" not supplied!'] call_name = kwargs['call_name'] del kwargs['call_name'] required_keys = {} if 'required_keys' in kwargs: required_keys = kwargs['required_keys'] del kwargs['required_keys'] acceptable_statuses = [200] if 'acceptable_statuses' in kwargs: acceptable_statuses = kwargs['acceptable_statuses'] del kwargs['acceptable_statuses'] exception_handler = log_response_error if 'exception_handler' in kwargs: exception_handler = kwargs['exception_handler'] del kwargs['exception_handler'] response, exception = requests_call[verb, url, **kwargs] if response == None: exception_handler['No', call_name, exception, verb, url, **kwargs] return [False, exception] try: response_json = response.json[] except BaseException as e: logger.error[f'response.status_code {response.status_code} but calling json[] raised exception'] # an exception raised at this point can't truthfully lead to a "No response" message... so say "bad" exception_handler['Bad', call_name, e, verb, url, **kwargs] return [False, response] status_ok = response.status_code in acceptable_statuses if not status_ok: response.added_message = f'status code was {response.status_code}' log_response_error['Bad', call_name, response, verb, url, **kwargs] return [False, response] check_result = check_keys[required_keys, response_json, response] if not check_result: log_response_error['Anomalous', call_name, response, verb, url, **kwargs] return [check_result, response]
Tái bút Vâng, tôi biết về mô -đun Python
def log_response_error[response_type, call_name, deliverable, verb, url, **kwargs]:
# NB this function can also be used independently
if response_type == 'No': # exception was raised [and logged]
if isinstance[deliverable, requests.Timeout]:
MainWindow.the[].visual_log[f'Time out of {call_name} before response received!', logging.ERROR]
return
else:
if isinstance[deliverable, BaseException]:
# NB if response.json[] raises an exception we end up here
log_exception[deliverable, verb, url, kwargs]
else:
# if we get here no exception has been raised, so no stack trace has yet been logged.
# a response has been returned, but is either "Bad" or "Anomalous"
response_json = deliverable.json[]
raw_tb = traceback.extract_stack[]
if 'data' in kwargs and len[kwargs['data']] > 500: # anticipate giant data string
kwargs['data'] = f'{kwargs["data"][:500]}...'
added_message = ''
if hasattr[deliverable, 'added_message']:
added_message = deliverable.added_message + '\n'
del deliverable.added_message
call_and_response_details = f'{response_type} response\n{added_message}' \
+ f'verb {verb}, url {url}, kwargs {kwargs}\nresponse:\n{json.dumps[response_json, indent=4]}'
logger.error[f'{call_and_response_details}\nStack trace: {"".join[traceback.format_list[raw_tb[:-1]]]}']
MainWindow.the[].visual_log[f'{response_type} response {call_name}. See log.', logging.ERROR]
def check_keys[req_dict_structure, response_dict_structure, response]:
# so this function is about checking the keys in the returned json object...
# NB both structures MUST be dicts
if not isinstance[req_dict_structure, dict]:
response.added_message = f'req_dict_structure not dict: {type[req_dict_structure]}\n'
return False
if not isinstance[response_dict_structure, dict]:
response.added_message = f'response_dict_structure not dict: {type[response_dict_structure]}\n'
return False
for dict_key in req_dict_structure.keys[]:
if dict_key not in response_dict_structure:
response.added_message = f'key |{dict_key}| missing\n'
return False
req_value = req_dict_structure[dict_key]
response_value = response_dict_structure[dict_key]
if isinstance[req_value, dict]:
# if the response at this point is a list apply the req_value dict to each element:
# failure in just one such element leads to "Anomalous response"...
if isinstance[response_value, list]:
for resp_list_element in response_value:
if not check_keys[req_value, resp_list_element, response]:
return False
elif not check_keys[req_value, response_value, response]: # any other response value must be a dict [tested in next level of recursion]
return False
elif isinstance[req_value, list]:
if not isinstance[response_value, list]: # if the req_value is a list the reponse must be one
response.added_message = f'key |{dict_key}| not list: {type[response_value]}\n'
return False
# it is OK for the value to be a list, but these must be strings [keys] or dicts
for req_list_element, resp_list_element in zip[req_value, response_value]:
if isinstance[req_list_element, dict]:
if not check_keys[req_list_element, resp_list_element, response]:
return False
if not isinstance[req_list_element, str]:
response.added_message = f'req_list_element not string: {type[req_list_element]}\n'
return False
if req_list_element not in response_value:
response.added_message = f'key |{req_list_element}| missing from response list\n'
return False
# put None as a dummy value [otherwise something like {'my_key'} will be seen as a set, not a dict
elif req_value != None:
response.added_message = f'required value of key |{dict_key}| must be None [dummy], dict or list: {type[req_value]}\n'
return False
return True
def process_json_requests_call[verb, url, **kwargs]:
# "call_name" is a mandatory kwarg
if 'call_name' not in kwargs:
raise Exception['kwarg "call_name" not supplied!']
call_name = kwargs['call_name']
del kwargs['call_name']
required_keys = {}
if 'required_keys' in kwargs:
required_keys = kwargs['required_keys']
del kwargs['required_keys']
acceptable_statuses = [200]
if 'acceptable_statuses' in kwargs:
acceptable_statuses = kwargs['acceptable_statuses']
del kwargs['acceptable_statuses']
exception_handler = log_response_error
if 'exception_handler' in kwargs:
exception_handler = kwargs['exception_handler']
del kwargs['exception_handler']
response, exception = requests_call[verb, url, **kwargs]
if response == None:
exception_handler['No', call_name, exception, verb, url, **kwargs]
return [False, exception]
try:
response_json = response.json[]
except BaseException as e:
logger.error[f'response.status_code {response.status_code} but calling json[] raised exception']
# an exception raised at this point can't truthfully lead to a "No response" message... so say "bad"
exception_handler['Bad', call_name, e, verb, url, **kwargs]
return [False, response]
status_ok = response.status_code in acceptable_statuses
if not status_ok:
response.added_message = f'status code was {response.status_code}'
log_response_error['Bad', call_name, response, verb, url, **kwargs]
return [False, response]
check_result = check_keys[required_keys, response_json, response]
if not check_result:
log_response_error['Anomalous', call_name, response, verb, url, **kwargs]
return [check_result, response]
4 [một "trình bao bọc mỏng" xung quanh requests
]. Tất cả những điều trên là cho mục đích minh họa.