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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
|
from functools import wraps
from flask import redirect, url_for
from oauthlib.common import to_unicode
from requests_oauthlib import OAuth1Session as BaseOAuth1Session
from requests_oauthlib import OAuth2Session as BaseOAuth2Session
from urlobject import URLObject
from werkzeug.utils import cached_property
class OAuth1Session(BaseOAuth1Session):
"""
A :class:`requests.Session` subclass that can do some special things:
* lazy-loads OAuth1 tokens from the storage via the blueprint
* handles OAuth1 authentication
(from :class:`requests_oauthlib.OAuth1Session` superclass)
* has a ``base_url`` property used for relative URL resolution
Note that this is a session between the consumer (your website) and the
provider (e.g. Google), and *not* a session between a user of your website
and your website.
"""
def __init__(self, blueprint=None, base_url=None, *args, **kwargs):
super().__init__(*args, **kwargs)
self.blueprint = blueprint
self.base_url = URLObject(base_url)
@cached_property
def token(self):
"""
Get and set the values in the OAuth token, structured as a dictionary.
"""
return self.blueprint.token
def load_token(self):
t = self.token
if t and "oauth_token" in t and "oauth_token_secret" in t:
# This really, really violates the Law of Demeter, but
# I don't see a better way to set these parameters. :(
self.auth.client.resource_owner_key = to_unicode(t["oauth_token"])
self.auth.client.resource_owner_secret = to_unicode(t["oauth_token_secret"])
return True
return False
@property
def authorized(self):
"""This is the property used when you have a statement in your code
that reads "if <provider>.authorized:", e.g. "if google.authorized:".
The way it works is kind of complicated: this function just tries
to load the token, and then the 'super()' statement basically just
tests if the token exists (see BaseOAuth1Session.authorized).
To load the token, it calls the load_token() function within this class,
which in turn checks the 'token' property of this class (another
function), which in turn checks the 'token' property of the blueprint
(see base.py), which calls 'storage.get()' to actually try to load
the token from the cache/db (see the 'get()' function in
storage/sqla.py).
"""
self.load_token()
return super().authorized
@property
def authorization_required(self):
"""
.. versionadded:: 1.3.0
This is a decorator for a view function. If the current user does not
have an OAuth token, then they will be redirected to the
:meth:`~flask_dance.consumer.oauth1.OAuth1ConsumerBlueprint.login`
view to obtain one.
"""
def wrapper(func):
@wraps(func)
def check_authorization(*args, **kwargs):
if not self.authorized:
endpoint = f"{self.blueprint.name}.login"
return redirect(url_for(endpoint))
return func(*args, **kwargs)
return check_authorization
return wrapper
def prepare_request(self, request):
if self.base_url:
request.url = self.base_url.relative(request.url)
return super().prepare_request(request)
def request(
self, method, url, data=None, headers=None, should_load_token=True, **kwargs
):
if should_load_token:
self.load_token()
return super().request(
method=method, url=url, data=data, headers=headers, **kwargs
)
class OAuth2Session(BaseOAuth2Session):
"""
A :class:`requests.Session` subclass that can do some special things:
* lazy-loads OAuth2 tokens from the storage via the blueprint
* handles OAuth2 authentication
(from :class:`requests_oauthlib.OAuth2Session` superclass)
* has a ``base_url`` property used for relative URL resolution
Note that this is a session between the consumer (your website) and the
provider (e.g. Google), and *not* a session between a user of your website
and your website.
"""
def __init__(self, blueprint=None, base_url=None, *args, **kwargs):
super().__init__(*args, **kwargs)
self.blueprint = blueprint
self.base_url = URLObject(base_url)
del self.token
@cached_property
def token(self):
"""
Get and set the values in the OAuth token, structured as a dictionary.
"""
return self.blueprint.token
def load_token(self):
self._client.token = self.token
if self.token:
self._client.populate_token_attributes(self.token)
return True
return False
@property
def access_token(self):
"""
Returns the ``access_token`` from the OAuth token.
"""
return self.token and self.token.get("access_token")
@property
def authorized(self):
"""This is the property used when you have a statement in your code
that reads "if <provider>.authorized:", e.g. "if google.authorized:".
The way it works is kind of complicated: this function just tries
to load the token, and then the 'super()' statement basically just
tests if the token exists (see BaseOAuth1Session.authorized).
To load the token, it calls the load_token() function within this class,
which in turn checks the 'token' property of this class (another
function), which in turn checks the 'token' property of the blueprint
(see base.py), which calls 'storage.get()' to actually try to load
the token from the cache/db (see the 'get()' function in
storage/sqla.py).
"""
self.load_token()
return super().authorized
@property
def authorization_required(self):
"""
.. versionadded:: 1.3.0
This is a decorator for a view function. If the current user does not
have an OAuth token, then they will be redirected to the
:meth:`~flask_dance.consumer.oauth2.OAuth2ConsumerBlueprint.login`
view to obtain one.
"""
def wrapper(func):
@wraps(func)
def check_authorization(*args, **kwargs):
if not self.authorized:
endpoint = f"{self.blueprint.name}.login"
return redirect(url_for(endpoint))
return func(*args, **kwargs)
return check_authorization
return wrapper
def request(self, method, url, data=None, headers=None, **kwargs):
if self.base_url:
url = self.base_url.relative(url)
self.load_token()
return super().request(
method=method,
url=url,
data=data,
headers=headers,
client_id=self.blueprint.client_id,
client_secret=self.blueprint.client_secret,
**kwargs,
)
|
