Python 요청의 고급 사용 - 시간 초과, 재시도, 연결
31525 단어 pythonwebdevtutorialtodayilearned
간단한 API로 인해 요청이 즉시 적용되기 쉽지만 이 라이브러리는 고급 용례에 확장성을 제공합니다.API 집약형 클라이언트나 웹 스크레이퍼를 작성하고 있다면, 네트워크 고장, 유용한 디버깅 추적, 문법 분석을 용인해야 할 수도 있습니다.
다음은 JSON API를 광범위하게 사용하는 웹 캡처 도구나 프로그램을 작성할 때 요청에 유용한 기능을 발견한 요약입니다.
카탈로그
연결 요청
일반적으로 타사 API를 사용할 때 반환된 응답이 유효한지 확인해야 합니다.Requests는 응답 HTTP 상태 코드가 4xx 또는 5xx가 아니라고 단언하는 속기 조수
raise_for_status()를 제공합니다. 요청이 클라이언트나 서버 오류를 일으키지 않았습니다.예:
response = requests.get('https://api.github.com/user/repos?page=1')
# Assert that there were no errors
response.raise_for_status()
raise_for_status(), 이것은 중복될 수 있다.다행히도, 요청 라이브러리는 "hooks"인터페이스를 제공합니다. 이 인터페이스에 리셋을 추가할 수 있습니다.요청 과정의 일부분입니다.
우리는 응답 대상마다 호출할 수 있도록 갈고리를 사용할 수 있다
raise_for_status().# Create a custom requests object, modifying the global module throws an error
http = requests.Session()
assert_status_hook = lambda response, *args, **kwargs: response.raise_for_status()
http.hooks["response"] = [assert_status_hook]
http.get("https://api.github.com/user/repos?page=1")
> HTTPError: 401 Client Error: Unauthorized for url: https://api.github.com/user/repos?page=1
기본 URL 설정
API 호스팅된 API 하나만 사용한다고 가정합니다.조직http 호출마다 프로토콜과 도메인이 반복됩니다.
requests.get('https://api.org/list/')
requests.get('https://api.org/list/3/item')
from requests_toolbelt import sessions
http = sessions.BaseUrlSession(base_url="https://api.org")
http.get("/list")
http.get("/list/item")
기본 시간 초과 설정
모든 생산 코드에 시간 초과 문서를 설정해 주십시오 recommends.시간 초과를 설정하는 것을 잊어버리면 비헤이비어가 잘못된 서버가 프로그램이 중단될 수 있습니다. 특히 대부분의 Python 코드가 동기화되어 있기 때문입니다.
requests.get('https://github.com/', timeout=0.001)
Transport Adapters를 사용하면 모든 HTTP 호출에 대해 기본 시간 초과를 설정할 수 있습니다.이것은 개발자가 타임아웃=1 파라미터를 단일 호출에 추가하는 것을 잊어버리더라도 합리적인 시간 초과를 설정하지만, 매번 호출을 바탕으로 덮어쓸 수 있도록 합니다.
다음은 기본 시간 초과가 있는 사용자 정의 전송 어댑터의 예입니다. 그 영감은 this Github comment 입니다.http 클라이언트와send () 방법을 구성할 때, 우리는 시간 초과 파라미터가 제공되지 않은 상황에서 기본 시간 초과를 사용할 수 있도록 구조 함수를 다시 작성합니다.
from requests.adapters import HTTPAdapter
DEFAULT_TIMEOUT = 5 # seconds
class TimeoutHTTPAdapter(HTTPAdapter):
def __init__(self, *args, **kwargs):
self.timeout = DEFAULT_TIMEOUT
if "timeout" in kwargs:
self.timeout = kwargs["timeout"]
del kwargs["timeout"]
super().__init__(*args, **kwargs)
def send(self, request, **kwargs):
timeout = kwargs.get("timeout")
if timeout is None:
kwargs["timeout"] = self.timeout
return super().send(request, **kwargs)
import requests
http = requests.Session()
# Mount it for both http and https usage
adapter = TimeoutHTTPAdapter(timeout=2.5)
http.mount("https://", adapter)
http.mount("http://", adapter)
# Use the default 2.5s timeout
response = http.get("https://api.twilio.com/")
# Override the timeout as usual for specific requests
response = http.get("https://api.twilio.com/", timeout=10)
실패 시 재시도
네트워크 연결이 손상되고 혼잡하며 서버에 장애가 발생했습니다.만약 우리가 진정으로 건장한 프로그램을 구축하고 싶다면, 우리는 실패를 고려하고 재시도 전략을 세워야 한다.
HTTP 클라이언트에 재시도 정책을 추가하는 것은 간단합니다.HTTPAdapter를 만들고 정책을 어댑터에 전달합니다.
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
retry_strategy = Retry(
total=3,
status_forcelist=[429, 500, 502, 503, 504],
method_whitelist=["HEAD", "GET", "OPTIONS"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
http = requests.Session()
http.mount("https://", adapter)
http.mount("http://", adapter)
response = http.get("https://en.wikipedia.org/w/api.php")
다음 매개 변수는 요청 라이브러리에서 사용하는 기본 매개 변수를 포함합니다.
total=10
urllib3.exceptions.MaxRetryError 예외.나는 내가 사용하는 API에 따라 이 매개 변수를 바꾸지만, 보통 10보다 낮게 설정하고, 보통 세 번 다시 시도하면 충분하다.status_forcelist=[413, 429, 503]
method_whitelist=["HEAD", "GET", "PUT", "DELETE", "OPTIONS", "TRACE"]
backoff_factor=0
{backoff factor} * (2 ** ({number of total retries} - 1))
0.5, 1, 2, 4, 8, 16, 32, 64, 128, 2561초.1, 2, 4, 8, 16, 32, 64, 128, 256, 5125, 10, 20, 40, 80, 160, 320, 640, 1280, 2560기본적으로 이 값은 0입니다. 이것은 지수 회피를 설정하지 않고 즉시 재시도를 실행하는 것을 의미합니다.해머를 피하기 위해 1인치로 설정해야 합니다
서버!
재시도 모듈의 전체 문서는 here 입니다.
시간 초과와 재시도 결합
HTTPAdapter는 비교가 가능하므로 다음과 같이 재시도와 시간 초과를 결합할 수 있습니다.
retries = Retry(total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504])
http.mount("https://", TimeoutHTTPAdapter(max_retries=retries))
HTTP 요청 디버그
때로는 요청이 실패할 수도 있습니다. 원인을 찾을 수 없습니다.요청과 응답을 기록하면 고장을 알 수 있습니다.이 두 가지 방법이 있습니다. 내장된 디버그 로그 설정을 사용하거나 요청 연결을 사용하십시오.
HTTP 헤더 인쇄
0 이상의 로그 디버그 수준을 변경하면 응답 HTTP 헤더가 기록됩니다.이것은 가장 간단한 옵션이지만 HTTP 요청이나 응답체를 볼 수 없습니다.로깅에 적합하지 않거나 바이너리 컨텐츠를 포함하는 대규모 바디 로드를 반환하는 API를 처리하는 경우 유용합니다.
0보다 큰 값은 디버그 로깅을 사용합니다.
import requests
import http
http.client.HTTPConnection.debuglevel = 1
requests.get("https://www.google.com/")
# Output
send: b'GET / HTTP/1.1\r\nHost: www.google.com\r\nUser-Agent: python-requests/2.22.0\r\nAccept-Encoding: gzip, deflate\r\nAccept: */*\r\nConnection: keep-alive\r\n\r\n'
reply: 'HTTP/1.1 200 OK\r\n'
header: Date: Fri, 28 Feb 2020 12:13:26 GMT
header: Expires: -1
header: Cache-Control: private, max-age=0
모든 항목 인쇄
요청과 응답을 포함한 전체 HTTP 라이프 사이클을 기록하려면 requests\u 도구 모음의 요청 연결 및 UTIL 덤프를 사용할 수 있습니다.
REST 기반 API를 처리할 때 이 옵션이 매우 큰 응답으로 돌아오지 않기 때문에 더욱 좋습니다.
import requests
from requests_toolbelt.utils import dump
def logging_hook(response, *args, **kwargs):
data = dump.dump_all(response)
print(data.decode('utf-8'))
http = requests.Session()
http.hooks["response"] = [logging_hook]
http.get("https://api.openaq.org/v1/cities", params={"country": "BA"})
# Output
< GET /v1/cities?country=BA HTTP/1.1
< Host: api.openaq.org
> HTTP/1.1 200 OK
> Content-Type: application/json; charset=utf-8
> Transfer-Encoding: chunked
> Connection: keep-alive
>
{
"meta":{
"name":"openaq-api",
"license":"CC BY 4.0",
"website":"https://docs.openaq.org/",
"page":1,
"limit":100,
"found":1
},
"results":[
{
"country":"BA",
"name":"Goražde",
"city":"Goražde",
"count":70797,
"locations":1
}
]
}
테스트 및 시뮬레이션 요청
제3자 API를 사용하여 개발에 난점을 도입했다. 단원 테스트를 하기 어렵다.Sentry의 엔지니어는 라이브러리를 작성하여 개발 과정에서의 요구를 모의하여 약간의 고통을 경감시켰다.
서버에 HTTP 응답을 보내는 것이 아니라 HTTP 요청을 캡처하여 테스트 중에 추가한 미리 정의된 응답을 반환합니다.
가장 좋은 것은 하나의 예로 설명하는 것이다.
import unittest
import requests
import responses
class TestAPI(unittest.TestCase):
@responses.activate # intercept HTTP calls within this method
def test_simple(self):
response_data = {
"id": "ch_1GH8so2eZvKYlo2CSMeAfRqt",
"object": "charge",
"customer": {"id": "cu_1GGwoc2eZvKYlo2CL2m31GRn", "object": "customer"},
}
# mock the Stripe API
responses.add(
responses.GET,
"https://api.stripe.com/v1/charges",
json=response_data,
)
response = requests.get("https://api.stripe.com/v1/charges")
self.assertEqual(response.json(), response_data)
class TestAPI(unittest.TestCase):
@responses.activate
def test_simple(self):
responses.add(responses.GET, "https://api.stripe.com/v1/charges")
response = requests.get("https://invalid-request.com")
requests.exceptions.ConnectionError: Connection refused by Responses - the call doesn't match any registered mock.
Request:
- GET https://invalid-request.com/
Available matches:
- GET https://api.stripe.com/v1/charges
getsentry/응답 브라우저 동작 모방
웹 스크레이퍼 코드를 충분히 작성했다면, 브라우저를 사용하느냐, 프로그래밍 방식으로 이 사이트를 방문하느냐에 따라 일부 사이트가 어떻게 다른 HTML로 되돌아오는지 알 수 있습니다.때때로 이것은 반캡처 조치이지만, 일반적으로 서버는 사용자 에이전트 탐지에 참여하여 장치에 가장 적합한 내용(예를 들어 데스크톱이나 모바일 장치)을 찾아낸다.
브라우저와 같은 내용을 표시하려면 Firefox나 Chrome에서 보낸 내용을 사용하여 사용자 프록시 제목 요청 집합을 덮어쓸 수 있습니다.
import requests
http = requests.Session()
http.headers.update({
"User-Agent": "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:68.0) Gecko/20100101 Firefox/68.0"
})
Reference
이 문제에 관하여(Python 요청의 고급 사용 - 시간 초과, 재시도, 연결), 우리는 이곳에서 더 많은 자료를 발견하고 링크를 클릭하여 보았다 https://dev.to/danihodovic/advanced-usage-of-python-requests-timeouts-retries-hooks-kok텍스트를 자유롭게 공유하거나 복사할 수 있습니다.하지만 이 문서의 URL은 참조 URL로 남겨 두십시오.
우수한 개발자 콘텐츠 발견에 전념
(Collection and Share based on the CC Protocol.)
좋은 웹페이지 즐겨찾기
