def riakGetStock(searchVal, riakIP):
    myData = []
    myBucket = RiakClient(protocol="pbc", host=riakIP, pb_port=8087).bucket("stocks")
    for key in myBucket.get_index("ticker_bin", searchVal):  # get all from 2002 to 2012
        value = json.loads(myBucket.get(key).data)
        myData.append([(value["CLOSE"]), (value["VOLUME"]), str(value["DATE"])])
    return myData

def writeSinglePair(pair, bucketName, riakIP):

    rc = RiakClient(protocol="pbc", host=riakIP, pb_port=8087)
    bucket = rc.bucket(bucketName)

    # create key value pairs to stock in riak
    key = str(str(pair[0]) + "_" + str(pair[1]))
    val = {
        "StockA": pair[0],
        "StockB": pair[1],
        "Date": pair[2],
        "CloseA": pair[3],
        "CloseB": pair[4],
        "ZScore": pair[5],
        "Beta": pair[6],
        "SignalMean": pair[7],
        "SignalSD": pair[8],
    }
    myDate = pair[2].split("-")
    obj = RiakObject(rc, bucket, key)

    # add 2i tags
    obj.add_index("stocka_bin", str(pair[0]))
    obj.add_index("stockb_bin", str(pair[3]))
    obj.add_index("year_int", int(myDate[0]))
    obj.add_index("month_int", int(myDate[1]))
    obj.add_index("day_int", int(myDate[2]))
    obj.content_type = "text/json"
    obj.data = val
    obj.data = json.dumps(val)
    # store
    obj.store()

    # return a list of written pairs
    return pair

    def test_close_underlying_socket_retry(self):
        c = RiakClient(PB_HOST, PB_PORT, transport_class=RiakPbcTransport,
                                         transport_options={"max_attempts": 2})

        bucket = c.bucket('bucket_test_close')
        rand = self.randint()
        obj = bucket.new('barbaz', rand)
        obj.store()
        obj = bucket.get('barbaz')
        self.assertTrue(obj.exists)
        self.assertEqual(obj.bucket.name, 'bucket_test_close')
        self.assertEqual(obj.key, 'barbaz')
        self.assertEqual(obj.data, rand)

        # Close the underlying socket. This gets a bit sketchy,
        # since we are reaching into the internals, but there is
        # no other way to get at the socket
        conns = c._cm.conns
        conns[0].sock.close()

        # This should work, since we have a retry
        obj = bucket.get('barbaz')
        self.assertTrue(obj.exists)
        self.assertEqual(obj.bucket.name, 'bucket_test_close')
        self.assertEqual(obj.key, 'barbaz')
        self.assertEqual(obj.data, rand)

 def test_uses_client_id_if_given(self):
     self.host = PB_HOST
     self.port = PB_PORT
     zero_client_id = "\0\0\0\0"
     c = RiakClient(PB_HOST, PB_PORT,
                         transport_class = RiakPbcTransport,
                         client_id = zero_client_id)
     self.assertEqual(zero_client_id, c.get_client_id()) #

def register(host,user_id,time):
	currenttld = extractdom(host)
	riakclient = RiakClient()
	#bucket : trackers, key:user_id, object:json map
	str_bucket = 'trackers'
	bucket = riakclient.bucket(str_bucket)

	str_key = user_id
	obj = RiakObject(riakclient, bucket, str_key)

    def conn(self):
        """Open connection on Riak DataBase"""
        client = RiakClient(
            protocol=self.conf.get("protocol"),
            http_port=self.conf.get("port"),
            host=self.conf.get("host"))

        conn = client.bucket(self.conf.get("db"))
        conn.enable_search()
        return conn

    def from_config(cls, config):
        config = config.copy()
        bucket_prefix = config.pop('bucket_prefix')
        load_bunch_size = config.pop('load_bunch_size',
                                     cls.DEFAULT_LOAD_BUNCH_SIZE)
        mapreduce_timeout = config.pop('mapreduce_timeout',
                                       cls.DEFAULT_MAPREDUCE_TIMEOUT)
        transport_type = config.pop('transport_type', 'http')

        host = config.get('host', '127.0.0.1')
        port = config.get('port')
        prefix = config.get('prefix', 'riak')
        mapred_prefix = config.get('mapred_prefix', 'mapred')
        client_id = config.get('client_id')
        transport_options = config.get('transport_options', {})

        client_args = dict(
            host=host, prefix=prefix, mapred_prefix=mapred_prefix,
            protocol=transport_type, client_id=client_id,
            transport_options=transport_options)

        if port is not None:
            client_args['port'] = port

        client = RiakClient(**client_args)
        # Some versions of the riak client library use simplejson by
        # preference, which breaks some of our unicode assumptions. This makes
        # sure we're using stdlib json which doesn't sometimes return
        # bytestrings instead of unicode.
        client.set_encoder('application/json', json.dumps)
        client.set_encoder('text/json', json.dumps)
        client.set_decoder('application/json', json.loads)
        client.set_decoder('text/json', json.loads)
        return cls(client, bucket_prefix, load_bunch_size=load_bunch_size,
                   mapreduce_timeout=mapreduce_timeout)

