Hướng dẫn try except requests python - thử ngoại trừ yêu cầu python

Question

Đâ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:

Nội dung chính Show

Phiên bản cơ bản
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!)

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ề,
```
search_response, exception = utilities.requests_call('get',
    f'http://localhost:9200/my_index/_search?q={search_string}')
```
0, siêu lớp của tất cả các ngoại lệ requests (bao gồm requests.ConnectionError), không phải là "
```
search_response, exception = utilities.requests_call('get',
    f'http://localhost:9200/my_index/_search?q={search_string}')
```
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. **
Rõ ràng điều này giả định một logger đã được cấu hình. Gọi
```
search_response, exception = utilities.requests_call('get',
    f'http://localhost:9200/my_index/_search?q={search_string}')
```
4 trong khối
```
search_response, exception = utilities.requests_call('get',
    f'http://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)

*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'http://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'http://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ị

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)...

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"

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:

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)...

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)...

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)...

8 có thể được lồng đến bất kỳ độ sâu nào

Có thể xử lý ngoại lệ hạt mịn hơn bằng cách bao gồm một hàm

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 trong

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)

0 (mặc dù đừng quên rằ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 sẽ ghi lại các chi tiết cuộc gọi, loại ngoại lệ và

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)

2 và dấu vết ngăn xếp).

Ở trên, tôi cũng thực hiện kiểm tra "dữ liệu" khóa trong bất kỳ

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)

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.

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.

programming python Python request HTTPError Python requests retry Python requests timeout Python requests headers

Hướng dẫn try except requests python - thử ngoại trừ yêu cầu python

Phiên bản cơ bản

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!)

Bài Viết Liên Quan

Quảng Cáo

Có thể bạn quan tâm

Toplist được quan tâm

Quảng cáo

Xem Nhiều

Quảng cáo

Chúng tôi

Điều khoản

Trợ giúp

Mạng xã hội