Redis PubSub wrapper for Python

Recently I've found that there's no reasonable simple and useful Redis pub sub examples around. So, here is my dead simple wrapper how to implement it without any unnecessary overhead.

Sphinx Doc, JSON highlight and Sphinx extensions: kung-fu

At the moment Pygments which used by Sphinx Doc haven't support for JSON code highlight which is really sad.

I've not found any useful information how to do it quickly. So here is my way:

1. I've found custom pygments lexer which support JSON: pygments-json . I will be part of pygments soon
2. It wasn't clear to me how to add custom pygments lexer to sphinx, my google-fu isn't good today
3. A bit more googling gave me Sphinx Extensions API , especially add_lexer method of Sphinx instance

Ok, now it's clear to me how to add new lexer. I've created ext/hijson folder within source, to __init__.py I've added setup function:

Also here is how to add support of ext folder as folder with custom extensions So, pip install pygments-json and use

.. code-block:: json

Nice and smooth:

PyFlakes-clean import of local_settings.py

Any project have own bunch of settings for different environments: for dev, production, staging. Everyone use it on daily basis. But annoying thing is that PyFlakes, great code analysis tool, warn about that. And it's reasonable.

So, to have this functionality, but without warning I use this pattern:

try:
    import local_settings
    for setting in dir(local_settings):
        if setting.startswith('_'):
            continue
        locals()[setting] = getattr(local_settings, setting)
except ImportError:
    pass

Simple and useful!


Update: I've added ignore of attributes which starting with underscore. Thanks to Igor Davydenko

Unstable HTTP services: what we can do to easily handle that?

The story: I have several HTTP service providers which works quite unstable. Yes, I had to have in mind this during development. But we're all thought that issues are "temporary" and will gone when we going to production. We accurately added logging.error in every place and move on with other stuff.

But our expectation about temporary nature of service behavior will never happen. Service sometimes work slowly, sometimes return HTTP errors and so on. We receive tons of exceptions every day. We had to do something with that.

The solution: Here is safe_exec decorator which help solve this problem. You can specify how many times you want to try execute function, what's timeout between them and what exceptions are expected during execution decorated function. For example, urllib2.urlopen may generate urllib2.URLError or  urllib2.HTTPError.

import logging
import time

from functools import wraps

__all__ = ("safe_exec",)


def safe_exec(exceptions, shakes=3, timeout=1, title="", **kwargs):
    """
    Decorator to safely execute function or method
    within `shakes` trying.

    In case provide argument `default` exception will not
    be raised and will return provided value.
    """
    def wrap(func):
        if not isinstance(exceptions, tuple):
            raise TypeError(
                "First argument of safe_exec should be tuple of exceptions"
            )

        @wraps(func)
        def wrapped(*args, **kwargs):
            result = None
            orig_exception = None
            for shake in range(shakes):
                try:
                    result = func(*args, **kwargs)
                    break
                except exceptions, orig_exception:
                    logging.warn("%s: Sorry, can't execute %s, shake #%d",
                        title,
                        func.__name__,
                        shake,
                        exc_info=True
                    )
                    time.sleep(timeout)
            else:
                logging.error(
                    "%s: Can't execute `%s` after %d shakes",
                    title,
                    func.__name__,
                    shakes
                )

                if "default" in kwargs:
                    return kwargs.get("default")

                raise orig_exception

            return result
        return wrapped
    return wrap

Sample usage:

import urllib2

@safe_exec((urllib2.URLError, urllib2.HTTPError), shakes=2)
def download(url):
    return urllib2.urlopen(url).read()

download("http://slow-resource.com/")

Flask-Jasmine: Execute Jasmine tests within Flask

Just finished Flask-Jasmine extension to execute beautiful Behavior Driven tests for Jasmine in JavaScript.

Such extensions already exists for Django and for Rails. Now it's available for Flask too.

Install with pip:

pip install Flask-Jasmine

Detailed instruction about configuration and usage

Activation/Deactivation of python virtualenv upon entering a directory

It's not a new or original idea – I've heard about it from Dmitry Gladkov but as usual didn't remember details. So, I've created my own implementation of activation/deactivation of python virtualenv:

#!/bin/bash

PREVPWD=`pwd`
PREVENV_PATH=
PREV_PS1=
PREV_PATH=

handle_virtualenv(){
  if [ "$PWD" != "$PREVPWD" ]; then
    PREVPWD="$PWD";
    if [ -n "$PREVENV_PATH" ]; then
      if [ "`echo "$PWD" | grep -c $PREVENV_PATH`" = "0"  ]; then
         source $PREVENV_PATH/.venv
         echo "> Virtualenv `basename $VIRTUALENV_PATH` deactivated"
         PS1=$PREV_PS1
         PATH=$PREV_PATH
         PREVENV_PATH=
      fi
    fi
    # activate virtualenv dynamically
    if [ -e "$PWD/.venv" ] && [ "$PWD" != "$PREVENV_PATH" ]; then
      PREV_PS1="$PS1"
      PREV_PATH="$PATH"
      PREVENV_PATH="$PWD"
      source $PWD/.venv
      source $VIRTUALENV_PATH/bin/activate
      echo "> Virtualenv `basename $VIRTUALENV_PATH` activated"
    fi
  fi
}

export PROMPT_COMMAND=handle_virtualenv

Just paste this code into your

$HOME/.bash_profile

and place

.venv

file with declaration like below:

VIRTUALENV_PATH=$HOME/.envs/sampleenvironment

And it should works like a charm. Script only for bash!