def riakSearchData(searchBucket, searchTerm, searchVal1, searchVal2, riakIP):
    myData = {}  # empty dict
    myBucket = RiakClient(protocol="pbc", host=riakIP, pb_port=8087).bucket(searchBucket)
    # check wether 1 or 2 search terms
    if searchVal2 != None:
        for key in myBucket.get_index(searchTerm, searchVal1, searchVal2):  # get all keys with 2i match
            myData[key] = json.loads(myBucket.get(key).data)  # store data for each key
    else:
        for key in myBucket.get_index(searchTerm, searchVal1):  # get all keys with 2i match
            myData[key] = json.loads(myBucket.get(key).data)  # store data for each key
    return myData

    def clear_expired(cls):
        # TODO: get_keys() method is not efficent operation for searching sessions, use search feature or mapreduce or ...
        client = RiakClient(host=RIAK_DB_HOST,
                            port=RIAK_DB_PORT,
                            transport_class=RiakPbcTransport)
        sessions = client.bucket(RIAK_DB_BUCKET)

        now = int(timezone.now().strftime("%s"))
        for key in sessions.get_keys():
            data = sessions.get(key)
            if int(data.get_usermeta()['expire_age']) < now:
                data.delete()

class RiakStorage(AbstractStorage):

    bucket_types = {
        'users':         'egara-lww',
        'users-current': 'egara-unique',
        'imap-events':   'egara-lww',
        'imap-folders':  'egara-lww',
        'imap-folders-current':  'egara-unique',
        'imap-message-timeline': 'egara-lww'
    }

    def __init__(self, *args, **kw):
        riak_host = 'localhost'
        riak_port = 8098

        self.client = RiakClient(
            protocol='http',
            host=conf['STORAGE'].get('riak_host', riak_host),
            http_port=conf['STORAGE'].get('riak_port', riak_port)
        )
        self.client.set_decoder('application/octet-stream', self._decode_binary)
        self.users_cache = CachedDict(ttl=10)

    def _decode_binary(self, data):
        return str(data).encode("utf-8")

    def _get_bucket(self, bucketname):
        _type = self.bucket_types.get(bucketname, None)
        if _type:
            return self.client.bucket_type(_type).bucket(bucketname)

        return None


    def get(self, key, index, doctype=None, fields=None, **kw):
        """
            Standard API for accessing key/value storage
        """
        result = None
        log.debug("Riak get key %r from %r", key, index)

        try:
            bucket = self._get_bucket(index)
            res = bucket.get(key)
            if res and res.data:
                result = res.data

        except Exception, e:
            log.warning("Riak exception: %s", str(e))
            result = None

        return result

 def __init__(self, option):
   print "Here we be initializing yon connections and things: %r" % option
   self.riak = RiakClient(host=option['RIAKTS_HOST'], pb_port=int(option['RIAKTS_PORT']))
   self.nodes_table = option['RIAKTS_NODES_TABLE']
   self.table_name = option['RIAKTS_TABLE_NAME']
   self.family = option['RIAKTS_FAMILY']
   self.nodes = self.riak.bucket(self.nodes_table)

class DS_Riak(DS):
    """ Riak DS
    """
    def __init__(self, host=PB_HOST, port=PB_PORT, bucketname='_scads'):
        """
        Params:
            host: Hostname or IP Address of the riak database
            port: Port of the riak database
            bucketname: Where to store stuff
        """
        self.client = RiakClient(host, port,
                                 transport_class=RiakPbcTransport,
                                 transport_options={'max_attempts': 3})
        self.bucket = self.client.bucket(bucketname)

    def store(self, key, value, indexes=None):
        obj = self.bucket.get(key)
        obj.set_data(value)
        if indexes is not None:
            for index in indexes:
                obj.add_index('scads_bin', index)
        obj.store()

    def get_index(self, index):
        return self.bucket.get_index('scads_bin', index)

    def get(self, key):
        d = self.bucket.get(key).get_data()
        if d is None:
            return None
        return d

    def clear(self):
        for key in self.bucket.get_keys():
            self.bucket.get(key).delete()

    def __init__(
        self, protocol="pbc", nodes=[], objects_bucket_name="objects",
            activities_bucket_name="activities", **kwargs):

        self._riak_backend = RiakClient(protocol=protocol, nodes=nodes)

        r_value = kwargs.get("r")
        w_value = kwargs.get("w")
        dw_value = kwargs.get("dw")
        pr_value = kwargs.get("pr")
        pw_value = kwargs.get("pw")

        self._objects = self._riak_backend.bucket(objects_bucket_name)
        self._activities = self._riak_backend.bucket(activities_bucket_name)

        if r_value:
            self._objects.r = r_value
            self._activities.r = r_value

        if w_value:
            self._objects.w = w_value
            self._activities.w = w_value

        if dw_value:
            self._objects.dw = dw_value
            self._activities.dw = dw_value

        if pr_value:
            self._objects.pr = pr_value
            self._activities.pr = pr_value

        if pw_value:
            self._objects.pw = pw_value
            self._activities.pw = pw_value

def getDataByTicker(ticker, dataSource, start, end, riakIP):

    rc = RiakClient(protocol="pbc", host=riakIP, pb_port=8087)
    # get daily data for each ticker
    gtemp = pd.DataFrame()
    bucket = rc.bucket("stocks")
    try:
        gtemp = DataReader(ticker, dataSource, start, end)
        print ticker
    except:
        pass

        # didnt get any data
    if len(gtemp) == 0:
        return 0
    # got data
    else:

        for j in range(0, len(gtemp.index)):

            # upload json to Riak Bucket
            date = gtemp.index[j].date()
            riakKey = str(ticker + "_" + str(date))
            riakVal = {
                "OPEN": gtemp.values[j, 0],
                "HIGH": gtemp.values[j, 1],
                "LOW": gtemp.values[j, 2],
                "CLOSE": gtemp.values[j, 3],
                "VOLUME": gtemp.values[j, 4],
                "DATE": str(date),
                "TICKER": str(ticker),
            }

            obj = RiakObject(rc, bucket, riakKey)

            obj.add_index("ticker_bin", str(ticker))
            obj.add_index("year_int", int(date.year))
            obj.add_index("month_int", int(date.month))
            obj.add_index("day_int", int(date.day))

            obj.content_type = "text/json"
            # obj.data = riakVal
            obj.data = json.dumps(riakVal)
            obj.store()

    return len(gtemp.index)

 def _get_client(self):
     """Get client connection."""
     if self._client is None or not self._client.is_alive():
         self._client = RiakClient(protocol=self.protocol,
                                   host=self.host,
                                   pb_port=self.port)
         self._client.resolver = last_written_resolver
     return self._client

class UserService:
    def __init__(self):
        self.users_bucket = settings.RIAK_USERS_BUCKET
        self.riak = RiakClient(host=settings.RIAK_HOST,
                               port=settings.RIAK_PORT)

    def save(self, user_data):
        key = self.generate_key()
        password = user_data['password']
        user_data['encrypted_password'] = make_password('sha1', password)
        del user_data['password']
        user = self.riak.bucket(self.users_bucket).new(key, user_data)
        user.store()
        return key

    def get(self, user_id):
        user = self.riak.bucket(settings.RIAK_USERS_BUCKET).get(str(user_id))
        if user.exists():
            return user.get_data()
        else:
            return None

    def login(self, username, password):
        try:
            user = self.riak.search(settings.RIAK_USERS_BUCKET,
                                    "username:%s" % username).run()
        except ValueError:
            user = None

        if user:
            user = user[0]
            user.set_bucket(str(user.get_bucket()))
            user.set_key(str(user.get_key()))
            user_data = user.get().get_data()
            if check_password(password, user_data['encrypted_password']):
                return user
            else:
                return False
        else:
            return False

    def generate_key(self):
        return uuid.uuid1().hex

    def __init__(self, *args, **kw):
        riak_host = 'localhost'
        riak_port = 8098

        self.client = RiakClient(
            protocol='http',
            host=conf['STORAGE'].get('riak_host', riak_host),
            http_port=conf['STORAGE'].get('riak_port', riak_port)
        )
        self.client.set_decoder('application/octet-stream', self._decode_binary)
        self.users_cache = CachedDict(ttl=10)

 def __init__(self, host=PB_HOST, port=PB_PORT, bucketname='_scads'):
     """
     Params:
         host: Hostname or IP Address of the riak database
         port: Port of the riak database
         bucketname: Where to store stuff
     """
     self.client = RiakClient(host, port,
                              transport_class=RiakPbcTransport,
                              transport_options={'max_attempts': 3})
     self.bucket = self.client.bucket(bucketname)

 def __init__(self, **client_args):
     self._closed = False
     self._raw_client = RiakClient(**client_args)
     # Some versions of the riak client library use simplejson by
     # preference, which breaks some of our unicode assumptions. This makes
     # sure we're using stdlib json which doesn't sometimes return
     # bytestrings instead of unicode.
     self._client.set_encoder("application/json", json.dumps)
     self._client.set_encoder("text/json", json.dumps)
     self._client.set_decoder("application/json", json.loads)
     self._client.set_decoder("text/json", json.loads)

    def test_close_underlying_socket_fails(self):
        c = RiakClient(PB_HOST, PB_PORT, transport_class=RiakPbcTransport)

        bucket = c.bucket('bucket_test_close')
        rand = self.randint()
        obj = bucket.new('foo', rand)
        obj.store()
        obj = bucket.get('foo')
        self.assertTrue(obj.exists)
        self.assertEqual(obj.bucket.name, 'bucket_test_close')
        self.assertEqual(obj.key, 'foo')
        self.assertEqual(obj.data, rand)

        # Close the underlying socket. This gets a bit sketchy,
        # since we are reaching into the internals, but there is
        # no other way to get at the socket
        conns = c._cm.conns
        conns[0].sock.close()

        # This shoud fail with a socket error now
        self.assertRaises(socket.error, bucket.get, 'foo')