Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""Test logger."""
from routemaster.logging import BaseLogger
class TestLogger(BaseLogger):
"""This logger is loaded during tests."""
def __init__(self, config, *args, **kwargs):
super().__init__(config, *args, **kwargs)
self.kwargs = kwargs
class InvalidLogger(object):
"""
This logger cannot be loaded.
It does not inherit from `routemaster.logging.BaseLogger`.
"""
def dynamic_logger(config):
"""This is a callable rather than a class."""
def dynamic_logger(config):
"""This is a callable rather than a class."""
return BaseLogger(config)
pass
def critical(self, *args, **kwargs):
"""Mirror Python logging API for `critical`."""
pass
def log(self, *args, **kwargs):
"""Mirror Python logging API for `log`."""
pass
def exception(self, *args, **kwargs):
"""Mirror Python logging API for `exception`."""
pass
class PythonLogger(BaseLogger):
"""Routemaster logging interface for Python's logging library."""
def __init__(self, *args, log_level: str) -> None:
super().__init__(*args)
logging.basicConfig(
format=(
"[%(asctime)s] [%(process)d] [%(levelname)s] "
"[%(name)s] %(message)s"
),
datefmt="%Y-%m-%d %H:%M:%S %z",
level=getattr(logging, log_level),
)
self.logger = logging.getLogger('routemaster')
@contextlib.contextmanager
self.logger.error(*args, **kwargs)
def critical(self, *args, **kwargs):
"""Mirror Python logging API for `critical`."""
self.logger.critical(*args, **kwargs)
def log(self, *args, **kwargs):
"""Mirror Python logging API for `log`."""
self.logger.log(*args, **kwargs)
def exception(self, *args, **kwargs):
"""Mirror Python logging API for `exception`."""
self.logger.exception(*args, **kwargs)
class LoggerProxy(BaseLogger):
"""Proxies logging calls to all loggers in a list."""
def __init__(self, loggers: List[BaseLogger]) -> None:
self.loggers = loggers
def __getattr__(self, name):
"""Return a proxy function that will dispatch to all loggers."""
def log_all(*args, **kwargs):
for logger in self.loggers:
getattr(logger, name)(*args, **kwargs)
@contextlib.contextmanager
def log_all_ctx(*args, **kwargs):
with contextlib.ExitStack() as stack:
for logger in self.loggers:
logger_ctx = getattr(logger, name)
)
if not callable(klass):
raise PluginConfigurationException(
f"{dotted_path} must be callable",
)
try:
logger = klass(config, **logger_config.kwargs)
except TypeError:
raise PluginConfigurationException(
f"Could not instantiate logger, {klass_name} must take a config "
f"argument and any kwargs specified in the plugin configuration.",
)
if not isinstance(logger, BaseLogger):
raise PluginConfigurationException(
f"{dotted_path} must inherit from routemaster.logging.BaseLogger",
)
return logger
def __getattr__(self, name):
"""Return a proxy function that will dispatch to all loggers."""
def log_all(*args, **kwargs):
for logger in self.loggers:
getattr(logger, name)(*args, **kwargs)
@contextlib.contextmanager
def log_all_ctx(*args, **kwargs):
with contextlib.ExitStack() as stack:
for logger in self.loggers:
logger_ctx = getattr(logger, name)
stack.enter_context(logger_ctx(*args, **kwargs))
yield
if isinstance(
getattr(BaseLogger, name),
contextlib.AbstractContextManager,
):
return log_all_ctx
return log_all
It adds per-request exception reporting to Flask for the API, and also wraps
the cron, webhook request, and feed request processes with Sentry reporting.
All wrapping re-raises exceptions, as Routemaster/Flask/the cron subsystem will
all handle exceptions appropriately.
"""
import contextlib
import pkg_resources
from raven import Client
from raven.contrib.flask import Sentry
from routemaster.logging import BaseLogger
class SentryLogger(BaseLogger):
"""Instruments Routemaster with Sentry."""
def __init__(self, *args, dsn):
try:
version = pkg_resources.working_set.by_key['routemaster'].version
except KeyError:
version = 'dev'
self.client = Client(
dsn,
release=version,
sample_rate=0 if 'dev' in version else 1,
include_paths=[
'routemaster',
],
)
"""
Prometheus metrics exporter for Routemaster.
This package provides a Routemaster logging plugin that interfaces to the
Python Prometheus API, to export monitoring metrics to Prometheus.
"""
import contextlib
from prometheus_client import Counter, CollectorRegistry
from prometheus_flask_exporter import PrometheusMetrics
from routemaster.logging import BaseLogger
class PrometheusLogger(BaseLogger):
"""Instruments Routemaster with Prometheus."""
def __init__(self, *args, path='/metrics'):
self.path = path
self.registry = CollectorRegistry(auto_describe=True)
self.exceptions = Counter(
'exceptions',
"Exceptions logged",
('type',),
registry=self.registry,
)
self.cron_jobs_processed = Counter(
'cron_jobs_processed',
"Cron jobs processed",