forked from microsoft/azure-devops-python-samples
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathhttp_logging.py
More file actions
119 lines (93 loc) · 2.48 KB
/
http_logging.py
File metadata and controls
119 lines (93 loc) · 2.48 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
"""
HTTP logger hook (for dumping the request/response cycle).
"""
from contextlib import contextmanager
import json
###
# Logging state management
###
_enabled_stack = [False]
target = None
def push_state(enabled):
_enabled_stack.append(enabled)
def pop_state():
if len(_enabled_stack) == 1:
# never pop the last state
return bool(_enabled_stack[0])
elif len(_enabled_stack) == 0:
# something's gone terribly wrong
raise RuntimeError("_enabled_stack should never be empty")
else:
return bool(_enabled_stack.pop())
def logging_enabled():
return bool(_enabled_stack[-1])
@contextmanager
def temporarily_disabled():
"""Temporarily disable logging if it's enabled.
with http_logging.temporarily_disabled():
my_client.do_thing()
"""
push_state(False)
yield
pop_state()
###
# Actual HTTP logging and hook
###
def _trim_headers(headers):
trimmable_headers = [
"X-VSS-PerfData",
"X-TFS-Session",
"X-VSS-E2EID",
"X-VSS-Agent",
"Authorization",
"X-TFS-ProcessId",
"X-VSS-UserData",
"ActivityId",
"P3P",
"X-Powered-By",
"Cookie",
"X-TFS-FedAuthRedirect",
"Strict-Transport-Security",
"X-Frame-Options",
"X-Content-Type-Options",
"X-AspNet-Version",
"Server",
"Pragma",
"vary",
"X-MSEdge-Ref",
"Cache-Control",
"Date",
"User-Agent",
"Accept-Language",
]
cleaned_headers = headers.copy()
for trim_header in trimmable_headers:
try:
del cleaned_headers[trim_header]
except KeyError:
pass
return dict(cleaned_headers)
def log_request(response, file):
try:
content = response.json()
except ValueError:
content = response.text
data = {
'request': {
'url': response.request.url,
'headers': _trim_headers(response.request.headers),
'body': str(response.request.body),
'method': response.request.method,
},
'response': {
'headers': _trim_headers(response.headers),
'body': content,
'status': response.status_code,
'url': response.url,
},
}
json.dump(data, file, indent=4)
def requests_hook(response, *args, **kwargs):
global target
if logging_enabled() and target is not None:
log_request(response, target)