파이썬 요청 모듈을 사용하는 것을 제외하고 시도하는 올바른 방법은 무엇입니까?

파이썬 요청 모듈을 사용하는 것을 제외하고 시도하는 올바른 방법은 무엇입니까?

try:
    r = requests.get(url, params={'s': thing})
except requests.ConnectionError, e:
    print(e)

이거 맞는건가요? 이것을 구성하는 더 좋은 방법이 있을까요? 이것이 내 모든 기지를 커버할 수 있을까?

---

Requests를 확인하십시오. 간단히 말해서:

네트워크 문제(예: DNS 오류, 연결 거부 등)가 발생하면 요청에서 예외를 발생시킵니다.

드물게 잘못된 HTTP 응답이 발생할 경우 요청은 예외를 발생시킵니다.

요청 시간이 초과되면 예외가 발생합니다.

요청이 구성된 최대 리디렉션 수를 초과하면 예외가 발생합니다.

요청이 명시적으로 제기하는 모든 예외는 에서 상속됩니다.

당신의 질문에 대답하자면, 당신이 보여주는 것은 당신의 모든 근거를 포함할 것이다. 연결 관련 오류만 발생하고 시간 초과된 오류는 발생하지 않습니다.

예외가 발생했을 때 수행할 작업은 스크립트/프로그램 설계에 달려 있습니다. 퇴장해도 되나요? 계속해서 다시 시도할 수 있습니까? 오류가 치명적이고 계속할 수 없는 경우에는 오류를 인쇄하고 호출하는 좋은 방법으로 프로그램을 중단할 수 있습니다.

모든 경우를 처리하는 기본 클래스 예외를 잡을 수 있습니다:

try:
    r = requests.get(url, params={'s': thing})
except requests.exceptions.RequestException as e:  # This is the correct syntax
    raise SystemExit(e)

아니면 따로따로 잡아서 다른 일을 할 수도 있어요.

try:
    r = requests.get(url, params={'s': thing})
except requests.exceptions.Timeout:
    # Maybe set up for a retry, or continue in a retry loop
except requests.exceptions.TooManyRedirects:
    # Tell the user their URL was bad and try a different one
except requests.exceptions.RequestException as e:
    # catastrophic error. bail.
    raise SystemExit(e)

---

지적된 바와 같이:

http 오류(예: 401 Unauthorized)로 예외를 발생시키려면 를 호출할 수 있습니다. 응답이 http 오류인 경우 를 발생시킵니다.

예:

try:
    r = requests.get('http://www.google.com/nothere')
    r.raise_for_status()
except requests.exceptions.HTTPError as err:
    raise SystemExit(err)

인쇄 예정:

404 Client Error: Not Found for url: http://www.google.com/nothere

---

한 가지 더 명시적인 제안이 있습니다. 특정 오류가 일반 오류에 의해 가려지지 않도록 특정 오류에서 일반 오류로 이어지는 것이 가장 좋습니다.

url='http://www.google.com/blahblah'

try:
    r = requests.get(url,timeout=3)
    r.raise_for_status()
except requests.exceptions.HTTPError as errh:
    print ("Http Error:",errh)
except requests.exceptions.ConnectionError as errc:
    print ("Error Connecting:",errc)
except requests.exceptions.Timeout as errt:
    print ("Timeout Error:",errt)
except requests.exceptions.RequestException as err:
    print ("OOps: Something Else",err)

Http Error: 404 Client Error: Not Found for url: http://www.google.com/blahblah

대

url='http://www.google.com/blahblah'

try:
    r = requests.get(url,timeout=3)
    r.raise_for_status()
except requests.exceptions.RequestException as err:
    print ("OOps: Something Else",err)
except requests.exceptions.HTTPError as errh:
    print ("Http Error:",errh)
except requests.exceptions.ConnectionError as errc:
    print ("Error Connecting:",errc)
except requests.exceptions.Timeout as errt:
    print ("Timeout Error:",errt)     

OOps: Something Else 404 Client Error: Not Found for url: http://www.google.com/blahblah

---

또한 예외 개체에는 원래 응답이 포함되어 있으며, 서버의 응답에서 오류 본문을 확인해야 할 경우 유용합니다. 예:

try:
    r = requests.post('somerestapi.com/post-here', data={'birthday': '9/9/3999'})
    r.raise_for_status()
except requests.exceptions.HTTPError as e:
    print (e.response.text)

---

다음은 일반적인 방법으로 모든 통화를 다음과 같이 처리할 필요가 없다는 것을 의미합니다:

기본 버전

# 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

1. 클래스와 무관한 것을 알아두십시오. *. 후자가 이러한 맥락에서 더 일반적이라고 생각하지만 실질적인 개념은 없습니다...
2. 반환되지 않은 예외를 검사할 때, 를 포함한 모든 예외의 상위 클래스는 에 해당되지 않습니다. 이후로 변경되었을 수 있습니다.**
3. 분명히 이것은 로거가 구성되었다고 가정한다. 블록을 호출하는 것이 좋은 생각처럼 보일 수 있지만, 그것은 이 방법 내에서 스택을 제공할 뿐입니다! 대신 이 메서드에 대한 호출을 유도하는 추적을 가져옵니다. 그런 다음 로그 기록(예외 및 문제를 일으킨 통화에 대한 세부 정보 포함)

*소스 코드를 살펴봤습니다: 단일 클래스를 하위 클래스로 분류하는 단일 클래스(기본 제공)

**하지만 하단에 "요청"이 있습니다.예외 사항.예외 요청"을 작성할 때(2022-02)... 하지만 그것은 위의 페이지로 연결된다: 혼란스럽다.

