✨ add leancloud.Status

Author: aisk

leancloud/__init__.py

@@ -39,6 +39,7 @@
 from .query import Query
 from .relation import Relation
 from .role import Role
+from .status import Status
 from .sys_message import SysMessage
 from .user import User
 
@@ -61,6 +62,7 @@
     'Query',
     'Relation',
     'Role',
+    'Status',
     'SysMessage',
     'User',
     'client',

leancloud/status.py

@@ -0,0 +1,86 @@
+# coding: utf-8
+
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+from __future__ import unicode_literals
+
+
+from collections import namedtuple
+from datetime import datetime
+
+from leancloud import client
+from leancloud import utils
+from leancloud.query import Query
+from leancloud.user import User
+
+__author__ = 'asaka'
+
+
+
+StatusesCount = namedtuple('StatusesCount', ['total', 'unread'])
+
+class Status(object):
+    def __init__(self, **data):
+        self.id = None  # type: str
+        self.created_at = None  # type: datetime
+        self._data = data
+        self.inbox_type = 'default'
+
+    def get(self, key):
+        return self._data.get(key)
+
+    def set(self, key, value):
+        self._data[key] = value
+        return self
+
+    def send(self, query):
+        current_user = User.get_current()
+        if not current_user:
+            raise ValueError('Please sign in an user')
+
+        params = {
+            'inboxType': self.inbox_type,
+            'data': self._data,
+            'query': query.dump(),
+            'source': self._data.get('source') or current_user._to_pointer(),
+        }
+        params['query']['className'] = query._query_class._class_name
+
+        content = client.post('/statuses', params=params).json()
+        self.id = content['objectId']
+        self.created_at = utils.decode('createdAt', content['createdAt'])
+
+    def send_to_followers(self):
+        current_user = User.get_current()
+        if not current_user:
+            raise ValueError('Please sign in an user')
+
+        query = Query('_Follower').select('follower').equal_to('user', current_user)
+        self.send(query)
+
+    def send_private_status(self, target):
+        current_user = User.get_current()
+        if not current_user:
+            raise ValueError('Please sign in an user')
+
+        if not isinstance(target, User):
+            raise TypeError('target must be a leancloud.User')
+
+        query = Query('User').equal_to('objectId', target.id)
+        self.inbox_type = 'private'
+        self.send(query)
+
+    def destroy(self):
+        if not self.id:
+            raise ValueError('This status is not saved')
+        client.delete('/statuses/' + self.id)
+
+    @staticmethod
+    def count_unread_statuses(owner, inbox_type='default'):
+        params = {
+            'inboxType': inbox_type,
+            'owner': utils.encode(owner),
+        }
+        content = client.get('/subscribe/statuses/count', params).json()
+        return StatusesCount(total=content['total'], unread=content['unread'])

leancloud/user.py

@@ -7,6 +7,7 @@
 
 
 import threading
+from typing import Optional
 
 import six
 
@@ -58,7 +59,7 @@ def create_followee_query(cls, user_id):
         return query
 
     @classmethod
-    def get_current(cls):
+    def get_current(cls):  # type: () -> Optional[User]
         return getattr(thread_locals, 'current_user', None)
 
     @classmethod

tests/test_status.py

@@ -0,0 +1,68 @@
+# coding: utf-8
+
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+from __future__ import unicode_literals
+
+import os
+import time
+
+from nose.tools import assert_equal
+from nose.tools import assert_true
+
+import leancloud
+from leancloud import Status
+
+
+def setup():
+    leancloud.init(os.environ['APP_ID'], app_key=os.environ['APP_KEY'], master_key=os.environ['MASTER_KEY'])
+    leancloud.use_master_key(True)
+    users = leancloud.Query(leancloud.User).find()
+    for u in users:
+        u.destroy()
+    leancloud.use_master_key(False)
+
+    user1 = leancloud.User()
+    user1.set('username', 'user1_name')
+    user1.set('password', 'password')
+    user1.set_email('wow@leancloud.rocks')
+    user1.set_mobile_phone_number('18611111111')
+    user1.sign_up()
+
+
+def test_send():
+    status = Status(image='http://www.example.com', message='hello world!')
+    status.inbox_type = 'privateMessage'
+    query = leancloud.Query('User').equal_to('username', 'user1_name')
+    status.send(query)
+    assert status.id
+    assert status.created_at
+    status.destroy()
+
+
+def test_send_to_followers():
+    status = Status(image='http://www.example.com', message='hello world!')
+    status.send_to_followers()
+    assert status.id
+    assert status.created_at
+    status.destroy()
+
+
+def test_send_private_status():
+    status = Status(image='http://www.example.com', message='hello world!')
+    target = leancloud.User.create_without_data('foo')
+    status.send_private_status(target)
+    assert status.id
+    assert status.created_at
+    assert_equal(status.inbox_type, 'private')
+    status.destroy()
+
+
+def test_count_():
+    status = Status(image='http://www.example.com', message='hello world!')
+    status.send_private_status(leancloud.User.get_current())
+    result = Status.count_unread_statuses(leancloud.User.get_current(), 'private')
+    assert_equal(result.total, 1)
+    assert_equal(result.unread, 1)
+    status.destroy()