Source code
Revision control
Copy as Markdown
Other Tools
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
import json
def read(log_f, raise_on_error=False):
"""Return a generator that will return the entries in a structured log file.
Note that the caller must not close the file whilst the generator is still
in use.
:param log_f: file-like object containing the raw log entries, one per line
:param raise_on_error: boolean indicating whether ValueError should be raised
for lines that cannot be decoded."""
while True:
line = log_f.readline()
if not line:
# This allows log_f to be a stream like stdout
break
try:
yield json.loads(line)
except ValueError:
if raise_on_error:
raise
def imap_log(log_iter, action_map):
"""Create an iterator that will invoke a callback per action for each item in a
iterable containing structured log entries
:param log_iter: Iterator returning structured log entries
:param action_map: Dictionary mapping action name to callback function. Log items
with actions not in this dictionary will be skipped.
"""
for item in log_iter:
if item["action"] in action_map:
yield action_map[item["action"]](item)
def each_log(log_iter, action_map):
"""Call a callback for each item in an iterable containing structured
log entries
:param log_iter: Iterator returning structured log entries
:param action_map: Dictionary mapping action name to callback function. Log items
with actions not in this dictionary will be skipped.
"""
for item in log_iter:
if item["action"] in action_map:
action_map[item["action"]](item)
class LogHandler(object):
"""Base class for objects that act as log handlers. A handler is a callable
that takes a log entry as the only argument.
Subclasses are expected to provide a method for each action type they
wish to handle, each taking a single argument for the test data.
For example a trivial subclass that just produces the id of each test as
it starts might be::
class StartIdHandler(LogHandler):
def test_start(data):
#For simplicity in the example pretend the id is always a string
return data["test"]
"""
def __call__(self, data):
if hasattr(self, data["action"]):
handler = getattr(self, data["action"])
return handler(data)
def handle_log(log_iter, handler):
"""Call a handler for each item in a log, discarding the return value"""
for item in log_iter:
handler(item)