class Client:
"""This is the client class to connect to Diaspora.
-
"""
def __init__(self, pod, username, password):
"""
:type username: str
:param password: The password used to log in.
:type password: str
-
"""
self._token_regex = re.compile(r'content="(.*?)"\s+name="csrf-token')
self.pod = pod
self.session = requests.Session()
self._post_data = {}
self._setlogindata(username, password)
+ self._login()
+
+ def _sessionget(self, string):
+ """This method gets data from session.
+ Performs additional checks if needed.
+
+ Example:
+ To obtain 'foo' from pod one should call `_sessionget('foo')`.
+
+ :param string: URL to get without the pod's URL and slash eg. 'stream'.
+ :type string: str
+ """
+ return self.session.get('{0}/{1}'.format(self.pod, string))
def get_token(self):
"""This function gets a token needed for authentication in most cases
:returns: string -- token used to authenticate
-
"""
- r = self.session.get('{0}/stream'.format(self.pod))
+ r = self._sessionget('stream')
token = self._token_regex.search(r.text).group(1)
return token
.. note::
It should be called before _login() function.
"""
- #r = self.session.get(self.pod + '/users/sign_in')
- #token = self._token_regex.search(r.text).group(1)
self._username, self._password = username, password
self._login_data = {
'user[username]': self._username,
if r.status_code != 201: raise Exception('{0}: Login failed.'.format(r.status_code))
- def _set_post_data(self, text, aspect_id, photos):
- """This method prepares data for post.
-
+ def _setpostdata(self, text, aspect_id, photos):
+ """This function prepares data for posting.
+
:param text: Text to post.
:type text: str
:param aspect_id: Aspect id to send post to.
:type aspect_id: str
"""
- data = {'aspect_ids': aspect_id,
- 'status_message': {'text': text}}
+ data = {}
+ data['aspect_id'] = aspect_id
+ data['status_message'] = {'text': text}
if photos: data['photos'] = photos
self._post_data = data
- def _send_post(self):
- """This method sends a post to an aspect.
- It uses `self._post_data`.
+ def _post(self):
+ """Sends post to an aspect.
- .. note::
- _set_post_data() must be called before.
+ :returns: diaspy.models.Post -- the Post which has been created
"""
+ r = self.session.post('{0}/status_messages'.format(self.pod),
+ data=json.dumps(self._post_data),
+ headers={'content-type': 'application/json',
+ 'accept': 'application/json',
+ 'x-csrf-token': self.get_token()})
+ if r.status_code != 201: raise Exception('{0}: Post could not be posted.'.format(r.status_code))
+ return diaspy.models.Post(str(r.json()['id']), self)
def post(self, text, aspect_id='public', photos=None):
"""This function sends a post to an aspect
:type aspect_id: str
:returns: diaspy.models.Post -- the Post which has been created
-
"""
- self._set_post_data(text, aspect_id, photos)
- r = self.session.post('{0}/status_messages'.format(self.pod),
- data=json.dumps(self._post_data),
- headers={'content-type': 'application/json',
- 'accept': 'application/json',
- 'x-csrf-token': self.get_token()})
- if r.status_code != 201: raise Exception('{0}: Post could not be posted.'.format(r.status_code))
-
- return diaspy.models.Post(str(r.json()['id']), self)
+ self._setpostdata(text, aspect_id, photos)
+ post = self._post()
+ self._post_data = {}
+ return post
def get_user_info(self):
"""This function returns the current user's attributes.
:returns: dict -- json formatted user info.
-
"""
- r = self.session.get('{0}/bookmarklet'.format(self.pod))
+ r = self._sessionget('bookmarklet')
regex = re.compile(r'window.current_user_attributes = ({.*})')
userdata = json.loads(regex.search(r.text).group(1))
return userdata
def post_picture(self, filename):
+ """This method posts a picture to D*.
+
+ :param filename: Path to picture file.
+ :type filename: str
+ """
aspects = self.get_user_info()['aspects']
params = {}
params['photo[pending]'] = 'true'
"""This functions returns a list of posts found in the stream.
:returns: list -- list of Post objects.
-
"""
+ r = self._sessionget('stream.json')
- data = {'authenticity_token': self.get_token()}
- r = self.session.get('{0}/stream.json'.format(self.pod))
-
- if r.status_code != 200:
+ if r.status_code != 200:
raise Exception('wrong status code: {0}'.format(r.status_code))
stream = r.json()
-
- posts = []
-
- for post in stream:
- posts.append(diaspy.models.Post(str(post['id']), self))
-
- return posts
+ return [ diaspy.models.Post(str(post['id']), self) for post in stream ]
def get_notifications(self):
"""This functions returns a list of notifications.
:returns: list -- list of json formatted notifications
-
"""
+ r = self._sessionget('notifications.json')
- data = {'authenticity_token': self.get_token()}
- r = self.session.get('{0}/notifications.json'.format(self.pod))
-
- if r.status_code != 200:
+ if r.status_code != 200:
raise Exception('wrong status code: {0}'.format(r.status_code))
notifications = r.json()
posts the current user is being mentioned in.
:returns: list -- list of Post objects
-
"""
-
- data = {'authenticity_token': self.get_token()}
- r = self.session.get('/mentions.json'.format(self.pod))
+ r = self._sessionget('mentions.json')
if r.status_code != 200:
raise Exception('wrong status code: {0}'.format(r.status_code))
mentions = r.json()
-
- posts = []
-
- for post in mentions:
- posts.append(diaspy.models.Post(str(post['id']), self))
-
- return posts
+ return [ diaspy.models.Post(str(post['id']), self) for post in mentions ]
def get_tag(self, tag):
"""This functions returns a list of posts containing the tag.
:type tag: str
:returns: list -- list of Post objects
-
"""
-
- data = {'authenticity_token': self.get_token()}
- r = self.session.get('{0}/tags/{1}.json'.format(self.pod, tag))
+ r = self._sessionget('tags/{0}.json'.format(tag))
if r.status_code != 200:
raise Exception('wrong status code: {0}'.format(r.status_code))
tagged_posts = r.json()
+ return [ diaspy.models.Post(str(post['id']), self) for post in tagged_posts ]
- posts = []
+ def get_mailbox(self):
+ """This functions returns a list of messages found in the conversation.
+
+ :returns: list -- list of Conversation objects.
+ """
+ r = self._sessionget('conversations.json')
- for post in tagged_posts:
- posts.append(diaspy.models.Post(str(post['id']), self))
+ if r.status_code != 200:
+ raise Exception('wrong status code: {0}'.format(r.status_code))
- return posts
+ mailbox = r.json()
+ return [ diaspy.conversations.Conversation(str(conversation['conversation']['id']), self) for conversation in mailbox ]
def add_user_to_aspect(self, user_id, aspect_id):
""" this function adds a user to an aspect.
def remove_aspect(self, aspect_id):
""" This function adds a new aspect.
"""
-
data = {'authenticity_token': self.get_token()}
r = self.session.delete('{0}/aspects/{1}'.format(self.pod, aspect_id),
if r.status_code != 404:
raise Exception('wrong status code: {0}'.format(r.status_code))
- def get_mailbox(self):
- """This functions returns a list of messages found in the conversation.
-
- :returns: list -- list of Conversation objects.
-
- """
-
- data = {'authenticity_token': self.get_token()}
- r = self.session.get('{0}/conversations.json'.format(self.pod))
-
- if r.status_code != 200:
- raise Exception('wrong status code: {0}'.format(r.status_code))
-
- mailbox = r.json()
-
- conversations = []
-
- for conversation in mailbox:
- conversations.append(diaspy.conversations.Conversation(str(conversation['conversation']['id']), self))
-
- return conversations
-
def new_conversation(self, contacts, subject, text):
- """ start a new conversation
+ """Start a new conversation.
:param contacts: recipients ids, no guids, comma sperated.
:type contacts: str
:type subject: str
:param text: text of the message.
:type text: str
-
"""
-
data = {'contact_ids': contacts,
'conversation[subject]': subject,
'conversation[text]': text,