Mając skrypt lub nawet podsystem aplikacji do debugowania protokołu sieciowego, pożądane jest, aby zobaczyć, jakie dokładnie są pary żądanie-odpowiedź, w tym skuteczne adresy URL, nagłówki, ładunki i status. I zazwyczaj niepraktyczne jest nagrywanie indywidualnych żądań w każdym miejscu. Jednocześnie istnieją rozważania dotyczące wydajności, które sugerują użycie pojedynczego (lub kilku specjalistycznych) requests.Session
, więc poniżej założono, że sugestia jest przestrzegana.
requests
obsługuje tak zwane podpięcia zdarzeń (od 2.23 jest właściwie tylko response
podpięcie). Jest to w zasadzie detektor zdarzeń, a zdarzenie jest emitowane przed zwróceniem kontroli z requests.request
. W tym momencie zarówno żądanie, jak i odpowiedź są w pełni zdefiniowane, dzięki czemu można je zarejestrować.
import logging
import requests
logger = logging.getLogger('httplogger')
def logRoundtrip(response, *args, **kwargs):
extra = {'req': response.request, 'res': response}
logger.debug('HTTP roundtrip', extra=extra)
session = requests.Session()
session.hooks['response'].append(logRoundtrip)
To w zasadzie sposób rejestrowania wszystkich obiegów HTTP sesji.
Formatowanie rekordów protokołu HTTP w obie strony
Aby powyższe rejestrowanie było przydatne, może istnieć wyspecjalizowany program formatujący rejestrowanie, który rozumie req
i res
dodatkowe informacje dotyczące rejestrowania rekordów. Może to wyglądać tak:
import textwrap
class HttpFormatter(logging.Formatter):
def _formatHeaders(self, d):
return '\n'.join(f'{k}: {v}' for k, v in d.items())
def formatMessage(self, record):
result = super().formatMessage(record)
if record.name == 'httplogger':
result += textwrap.dedent('''
---------------- request ----------------
{req.method} {req.url}
{reqhdrs}
{req.body}
---------------- response ----------------
{res.status_code} {res.reason} {res.url}
{reshdrs}
{res.text}
''').format(
req=record.req,
res=record.res,
reqhdrs=self._formatHeaders(record.req.headers),
reshdrs=self._formatHeaders(record.res.headers),
)
return result
formatter = HttpFormatter('{asctime} {levelname} {name} {message}', style='{')
handler = logging.StreamHandler()
handler.setFormatter(formatter)
logging.basicConfig(level=logging.DEBUG, handlers=[handler])
Teraz, jeśli wykonasz kilka żądań za pomocą session
, takich jak:
session.get('https://httpbin.org/user-agent')
session.get('https://httpbin.org/status/200')
Dane wyjściowe do stderr
będą wyglądać następująco.
2020-05-14 22:10:13,224 DEBUG urllib3.connectionpool Starting new HTTPS connection (1): httpbin.org:443
2020-05-14 22:10:13,695 DEBUG urllib3.connectionpool https://httpbin.org:443 "GET /user-agent HTTP/1.1" 200 45
2020-05-14 22:10:13,698 DEBUG httplogger HTTP roundtrip
---------------- request ----------------
GET https://httpbin.org/user-agent
User-Agent: python-requests/2.23.0
Accept-Encoding: gzip, deflate
Accept: */*
Connection: keep-alive
None
---------------- response ----------------
200 OK https://httpbin.org/user-agent
Date: Thu, 14 May 2020 20:10:13 GMT
Content-Type: application/json
Content-Length: 45
Connection: keep-alive
Server: gunicorn/19.9.0
Access-Control-Allow-Origin: *
Access-Control-Allow-Credentials: true
{
"user-agent": "python-requests/2.23.0"
}
2020-05-14 22:10:13,814 DEBUG urllib3.connectionpool https://httpbin.org:443 "GET /status/200 HTTP/1.1" 200 0
2020-05-14 22:10:13,818 DEBUG httplogger HTTP roundtrip
---------------- request ----------------
GET https://httpbin.org/status/200
User-Agent: python-requests/2.23.0
Accept-Encoding: gzip, deflate
Accept: */*
Connection: keep-alive
None
---------------- response ----------------
200 OK https://httpbin.org/status/200
Date: Thu, 14 May 2020 20:10:13 GMT
Content-Type: text/html; charset=utf-8
Content-Length: 0
Connection: keep-alive
Server: gunicorn/19.9.0
Access-Control-Allow-Origin: *
Access-Control-Allow-Credentials: true
Sposób GUI
Gdy masz dużo zapytań, przydatny jest prosty interfejs użytkownika i sposób filtrowania rekordów. Pokażę, że do tego celu użyję Chronologera (którego jestem autorem).
Po pierwsze, hak został przepisany w celu utworzenia rekordów, które logging
mogą być serializowane podczas przesyłania przez przewód. Może to wyglądać tak:
def logRoundtrip(response, *args, **kwargs):
extra = {
'req': {
'method': response.request.method,
'url': response.request.url,
'headers': response.request.headers,
'body': response.request.body,
},
'res': {
'code': response.status_code,
'reason': response.reason,
'url': response.url,
'headers': response.headers,
'body': response.text
},
}
logger.debug('HTTP roundtrip', extra=extra)
session = requests.Session()
session.hooks['response'].append(logRoundtrip)
Po drugie, konfiguracja logowania musi być dostosowana do użycia logging.handlers.HTTPHandler
(co Chronologer rozumie).
import logging.handlers
chrono = logging.handlers.HTTPHandler(
'localhost:8080', '/api/v1/record', 'POST', credentials=('logger', ''))
handlers = [logging.StreamHandler(), chrono]
logging.basicConfig(level=logging.DEBUG, handlers=handlers)
Na koniec uruchom instancję Chronologer. np. używając Dockera:
docker run --rm -it -p 8080:8080 -v /tmp/db \
-e CHRONOLOGER_STORAGE_DSN=sqlite:////tmp/db/chrono.sqlite \
-e CHRONOLOGER_SECRET=example \
-e CHRONOLOGER_ROLES="basic-reader query-reader writer" \
saaj/chronologer \
python -m chronologer -e production serve -u www-data -g www-data -m
I ponownie uruchom żądania:
session.get('https://httpbin.org/user-agent')
session.get('https://httpbin.org/status/200')
Procedura obsługi strumienia wygeneruje:
DEBUG:urllib3.connectionpool:Starting new HTTPS connection (1): httpbin.org:443
DEBUG:urllib3.connectionpool:https://httpbin.org:443 "GET /user-agent HTTP/1.1" 200 45
DEBUG:httplogger:HTTP roundtrip
DEBUG:urllib3.connectionpool:https://httpbin.org:443 "GET /status/200 HTTP/1.1" 200 0
DEBUG:httplogger:HTTP roundtrip
Teraz, jeśli otworzysz http: // localhost: 8080 / (użyj „logger” jako nazwy użytkownika i pustego hasła dla podstawowego okna autoryzacji) i klikniesz przycisk „Otwórz”, powinieneś zobaczyć coś takiego: