This is a pre-production deployment of Warehouse. Changes made here affect the production instance of PyPI (
Help us improve Python packaging - Donate today!

Log actions to Elastic Search via bulk upload

Project Description

bes is a flexible bulk uploader for Elastic Search. At the moment, it only supports index uploads over UDP. You can retrieve the logged data from Elastic Search and analyze it dynamically using Kibana (source). This package just makes the initial upload to Elastic Search as painless as possible.

The connection parameters and other configuration options are stored in bes.DEFAULT. Override them as you see fit (e.g. to connect to a remote Elastic Search server):

>>> import bes
>>> bes.DEFAULT['host'] = ''

Then log away:

>>> bes.log(type='record', user='jdoe', action='swims')

If you are so inclined, you can override the DEFAULT index on a per-call basis:

>>> bes.log(index='my-index', type='my-type', user='jdoe', action='bikes')

The log generated by the above will look like:

  "@timestamp": "2013-09-26T16:34:09.179048",
  "@version": 1,
  "action": "bikes",
  "user": "jdoe"

following Jordan Sissel’s new format for Logstash:

You should specify a unique type for each record you create, because Elastic Search doesn’t recalculate its mapping if you post a new event that uses an old key with a new data type. If you are only logging a single record type, you can configure a global default:

>>> bes.DEFAULT['type'] = 'record'
>>> bes.log(user='jdoe', action='runs')


Although the core of the library is framework-agnostic, we’ll be using this to log events in a Django website. There are a few helpers for extracting information from HttpRequest to make this as easy as possible. Drop it into your views with somthing like:

import django.http
imort django.shortcuts
import bes.django
from polls.models import Poll

def detail(request, poll_id):
        p = Poll.objects.get(pk=poll_id)
    except Poll.DoesNotExist:
            request=request, type='404', poll_id=poll_id)
        raise django.http.Http404
        request=request, type='poll-detail',
    return django.shortcutsrender_to_response(
        'polls/detail.html', {'poll': p})

You can also override bes’ defaults in your Django config:


The Django config names match the uppercased DEFAULT keys with a BULK_ELASTIC_SEARCH_LOGGING_ prefix (for example, hostBULK_ELASTIC_SEARCH_LOGGING_HOST).


To log activity for critical functions, bes has tracing decorators:

import bes.trace

def my_function():
    return 1 + 2 / 3

By default this logs with bes.log using the decorated function’s name (e.g. my_function) as the log type. You can override this if you want something different:

def my_logger(*args, **kwargs):
    print((args, kwargs))

@bes.trace.trace(type='my-type', logger=my_logger)
def my_function():
    return 1 + 2 / 3

Additional trace arguments are passed straight through to the logger:

@bes.trace.trace(index='my-index', type='my-type', sport='triathalon')
def my_function():
    return 1 + 2 / 3

By default, you’ll get separate logged messages before the decorated function starts, after it completes, or if it raises an exception. An action argument is passed to the logger in each case, with values of start, complete, and error respectively. In the error case, the logger is also passed the string-ified exception (error=str(e)) to help with debugging the problem. If you don’t want one or more of these logged messages, you can turn them off:

@bes.trace.trace(start=False, error=False, complete=False):
def my_function():
    return 1 + 2 / 3

although if you turn them all off, you’re not going to get any messages at all ;).

The trace decorator uses wrapt (if it’s installed) to create introspection-preserving decorators. If wrapt is not installed, we fallback to the builtin functools.wraps which handles the basics like docstrings, but doesn’t do as well for more involved introspection (e.g. inspect.getsource).


Testing uses unittest’s automatic test discovery. Run the test suite with:

$ python -m unittest discover

For Python <3.3, the test suite requires the external mock package, which is bundled as unittest.mock in Python 3.3.

Release History

Release History

This version
History Node


History Node


History Node


Download Files

Download Files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

File Name & Checksum SHA256 Checksum Help Version File Type Upload Date
bes-0.3.tar.gz (9.0 kB) Copy SHA256 Checksum SHA256 Source Sep 27, 2013

Supported By

WebFaction WebFaction Technical Writing Elastic Elastic Search Pingdom Pingdom Monitoring Dyn Dyn DNS Sentry Sentry Error Logging CloudAMQP CloudAMQP RabbitMQ Heroku Heroku PaaS Kabu Creative Kabu Creative UX & Design Fastly Fastly CDN DigiCert DigiCert EV Certificate Rackspace Rackspace Cloud Servers DreamHost DreamHost Log Hosting