Python goodies (functional helpers, sane logging, etc)

crypto.py

#!/usr/bin/python
from subprocess import Popen, PIPE


def crypt(enc, data, secret_key):
    """Encrypt/decrypt with OpenSSL"""
    mode = "-e" if enc else "-d"
    data = data if enc else data+'\n'
    command = ["openssl", "enc", mode, "-base64", "-aes-256-cbc", "-salt", "-k", secret_key]
    p = Popen(command, stdin=PIPE, stdout=PIPE, stderr=PIPE, universal_newlines=True)
    output, err = p.communicate(data)
    return output.strip()


# convenience helpers

def openssl_decrypt(data, secret_key):
    return crypt(False, data, secret_key)
    
def openssl_encrypt(data, secret_key):
    return crypt(True, data, secret_key)

functional.py

def get_in(obj, path):
    """Return data or None from the path in obj/dict.
    
    get_in({'foo': {'bar': 123}}, ['foo', 'bar']) # => 123
    
    See Clojure's get-in.
    """
    key = path[0]
    rest_path = path[1:]
    if len(path) > 1:
        return get_in(get(obj, key), rest_path)
    return get(obj, key)


def get(obj, key):
    """Get data or None with key.
    
    See Clojure's get."""
    if isinstance(obj, dict):
        return obj.get(key)
    try:
        return getattr(obj, key)
    except AttributeError:
        return None


def assoc(obj, key, value):
    """Mutably set key, value to the object/dict.
    
    See Clojure's assoc.
    """
    if isinstance(obj, dict):
        obj[key] = value
        return obj
    if obj is None:
        return assoc({}, key, value)
    return setattr(obj, key, value)


def assoc_in(obj, path, value):
    """Mutably set value to the path in object/dict.
    
    assoc_in({}, ['foo', 'bar'], 123) # => {'foo': {'bar': 123}}

    See Clojure's assoc-in."""
    key = path[0]
    rest_path = path[1:]
    if len(path) > 1:
        assoc(obj, key, assoc_in(get(obj, key), rest_path, value))
        return obj
    return assoc(obj, key, value)


def select_keys(data: dict, keys: list) -> dict:
    """Create a new dict by selecting the provided keys from data.

    Data can be a dict or an object. If it is not a dict, the the getattr is
    used to select the keys from data.

    Example:
    select_keys({'a': 1, 'b': 2, 'c': 3}, ['a', 'b']) -> {'a': 1, 'b': 2}
    """
    if type(data) == dict:
        return {key: data.get(key) for key in keys if key in data}
    return {key: getattr(data, key) for key in keys if hasattr(data, key)}


def diff(old: set, new: set) -> set:
    """diff two sets and return (deletes, inserts, updates)."""
    return (old - new, new - old, old & new)

logging.py

# Setting up python logging is a pain in the but
# ...until you found this gist.
import logging
from logging.config import dictConfig as LOGGING_CONFIG

LOG_LEVEL = logging.INFO

LOGGING_CONFIG({
    'version': 1,
    'formatters': {
        'f': {'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'}
    },
    'handlers': {
        'h': {'class': 'logging.StreamHandler',
              'formatter': 'f',
              'level': LOG_LEVEL}
    },
    'root': {
        'handlers': ['h'],
        'level': LOG_LEVEL,
    },
})

# anywhere in your code
# import logger
logger = logging.getLogger()
logger.info('we are logging to stdout with a nice format like we should be')

m_pipe.py

#
# ErrorMonad and its fmap, bind and m_pipe
# inspired by
# http://www.slideshare.net/eldariof/monads-in-python
#

ErrorMonad = namedtuple('ErrorMonad', ['error', 'value'])


def fmap(func):
    """
    fmap wraps func in a function so that:
    - it receives ErrorMonad's value as a parameter
    - its output set as a new ErrorMonad's value
    - it is skipped entirely in case of error
    """

    def fmap_func(error_monad):
        if error_monad.error:
            return ErrorMonad(error=error_monad.error, value=None)

        try:
            return ErrorMonad(error=None, value=func(error_monad.value))
        except Exception as e:
            return ErrorMonad(error=e, value=None)

    return fmap_func


def bind(error_monad, func):
    """
    Pass the error_monad to an fmap wrapped func if the error_monad has a value
    otherwise skip execution of func
    """
    return fmap(func)(error_monad) if error_monad.value is not None else error_monad


def m_pipe(value, *fns) -> ErrorMonad:
    """
    Pass the first value to the first function and its output to the next until
    the last function.

    Returns the output of the last functions as an ErrorMonad (err, val) tuple.

    Similar to the thread macro in Lisps: (-> param1 param2 param3 ...)
    """
    error_monad = ErrorMonad(value=value, error=None)
    for f in fns:
        error_monad = bind(error_monad, f)
    return error_monad