-
Notifications
You must be signed in to change notification settings - Fork 34
/
device.py
140 lines (98 loc) · 4 KB
/
device.py
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
from trakt import Trakt
from threading import Condition
import logging
import os
logging.basicConfig(level=logging.DEBUG)
class Application(object):
def __init__(self):
self.is_authenticating = Condition()
self.authorization = None
# Bind trakt events
Trakt.on('oauth.token_refreshed', self.on_token_refreshed)
def authenticate(self):
if not self.is_authenticating.acquire(blocking=False):
print('Authentication has already been started')
return False
# Request new device code
code = Trakt['oauth/device'].code()
print('Enter the code "%s" at %s to authenticate your account' % (
code.get('user_code'),
code.get('verification_url')
))
# Construct device authentication poller
poller = Trakt['oauth/device'].poll(**code)\
.on('aborted', self.on_aborted)\
.on('authenticated', self.on_authenticated)\
.on('expired', self.on_expired)\
.on('poll', self.on_poll)
# Start polling for authentication token
poller.start(daemon=False)
# Wait for authentication to complete
return self.is_authenticating.wait()
def run(self):
self.authenticate()
if not self.authorization:
print('ERROR: Authentication required')
exit(1)
# Simulate expired token
self.authorization['expires_in'] = 0
# Test authenticated calls
with Trakt.configuration.oauth.from_response(self.authorization):
# Expired token, requests will return `None`
print(Trakt['sync/collection'].movies())
with Trakt.configuration.oauth.from_response(self.authorization, refresh=True):
# Expired token will be refreshed automatically (as `refresh=True`)
print(Trakt['sync/collection'].movies())
with Trakt.configuration.oauth.from_response(self.authorization):
# Current token is still valid
print(Trakt['sync/collection'].movies())
def on_aborted(self):
"""Device authentication aborted.
Triggered when device authentication was aborted (either with `DeviceOAuthPoller.stop()`
or via the "poll" event)
"""
print('Authentication aborted')
# Authentication aborted
self.is_authenticating.acquire()
self.is_authenticating.notify_all()
self.is_authenticating.release()
def on_authenticated(self, authorization):
"""Device authenticated.
:param authorization: Authentication token details
:type authorization: dict
"""
# Acquire condition
self.is_authenticating.acquire()
# Store authorization for future calls
self.authorization = authorization
print('Authentication successful - authorization: %r' % self.authorization)
# Authentication complete
self.is_authenticating.notify_all()
self.is_authenticating.release()
def on_expired(self):
"""Device authentication expired."""
print('Authentication expired')
# Authentication expired
self.is_authenticating.acquire()
self.is_authenticating.notify_all()
self.is_authenticating.release()
def on_poll(self, callback):
"""Device authentication poll.
:param callback: Call with `True` to continue polling, or `False` to abort polling
:type callback: func
"""
# Continue polling
callback(True)
def on_token_refreshed(self, authorization):
# OAuth token refreshed, store authorization for future calls
self.authorization = authorization
print('Token refreshed - authorization: %r' % self.authorization)
if __name__ == '__main__':
# Configure
Trakt.base_url = 'http://api.staging.trakt.tv'
Trakt.configuration.defaults.client(
id=os.environ.get('CLIENT_ID'),
secret=os.environ.get('CLIENT_SECRET')
)
app = Application()
app.run()