# -*- coding: utf-8 -*-

from elasticsearch import Elasticsearch
from elasticsearch.client import IndicesClient
from elasticsearch.exceptions import NotFoundError
from elasticsearch.exceptions import TransportError
from storm import Bolt, log


class ObservationBolt(Bolt):
    def initialize(self, conf, context):
        log('Bolt id-%s is initializing.' % id(self))
        # TODO: Make connection params configurable.
        self.es = Elasticsearch(hosts=[{'host': 'localhost', 'port': 9200}])
        ic = IndicesClient(self.es)

        try:
            ic.get_mapping(
                index='observation',
                doc_type='user'
                )
        except NotFoundError:
            doc_type = {
                'user': {
                    'properties': {
                        'events': {
                            'type': 'nested'
                            }
                        }
                    }
                }
            ic.put_mapping(
                index='observation',
                ignore_conflicts=True,
                doc_type='user',
                body=doc_type
                )

    def process(self, tup):
        event = dict(
            timestamp=tup.values[0],
            path=tup.values[1]
            )

        kwargs = dict(
            id=tup.values[2]
            )

        try:
            events = self.es.get('observation', tup.values[2], 'user')
            body = {'events': events['_source']['events'] + [event]}
        except NotFoundError:
            kwargs['op_type'] = 'create'
            body = {'events': [event]}
        except TransportError, e:
            # TODO: What is going wrong here?
            log('[TransportError] Get failed: %s' % e)
            return

        try:
            self.es.index('observation', 'user', body, **kwargs)
        except TransportError, e:
            # TODO: What is going wrong here?
            log('[TransportError] Index failed: %s' % e)

ObservationBolt().run()

@@ -596,35 +596,36 @@
 dex='observation
+s
 ',%0A
-
                 
@@ -1014,16 +1014,17 @@
 ervation
+s
 ',%0A     
@@ -1397,16 +1397,17 @@
 ervation
+s
 ', tup.v
@@ -1763,32 +1763,32 @@
 n%0A%0A        try:%0A
-
             self
@@ -1809,16 +1809,17 @@
 ervation
+s
 ', 'user

Fix index name typo