from tweepy.mixins import HashableID
-class ResultSet(list):
- """A list like object that holds results from a Twitter API query."""
-
- def __init__(self, max_id=None, since_id=None):
- super().__init__()
- self._max_id = max_id
- self._since_id = since_id
-
- @property
- def max_id(self):
- if self._max_id:
- return self._max_id
- ids = self.ids()
- # Max_id is always set to the *smallest* id, minus one, in the set
- return (min(ids) - 1) if ids else None
-
- @property
- def since_id(self):
- if self._since_id:
- return self._since_id
- ids = self.ids()
- # Since_id is always set to the *greatest* id in the set
- return max(ids) if ids else None
-
- def ids(self):
- return [item.id for item in self if hasattr(item, 'id')]
-
-
class Model:
def __init__(self, api=None):
return f'{self.__class__.__name__}({", ".join(state)})'
-class Status(Model, HashableID):
-
- @classmethod
- def parse(cls, api, json):
- status = cls(api)
- setattr(status, '_json', json)
- for k, v in json.items():
- if k == 'user':
- try:
- user = api.parser.model_factory.user.parse(api, v)
- except AttributeError:
- user = User.parse(api, v)
- setattr(status, 'author', user)
- setattr(status, 'user', user) # DEPRECIATED
- elif k == 'created_at':
- setattr(status, k, parsedate_to_datetime(v))
- elif k == 'source':
- if '<' in v:
- # At this point, v should be of the format:
- # <a href="{source_url}" rel="nofollow">{source}</a>
- setattr(status, k, v[v.find('>') + 1:v.rfind('<')])
- start = v.find('"') + 1
- end = v.find('"', start)
- setattr(status, 'source_url', v[start:end])
- else:
- setattr(status, k, v)
- setattr(status, 'source_url', None)
- elif k == 'retweeted_status':
- setattr(status, k, Status.parse(api, v))
- elif k == 'quoted_status':
- setattr(status, k, Status.parse(api, v))
- elif k == 'place':
- if v is not None:
- setattr(status, k, Place.parse(api, v))
- else:
- setattr(status, k, None)
- else:
- setattr(status, k, v)
- return status
+class ResultSet(list):
+ """A list like object that holds results from a Twitter API query."""
- def destroy(self):
- return self._api.destroy_status(self.id)
+ def __init__(self, max_id=None, since_id=None):
+ super().__init__()
+ self._max_id = max_id
+ self._since_id = since_id
- def retweet(self):
- return self._api.retweet(self.id)
+ @property
+ def max_id(self):
+ if self._max_id:
+ return self._max_id
+ ids = self.ids()
+ # Max_id is always set to the *smallest* id, minus one, in the set
+ return (min(ids) - 1) if ids else None
- def retweets(self):
- return self._api.retweets(self.id)
+ @property
+ def since_id(self):
+ if self._since_id:
+ return self._since_id
+ ids = self.ids()
+ # Since_id is always set to the *greatest* id in the set
+ return max(ids) if ids else None
- def favorite(self):
- return self._api.create_favorite(self.id)
+ def ids(self):
+ return [item.id for item in self if hasattr(item, 'id')]
-class User(Model, HashableID):
+class BoundingBox(Model):
@classmethod
def parse(cls, api, json):
- user = cls(api)
- setattr(user, '_json', json)
- for k, v in json.items():
- if k == 'created_at':
- setattr(user, k, parsedate_to_datetime(v))
- elif k == 'status':
- setattr(user, k, Status.parse(api, v))
- elif k == 'following':
- # twitter sets this to null if it is false
- if v is True:
- setattr(user, k, True)
- else:
- setattr(user, k, False)
- else:
- setattr(user, k, v)
- return user
-
- @classmethod
- def parse_list(cls, api, json_list):
- if isinstance(json_list, list):
- item_list = json_list
- else:
- item_list = json_list['users']
-
- results = ResultSet()
- for obj in item_list:
- results.append(cls.parse(api, obj))
- return results
-
- def timeline(self, **kwargs):
- return self._api.user_timeline(user_id=self.id, **kwargs)
-
- def friends(self, **kwargs):
- return self._api.friends(user_id=self.id, **kwargs)
-
- def followers(self, **kwargs):
- return self._api.followers(user_id=self.id, **kwargs)
-
- def follow(self):
- self._api.create_friendship(user_id=self.id)
- self.following = True
-
- def unfollow(self):
- self._api.destroy_friendship(user_id=self.id)
- self.following = False
-
- def lists_memberships(self, *args, **kwargs):
- return self._api.lists_memberships(user_id=self.id, *args, **kwargs)
+ result = cls(api)
+ if json is not None:
+ for k, v in json.items():
+ setattr(result, k, v)
+ return result
- def lists_ownerships(self, *args, **kwargs):
- return self._api.lists_ownerships(user_id=self.id, *args, **kwargs)
+ def origin(self):
+ """
+ Return longitude, latitude of southwest (bottom, left) corner of
+ bounding box, as a tuple.
- def lists_subscriptions(self, *args, **kwargs):
- return self._api.lists_subscriptions(user_id=self.id, *args, **kwargs)
+ This assumes that bounding box is always a rectangle, which
+ appears to be the case at present.
+ """
+ return tuple(self.coordinates[0][0])
- def lists(self, *args, **kwargs):
- return self._api.lists_all(user_id=self.id, *args, **kwargs)
+ def corner(self):
+ """
+ Return longitude, latitude of northeast (top, right) corner of
+ bounding box, as a tuple.
- def followers_ids(self, *args, **kwargs):
- return self._api.followers_ids(user_id=self.id, *args, **kwargs)
+ This assumes that bounding box is always a rectangle, which
+ appears to be the case at present.
+ """
+ return tuple(self.coordinates[0][2])
class DirectMessage(Model):
return source, target
-class SavedSearch(Model):
-
- @classmethod
- def parse(cls, api, json):
- ss = cls(api)
- for k, v in json.items():
- if k == 'created_at':
- setattr(ss, k, parsedate_to_datetime(v))
- else:
- setattr(ss, k, v)
- return ss
-
- def destroy(self):
- return self._api.destroy_saved_search(self.id)
-
-
-class SearchResults(ResultSet):
-
- @classmethod
- def parse(cls, api, json):
- metadata = json['search_metadata']
- results = SearchResults()
- results.refresh_url = metadata.get('refresh_url')
- results.completed_in = metadata.get('completed_in')
- results.query = metadata.get('query')
- results.count = metadata.get('count')
- results.next_results = metadata.get('next_results')
-
- try:
- status_model = api.parser.model_factory.status
- except AttributeError:
- status_model = Status
-
- for status in json['statuses']:
- results.append(status_model.parse(api, status))
- return results
-
-
class List(Model):
@classmethod
)
+class Media(Model):
+
+ @classmethod
+ def parse(cls, api, json):
+ media = cls(api)
+ for k, v in json.items():
+ setattr(media, k, v)
+ return media
+
+
+class Place(Model):
+
+ @classmethod
+ def parse(cls, api, json):
+ place = cls(api)
+ for k, v in json.items():
+ if k == 'bounding_box':
+ # bounding_box value may be null (None.)
+ # Example: "United States" (id=96683cc9126741d1)
+ if v is not None:
+ t = BoundingBox.parse(api, v)
+ else:
+ t = v
+ setattr(place, k, t)
+ elif k == 'contained_within':
+ # contained_within is a list of Places.
+ setattr(place, k, Place.parse_list(api, v))
+ else:
+ setattr(place, k, v)
+ return place
+
+ @classmethod
+ def parse_list(cls, api, json_list):
+ if isinstance(json_list, list):
+ item_list = json_list
+ else:
+ item_list = json_list['result']['places']
+
+ results = ResultSet()
+ for obj in item_list:
+ results.append(cls.parse(api, obj))
+ return results
+
+
class Relationship(Model):
@classmethod
def parse(cls, api, json):
return result
-class JSONModel(Model):
+class SavedSearch(Model):
@classmethod
def parse(cls, api, json):
- return json
+ ss = cls(api)
+ for k, v in json.items():
+ if k == 'created_at':
+ setattr(ss, k, parsedate_to_datetime(v))
+ else:
+ setattr(ss, k, v)
+ return ss
+
+ def destroy(self):
+ return self._api.destroy_saved_search(self.id)
-class IDModel(Model):
+class SearchResults(ResultSet):
@classmethod
def parse(cls, api, json):
- if isinstance(json, list):
- return json
- else:
- return json['ids']
+ metadata = json['search_metadata']
+ results = SearchResults()
+ results.refresh_url = metadata.get('refresh_url')
+ results.completed_in = metadata.get('completed_in')
+ results.query = metadata.get('query')
+ results.count = metadata.get('count')
+ results.next_results = metadata.get('next_results')
+ try:
+ status_model = api.parser.model_factory.status
+ except AttributeError:
+ status_model = Status
-class BoundingBox(Model):
+ for status in json['statuses']:
+ results.append(status_model.parse(api, status))
+ return results
+
+
+class Status(Model, HashableID):
@classmethod
def parse(cls, api, json):
- result = cls(api)
- if json is not None:
- for k, v in json.items():
- setattr(result, k, v)
- return result
+ status = cls(api)
+ setattr(status, '_json', json)
+ for k, v in json.items():
+ if k == 'user':
+ try:
+ user = api.parser.model_factory.user.parse(api, v)
+ except AttributeError:
+ user = User.parse(api, v)
+ setattr(status, 'author', user)
+ setattr(status, 'user', user) # DEPRECIATED
+ elif k == 'created_at':
+ setattr(status, k, parsedate_to_datetime(v))
+ elif k == 'source':
+ if '<' in v:
+ # At this point, v should be of the format:
+ # <a href="{source_url}" rel="nofollow">{source}</a>
+ setattr(status, k, v[v.find('>') + 1:v.rfind('<')])
+ start = v.find('"') + 1
+ end = v.find('"', start)
+ setattr(status, 'source_url', v[start:end])
+ else:
+ setattr(status, k, v)
+ setattr(status, 'source_url', None)
+ elif k == 'retweeted_status':
+ setattr(status, k, Status.parse(api, v))
+ elif k == 'quoted_status':
+ setattr(status, k, Status.parse(api, v))
+ elif k == 'place':
+ if v is not None:
+ setattr(status, k, Place.parse(api, v))
+ else:
+ setattr(status, k, None)
+ else:
+ setattr(status, k, v)
+ return status
- def origin(self):
- """
- Return longitude, latitude of southwest (bottom, left) corner of
- bounding box, as a tuple.
+ def destroy(self):
+ return self._api.destroy_status(self.id)
- This assumes that bounding box is always a rectangle, which
- appears to be the case at present.
- """
- return tuple(self.coordinates[0][0])
+ def retweet(self):
+ return self._api.retweet(self.id)
- def corner(self):
- """
- Return longitude, latitude of northeast (top, right) corner of
- bounding box, as a tuple.
+ def retweets(self):
+ return self._api.retweets(self.id)
- This assumes that bounding box is always a rectangle, which
- appears to be the case at present.
- """
- return tuple(self.coordinates[0][2])
+ def favorite(self):
+ return self._api.create_favorite(self.id)
-class Place(Model):
+class User(Model, HashableID):
@classmethod
def parse(cls, api, json):
- place = cls(api)
+ user = cls(api)
+ setattr(user, '_json', json)
for k, v in json.items():
- if k == 'bounding_box':
- # bounding_box value may be null (None.)
- # Example: "United States" (id=96683cc9126741d1)
- if v is not None:
- t = BoundingBox.parse(api, v)
+ if k == 'created_at':
+ setattr(user, k, parsedate_to_datetime(v))
+ elif k == 'status':
+ setattr(user, k, Status.parse(api, v))
+ elif k == 'following':
+ # twitter sets this to null if it is false
+ if v is True:
+ setattr(user, k, True)
else:
- t = v
- setattr(place, k, t)
- elif k == 'contained_within':
- # contained_within is a list of Places.
- setattr(place, k, Place.parse_list(api, v))
+ setattr(user, k, False)
else:
- setattr(place, k, v)
- return place
+ setattr(user, k, v)
+ return user
@classmethod
def parse_list(cls, api, json_list):
if isinstance(json_list, list):
item_list = json_list
else:
- item_list = json_list['result']['places']
+ item_list = json_list['users']
results = ResultSet()
for obj in item_list:
results.append(cls.parse(api, obj))
return results
+ def timeline(self, **kwargs):
+ return self._api.user_timeline(user_id=self.id, **kwargs)
+
+ def friends(self, **kwargs):
+ return self._api.friends(user_id=self.id, **kwargs)
-class Media(Model):
+ def followers(self, **kwargs):
+ return self._api.followers(user_id=self.id, **kwargs)
+
+ def follow(self):
+ self._api.create_friendship(user_id=self.id)
+ self.following = True
+
+ def unfollow(self):
+ self._api.destroy_friendship(user_id=self.id)
+ self.following = False
+
+ def lists_memberships(self, *args, **kwargs):
+ return self._api.lists_memberships(user_id=self.id, *args, **kwargs)
+
+ def lists_ownerships(self, *args, **kwargs):
+ return self._api.lists_ownerships(user_id=self.id, *args, **kwargs)
+
+ def lists_subscriptions(self, *args, **kwargs):
+ return self._api.lists_subscriptions(user_id=self.id, *args, **kwargs)
+
+ def lists(self, *args, **kwargs):
+ return self._api.lists_all(user_id=self.id, *args, **kwargs)
+
+ def followers_ids(self, *args, **kwargs):
+ return self._api.followers_ids(user_id=self.id, *args, **kwargs)
+
+
+class IDModel(Model):
@classmethod
def parse(cls, api, json):
- media = cls(api)
- for k, v in json.items():
- setattr(media, k, v)
- return media
+ if isinstance(json, list):
+ return json
+ else:
+ return json['ids']
+
+
+class JSONModel(Model):
+
+ @classmethod
+ def parse(cls, api, json):
+ return json
class ModelFactory:
to add your own extended models.
"""
- status = Status
- user = User
+ bounding_box = BoundingBox
direct_message = DirectMessage
friendship = Friendship
- saved_search = SavedSearch
- search_results = SearchResults
list = List
- relationship = Relationship
media = Media
+ place = Place
+ relationship = Relationship
+ saved_search = SavedSearch
+ search_results = SearchResults
+ status = Status
+ user = User
- json = JSONModel
ids = IDModel
- place = Place
- bounding_box = BoundingBox
+ json = JSONModel
projects / tweepy.git / commitdiff