---

사용법은 매우 간단합니다:

search_response, exception = utilities.requests_call('get',
    f'http://localhost:9200/my_index/_search?q={search_string}')

먼저 반응을 확인합니다. 재미있는 일이 발생하여 상황(및 예외)에 따라 어떤 방식으로든 적용되어야 하는 예외가 발생할 경우.

고급 버전, json 개체가 반환될 때 특히 유용합니다

(... 잠재적으로 보일러 플레이트를 많이 절약할 수 있습니다!)

위와 관련하여 몇 가지가 지루합니다. 1) 하나 이상의 상태 코드만 허용된다고 어떻게 규정합니까? 2) 반환된 JSON 딕트가 특정 구조의 키와 하위 사전을 포함해야 한다고 어떻게 지정합니까? 예를 들어 Elastic search의 경우 JSON 개체가 반환되는 경우가 많으며 값을 얻기 전에 모든 키가 실제로 존재하는지 확인하는 것은 지루합니다.

따라서 이것은 위의 간단한 기능에 기초한다.

def process_request(url, verb='get', permitted_status_codes=200, required_dict=None, **kwargs):
    success, deliverable = requests_call(verb, url, **kwargs)
    if not success:
        deliverable.failure_reason = 'requests_call returned False'
        deliverable.failure_code = 1
        return (False, deliverable)
    response = deliverable
    # NB `permitted_status_codes` can either be a list or an int
    if isinstance(permitted_status_codes, list):
        good_code_found = False
        for permitted_status_code in permitted_status_codes:
            if deliverable.status_code == permitted_status_code:
                good_code_found = True
        if not good_code_found:
            deliverable.failure_reason = f'unacceptable status code: {deliverable.status_code}'
            deliverable.failure_code = 2
            return (False, deliverable)
    elif isinstance(permitted_status_codes, int):
        if deliverable.status_code != permitted_status_codes:
            deliverable.failure_reason = f'unacceptable status code: {deliverable.status_code}'
            deliverable.failure_code = 2
            return (False, deliverable)
    else:
        raise Exception(f'permitted_status_codes must be list or int, not {type(permitted_status_codes)}')
    if required_dict != None:
        if type(required_dict) != dict:
            raise Exception(f'required_dict must be dict, not {type(required_dict)}')
        delivered_json_dict = response.json()
        def check_dictionary_key_values(required_dict, comparison_dict):
            all_checks_passed = True
            for key, value in required_dict.items():
                if key not in comparison_dict:
                    logger.error(f'key absent: {key}')
                    all_checks_passed = False
                    break
                if type(required_dict[key]) == dict:
                    # logger.info(f'required subdictionary {required_dict[key]}')
                    sub_comparison_dict = comparison_dict[key]
                    if type(sub_comparison_dict) != dict:
                        logger.error(f'key which yields subdictionary in required does not in comparison: {sub_comparison_dict}')
                        all_checks_passed = False
                        break
                    if not check_dictionary_key_values(required_dict[key], sub_comparison_dict):
                        # logger.error(f'... so subdictionary comparison failed')
                        all_checks_passed = False
                        break
                # if a value of "None" is given for a key this means "can be anything"
                elif value != None and comparison_dict[key] != value:
                    logger.error(f'key {key} was found as expected but value {value} was not found, instead: {comparison_dict[key]}')
                    all_checks_passed = False
                    break
            return all_checks_passed
        if not check_dictionary_key_values(required_dict, delivered_json_dict):
            deliverable.failure_reason = f'delivered JSON\n{json.dumps(delivered_json_dict, indent=2)}\ndid not satisfy required_dict\n{json.dumps(required_dict, indent=2)}'
            deliverable.failure_code = 3
            return (False, deliverable)
    return (True, response)            

사용 예 1:

required_dict = {
    'found': True,
    '_source': {
        'es_version': None,
        
    },
    # 'cats_and_dogs': 'shanty town',
}
success, deliverable = process_request(f'{url_for_specific_index}/_doc/1', required_dict=required_dict)    
if not success:    
    logger.error(f'failed to get Ldoc 1 from index... deliverable.failure_reason:\n{deliverable.failure_reason}')
    ...
returned_dict = deliverable.json()
es_version_from_status_doc = returned_dict['_source']['es_version']

내가 "cats_and_dogs" 줄에 주석을 달면 이 필수 키가 없기 때문에 반환됩니다. 반대로 required_dict 검사를 통과하면 "_source"와 "es_version"이 sexict를 생성하지 않을 수 있습니다. 또한 "찾은" 키에 값이 있다는 것도 알고 있습니다.

NB 규정 키와 값은 임의의 깊이로 중첩될 수 있으며, 값 "없음"은 "전달된 딕트의 이 위치에 키가 있는지 확인하는 것 외에는 아무 것도 될 수 없음"을 의미한다. 코드가 재귀 함수의 s를 처리하도록 하는 것도 가능하지만, 이것은 여기에 포함시키기에는 상황을 너무 복잡하게 만들고, 틀림없이 매우 미미한 요구사항이 될 것이다.

사용 예 2:

이것은 "get" 이외의 동사를 사용하며, 새로운 자원을 만들기 때문에 상태 코드는 200이 아니라 201이어야 한다. NB 매개 변수는 또는 a일 수 있다(함수 참조).

data = {...}
headers = {'Content-type': 'application/json'}
success, deliverable = process_request(f'{ES_URL}{INDEX_NAME}/_doc/1', 
    'put', data=json.dumps(data), headers=headers, permitted_status_codes=201)
if not success:    
    logger.error(f'... {deliverable.failure_reason}')
    ....