Repository navigation
Expand file tree
/
Copy pathnotifications.py
More file actions
155 lines (123 loc) · 5.33 KB
/
Copy pathnotifications.py
File metadata and controls
155 lines (123 loc) · 5.33 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
"""
github3.notifications
=====================
This module contains the classes relating to notifications.
See also: http://developer.github.com/v3/activity/notifications/
"""
from json import dumps
from github3.models import GitHubCore
class Thread(GitHubCore):
"""The :class:`Thread <Thread>` object wraps notification threads. This
contains information about the repository generating the notification, the
subject, and the reason.
Two thread instances can be checked like so::
t1 == t2
t1 != t2
And is equivalent to::
t1.id == t2.id
t1.id != t2.id
See also:
http://developer.github.com/v3/activity/notifications/#view-a-single-thread
"""
def __init__(self, notif, session=None):
super(Thread, self).__init__(notif, session)
self._api = notif.get('url')
#: Comment responsible for the notification
self.comment = notif.get('comment', {})
#: Thread information
self.thread = notif.get('thread', {})
from github3.repos import Repository
#: Repository the comment was made on
self.repository = Repository(notif.get('repository', {}), self)
#: When the thread was last updated
self.updated_at = self._strptime(notif.get('updated_at'))
#: Id of the thread
self.id = notif.get('id')
#: Dictionary of urls for the thread
self.urls = notif.get('urls')
#: datetime object representing the last time the user read the thread
self.last_read_at = notif.get('last_read_at')
if self.last_read_at:
self.last_read_at = self._strptime(self.last_read_at)
#: The reason you're receiving the notification
self.reason = notif.get('reason')
#: Subject of the Notification, e.g., which issue/pull/diff is this in
#: relation to. This is a dictionary
self.subject = notif.get('subject')
self.unread = notif.get('unread')
def __repr__(self):
return '<Thread [{0}]>'.format(self.subject.get('title'))
def __eq__(self, other):
return self.id == other.id
def __ne__(self, other):
return self.id != other.id
def delete_subscription(self):
"""Delete subscription for this thread.
:returns: bool
"""
url = self._build_url('subscription', base_url=self._api)
return self._boolean(self._delete(url), 204, 404)
def is_unread(self):
"""Tells you if the thread is unread or not."""
return self.unread
def mark(self):
"""Mark the thread as read.
:returns: bool
"""
return self._boolean(self._patch(self._api), 205, 404)
def set_subscription(self, subscribed, ignored):
"""Set the user's subscription for this thread
:param bool subscribed: (required), determines if notifications should
be received from this thread.
:param bool ignored: (required), determines if notifications should be
ignored from this thread.
:returns: :class;`Subscription <Subscription>`
"""
url = self._build_url('subscription', base_url=self._api)
sub = {'subscribed': subscribed, 'ignored': ignored}
json = self._json(self._put(url, data=dumps(sub)), 200)
return Subscription(json, self) if json else None
def subscription(self):
"""Checks the status of the user's subscription to this thread.
:returns: :class:`Subscription <Subscription>`
"""
url = self._build_url('subscription', base_url=self._api)
json = self._json(self._get(url), 200)
return Subscription(json, self) if json else None
class Subscription(GitHubCore):
"""The :class:`Subscription <Subscription>` object wraps thread and
repository subscription information.
See also:
developer.github.com/v3/activity/notifications/#get-a-thread-subscription
"""
def __init__(self, sub, session=None):
super(Subscription, self).__init__(sub, session)
self._api = sub.get('url')
#: reason user is subscribed to this thread/repository
self.reason = sub.get('reason')
#: datetime representation of when the subscription was created
self.created_at = self._strptime(sub.get('created_at'))
#: API url of the thread if it exists
self.thread_url = sub.get('thread_url')
#: API url of the repository if it exists
self.repository_url = sub.get('repository_url')
self.ignored = sub.get('ignored', False)
self.subscribed = sub.get('subscribed', False)
def __repr__(self):
return '<Subscription [{0}]>'.format(self.subscribed)
def delete(self):
return self._boolean(self._delete(self._api), 204, 404)
def is_ignored(self):
return self.ignored
def is_subscribed(self):
return self.subscribed
def set(self, subscribed, ignored):
"""Set the user's subscription for this subscription
:param bool subscribed: (required), determines if notifications should
be received from this thread.
:param bool ignored: (required), determines if notifications should be
ignored from this thread.
"""
sub = {'subscribed': subscribed, 'ignored': ignored}
json = self._json(self._put(self._api, data=dumps(sub)), 200)
self.__init__(json, self._session)