def __redact_key_from_response(self, record):
# type: (logging.LogRecord) -> None
"""Redact the ``Plaintext`` value from a KMS response body.
:param record: Logging record to filter
:type record: logging.LogRecord
"""
try:
parsed_body = json.loads(self.__to_str(cast(tuple, record.args)[0]))
parsed_body['Plaintext'] = _REDACTED
new_args = (json.dumps(parsed_body, sort_keys=True),) + cast(tuple, record.args)[1:]
record.args = new_args
except Exception: # pylint: disable=broad-except
return
python类LogRecord()的实例源码
def __redact_record(self, record):
# type: (logging.LogRecord) -> logging.LogRecord
"""Redact any values from a record, as necessary.
:param record: Logging record to filter
:type record: logging.LogRecord
"""
_record = copy.deepcopy(record)
if self.__is_kms_encrypt_request(_record):
self.__redact_encrypt_request(_record)
elif self.__is_kms_response_with_plaintext(_record):
self.__redact_key_from_response(_record)
return _record
def filter(self, record):
# type: (logging.LogRecord) -> bool
"""Determines whether to filter record.
:param record: Logging record to filter
:type record: logging.LogRecord
:rtype: bool
"""
return record.name not in self.__blacklist
def create_record(name='test_logger', level='DEBUG', msg='foobar'):
return LogRecord(name=name, level=LEVEL_MAP[level], msg=msg, lineno=None,
args=None, exc_info=None, pathname=None)
# LoggerFilter
def assertLogRecords(self, records, matches):
self.assertEqual(len(records), len(matches))
for rec, match in zip(records, matches):
self.assertIsInstance(rec, logging.LogRecord)
for k, v in match.items():
self.assertEqual(getattr(rec, k), v)
def __init__(self, records: typing.List[logging.LogRecord]):
self._records = records
def filter_by_level(self, minimum_level=logging.NOTSET):
minimum_level = self._sanitize_level(minimum_level)
def filter_func(rec: logging.LogRecord):
return rec.levelno >= minimum_level
return self.__filter(filter_func)
def _filter_by_str(self, text: str or None, rec_attrib):
if text is None:
return self
def filter_func(rec: logging.LogRecord):
return text.lower() in getattr(rec, rec_attrib)
return self.__filter(filter_func)
def __iter__(self) -> typing.Generator[logging.LogRecord, None, None]:
for x in self._records:
yield x
def handle_record(self, record: logging.LogRecord):
raise NotImplementedError()
def emit(self, record: logging.LogRecord):
self._records.append(record)
for follower in self._followers:
follower.handle_record(record)
def records(self) -> typing.Iterator[logging.LogRecord]:
return self.filter_records(**self.filters)
def handle_record(self, record: logging.LogRecord):
if self.filter_record(record, **self.filters):
I.tab_log_write(self.format(record), str(self.colors[record.levelname]))
def _send(self):
content = []
for rec in self.filter_records():
assert isinstance(rec, logging.LogRecord)
content.append(self.format(rec))
url = pastebin.create_new_paste('\n'.join(content))
if url:
SENTRY.captureMessage('Logfile', extra={'log_url': url})
self.tab_log_write('Log file sent; thank you !')
else:
self.tab_log_write('Could not send log file')
def shouldFlush(self, record):
"""Should the buffer be automatically flushed?
:param logging.LogRecord record: log record to be considered
:returns: False because the buffer should never be auto-flushed
:rtype: bool
"""
return False
def emit(self, record):
"""Log the specified logging record.
:param logging.LogRecord record: Record to be formatted
"""
self._delete = False
# logging handlers use old style classes in Python 2.6 so
# super() cannot be used
if sys.version_info < (2, 7): # pragma: no cover
logging.StreamHandler.emit(self, record)
else:
super(TempHandler, self).emit(record)
def filter(self, record: logging.LogRecord):
if record.exc_info == (None, None, None):
return False
else:
return True
def __init__(self, level=logging.NOTSET):
# type: (int) -> None
super(MemoryLogHandler, self).__init__(level)
self._records = [] # type: List[logging.LogRecord]
self.max_level_emitted = logging.NOTSET
