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.
requestsobsługuje tak zwane podpięcia zdarzeń (od 2.23 jest właściwie tylko responsepodpię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 reqi resdodatkowe 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 stderrbę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 loggingmogą 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:
