Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Merge pull request #2 from dcramer/master

Added arbitrary score value and cleaned up setup.py
  • Loading branch information...
commit a271371e1eef11daeddd8695de798d58fb2fdadd 2 parents 168ff59 + 97b978c
Brett Hoerner authored
Showing with 63 additions and 3 deletions.
  1. +3 −0  README.rst
  2. +3 −0  setup.py
  3. +51 −0 tests/tests.py
  4. +6 −3 timak/timelines.py
3  README.rst
View
@@ -19,6 +19,7 @@ Timelines are unique sets of objects (unique by the ID you provide) ordered by a
>>> tl = Timeline(connection=conn, max_items=3)
+ >>> # t1.add("key", "unique_id", "score")
>>> tl.add("brett:tweets", 1, datetime(2011, 1, 1))
[1]
>>> tl.add("brett:tweets", 2, datetime(2011, 1, 2))
@@ -30,6 +31,8 @@ Timelines are unique sets of objects (unique by the ID you provide) ordered by a
>>> tl.delete("brett:tweets", 2, datetime(2011, 1, 2))
[4, 3]
+If you provide a ``datetime.datetime`` value to score Timak will automatically convert to a sortable score value.
+
As you can see the default order is descending by the date you provide, and the object IDs are returned by default. You can also provide an ``obj_data`` argument (must be JSON serializable) which will be returned instead.
>>> tl.add("brett:tweets", 5, datetime(2011, 1, 5), obj_data={'body': 'Hello world, this is my first tweet'})
3  setup.py
View
@@ -7,6 +7,9 @@
author_email='brett@bretthoerner.com',
url='http://github.com/bretthoerner/timak',
packages=find_packages(),
+ test_suite='unittest2.collector',
+ install_requires=['riak'],
+ tests_require=['unittest2'],
classifiers=[
"Intended Audience :: Developers",
"Intended Audience :: System Administrators",
51 tests/tests.py
View
@@ -71,3 +71,54 @@ def test_multi_writers(self):
t2.add(self.key, 2, now + timedelta(minutes=1))
self.assertEqual(t1.get(self.key), [2, 1])
+
+ def test_timestamp_scores(self):
+ timeline = Timeline(connection=self.c1, bucket=self.bucket, max_items=3)
+ now = datetime.utcnow()
+
+ timeline.add(self.key, 1, now)
+ timeline.add(self.key, 2, now + timedelta(seconds=1))
+ timeline.add(self.key, 3, now + timedelta(seconds=2))
+ results = timeline.get(self.key)
+ self.assertEqual(len(results), 3)
+ self.assertEquals(results[0], 3)
+ self.assertEquals(results[1], 2)
+ self.assertEquals(results[2], 1)
+
+ def test_non_timestamp_scores(self):
+ timeline = Timeline(connection=self.c1, bucket=self.bucket, max_items=3)
+
+ timeline.add(self.key, 1, 3)
+ timeline.add(self.key, 2, 2)
+ timeline.add(self.key, 3, 1)
+ results = timeline.get(self.key)
+ self.assertEqual(len(results), 3)
+ self.assertEquals(results[0], 1)
+ self.assertEquals(results[1], 2)
+ self.assertEquals(results[2], 3)
+
+ def test_score_scoping(self):
+ timeline = Timeline(connection=self.c1, bucket=self.bucket, max_items=3)
+
+ timeline.add(self.key, 1, 3)
+ timeline.add(self.key, 2, 2)
+ timeline.add(self.key, 3, 1)
+ results = timeline.get(self.key)
+ self.assertEqual(len(results), 3)
+ self.assertEquals(results[0], 1)
+ self.assertEquals(results[1], 2)
+ self.assertEquals(results[2], 3)
+
+ timeline.add(self.key, 4, 0)
+ results = timeline.get(self.key)
+ self.assertEqual(len(results), 3)
+ self.assertEquals(results[0], 1)
+ self.assertEquals(results[1], 2)
+ self.assertEquals(results[2], 3)
+
+ timeline.add(self.key, 5, 5)
+ results = timeline.get(self.key)
+ self.assertEqual(len(results), 3)
+ self.assertEquals(results[0], 5)
+ self.assertEquals(results[1], 1)
+ self.assertEquals(results[2], 2)
9 timak/timelines.py
View
@@ -42,7 +42,7 @@ def _dict_to_list(self, d):
if not d: return []
l = d.values()
reverse = self.order == 'desc'
- l.sort(key=lambda x: x['timestamp'], reverse=reverse)
+ l.sort(key=lambda x: x['score'], reverse=reverse)
return l
def _list_to_data(self, l):
@@ -98,12 +98,15 @@ def get(self, key, raw=False):
def _make_op(action):
assert action in ('add', 'delete')
- def _op(self, key, uniq_ident, obj_datetime, obj_data=None, raw=False):
+ def _op(self, key, uniq_ident, obj_score, obj_data=None, raw=False):
now = self._datetime_to_js(datetime.datetime.utcnow())
obj, data = self._get_obj_and_data(key, write_merged=False)
+ if isinstance(obj_score, datetime.datetime):
+ obj_score = self._datetime_to_js(obj_score)
+
new_item = {'id': uniq_ident,
- 'timestamp': self._datetime_to_js(obj_datetime),
+ 'score': obj_score,
'modified': now}
if obj_data:
new_item['data'] = obj_data
Please sign in to comment.
Something went wrong with that request. Please try again.