5
5
from collections .abc import Collection
6
6
from functools import wraps
7
7
from json import JSONDecodeError
8
- from typing import Any , Callable , Optional , Union
8
+ from typing import Any , Callable , Optional , TypeVar , Union , overload
9
9
10
10
import backoff
11
11
import requests
@@ -34,7 +34,7 @@ def _read_mfa_code_from_stdin() -> str:
34
34
return inputted_mfa_code
35
35
36
36
37
- def _mask_senritive_value_for_dict (data : dict [str , Any ], keys : Collection [str ]) -> dict [str , Any ]:
37
+ def _mask_sensitive_value_for_dict (data : dict [str , Any ], keys : Collection [str ]) -> dict [str , Any ]:
38
38
"""
39
39
dictに含まれているセンシティブな情報を"***"でマスクします。
40
40
@@ -108,7 +108,7 @@ def mask_str_rquest_body(str_request_body: str) -> Any: # noqa: ANN401
108
108
109
109
if 400 <= response .status_code < 600 :
110
110
# logにAuthorizationを出力しないようにマスクする
111
- headers_for_logger = _mask_senritive_value_for_dict (dict (response .request .headers ), {"Authorization" })
111
+ headers_for_logger = _mask_sensitive_value_for_dict (dict (response .request .headers ), {"Authorization" })
112
112
113
113
# request_bodyのpassword関係をマスクして、logに出力する
114
114
request_body_for_logger : Optional [Any ] = None
@@ -140,28 +140,40 @@ def mask_str_rquest_body(str_request_body: str) -> Any: # noqa: ANN401
140
140
)
141
141
142
142
143
- def _create_request_body_for_logger (data : Any ) -> Any : # noqa: ANN401
143
+ T = TypeVar ("T" )
144
+
145
+
146
+ @overload
147
+ def _create_request_body_for_logger (data : bytes ) -> str :
148
+ pass
149
+
150
+
151
+ @overload
152
+ def _create_request_body_for_logger (data : T ) -> T : ...
153
+
154
+
155
+ def _create_request_body_for_logger (data : Any ) -> Any :
144
156
"""
145
- ログに出力するためのreqest_bodyを生成する 。
157
+ ログに出力するためのrequest_bodyを生成する 。
146
158
* パスワードやトークンなどの機密情報をマスクする
147
159
* bytes型の場合は `(bytes)`と記載する。
148
160
149
-
150
161
Args:
151
162
data: request_body
152
163
153
164
Returns:
154
165
ログ出力用のrequest_body
155
166
"""
156
- if not isinstance (data , dict ):
157
- return data
158
- elif isinstance (data , bytes ):
167
+ if isinstance (data , bytes ):
159
168
# bytes型のときは値を出力しても意味がないので、bytesであることが分かるようにする
160
169
return "(bytes)"
161
170
162
- return _mask_senritive_value_for_dict (
163
- data , keys = {"password" , "old_password" , "new_password" , "id_token" , "refresh_token" , "access_token" , "session" , "mfa_code" }
164
- )
171
+ # dictの場合は機密情報をマスクする
172
+ sensitive_keys = {"password" , "old_password" , "new_password" , "id_token" , "refresh_token" , "access_token" , "session" , "mfa_code" }
173
+ if isinstance (data , dict ):
174
+ return _mask_sensitive_value_for_dict (data , keys = sensitive_keys )
175
+
176
+ return data
165
177
166
178
167
179
def _create_query_params_for_logger (params : dict [str , Any ]) -> dict [str , Any ]:
@@ -175,7 +187,7 @@ def _create_query_params_for_logger(params: dict[str, Any]) -> dict[str, Any]:
175
187
Returns:
176
188
ログ出力用のparams
177
189
"""
178
- return _mask_senritive_value_for_dict (params , keys = {"X-Amz-Security-Token" , "X-Amz-Credential" })
190
+ return _mask_sensitive_value_for_dict (params , keys = {"X-Amz-Security-Token" , "X-Amz-Credential" })
179
191
180
192
181
193
def _should_retry_with_status (status_code : int ) -> bool :
0 commit comments