Source code for lago.sdk_utils

import functools
import inspect
import wrapt
import logging
from lago.log_utils import get_default_log_formatter

[docs]class SDKWrapper(wrapt.ObjectProxy): """A proxy object that exposes only methods which were decorated with :func:`expose` decorator.""" def __getattr__(self, name): attr = getattr(self.__wrapped__, name) return getattr_sdk(attr, name) def __dir__(self): orig = super(SDKWrapper, self).__dir__() filtered = [] for name in orig: attr = getattr(self.__wrapped__, name) try: getattr_sdk(attr, name) filtered.append(name) except AttributeError: pass return filtered
[docs]class SDKMethod(object): """Metadata to store inside the decorated function""" def __init__(self, name): = name
[docs]def expose(func): """ Decorator to be used with :class:`SDKWrapper`. This decorator indicates that the wrapped method or class should be exposed in the proxied object. Args: func(types.FunctionType/types.MethodType): function to decorate Returns: None """ @functools.wraps(func) def wrapped(*args, **kwargs): return func(*args, **kwargs) if inspect.isclass(func): wrapped._sdkmetaclass = SDKMethod(func.__name__) else: wrapped._sdkmeta = SDKMethod(func.__name__) return wrapped
[docs]def getattr_sdk(attr, name): """ Filter SDK attributes Args: attr(attribute): Attribute as returned by :func:`getattr`. name(str): Attribute name. Returns: `attr` if passed. """ if inspect.isroutine(attr): if hasattr(attr, '_sdkmeta'): return attr raise AttributeError(name)
[docs]def setup_sdk_logging(logfile=None, loglevel=logging.INFO): """ Setup a NullHandler to the root logger. If ``logfile`` is passed, additionally add a FileHandler in ``loglevel`` level. Args: logfile(str): A path to setup a log file. loglevel(int): :mod:`logging` log level. Returns: None """ logging.root.setLevel(logging.DEBUG) logging.root.addHandler(logging.NullHandler()) if logfile: fh = logging.FileHandler(logfile) fh.setLevel(loglevel) fh.setFormatter(get_default_log_formatter()) logging.root.addHandler(fh)