61
61
"developer_message": string?
62
62
}
63
63
"""
64
- from django .contrib .auth .models import AbstractUser
64
+ from django .contrib .auth .models import User # pylint: disable=imported-auth-user
65
65
from opaque_keys import InvalidKeyError
66
66
from opaque_keys .edx .keys import UsageKey
67
67
from rest_framework .exceptions import NotFound , ValidationError
82
82
from xmodule .modulestore .exceptions import ItemNotFoundError
83
83
84
84
85
+ class _AuthenticatedRequest (Request ):
86
+ """
87
+ Alias for the `Request` class which tells mypy to assume that `.user` is not an AnonymousUser.
88
+
89
+ Using this does NOT ensure the request is actually authenticated--
90
+ you will some other way to ensure that, such as `@view_auth_classes(is_authenticated=True)`.
91
+ """
92
+ user : User
93
+
94
+
85
95
# TODO: Potential future view.
86
96
# @view_auth_classes(is_authenticated=True)
87
97
# class DownstreamListView(DeveloperErrorViewMixin, APIView):
88
98
# """
89
99
# List all blocks which are linked to upstream content, with optional filtering.
90
100
# """
91
- # def get(self, request: Request ) -> Response:
101
+ # def get(self, request: _AuthenticatedRequest ) -> Response:
92
102
# """
93
103
# Handle the request.
94
104
# """
@@ -102,22 +112,20 @@ class DownstreamView(DeveloperErrorViewMixin, APIView):
102
112
"""
103
113
Inspect or manage an XBlock's link to upstream content.
104
114
"""
105
- def get (self , request : Request , usage_key_string : str ) -> Response :
115
+ def get (self , request : _AuthenticatedRequest , usage_key_string : str ) -> Response :
106
116
"""
107
117
Inspect an XBlock's link to upstream content.
108
118
"""
109
- assert isinstance (request .user , AbstractUser )
110
119
downstream = _load_accessible_block (request .user , usage_key_string , require_write_access = False )
111
120
_ensure_upstream_ref (downstream )
112
121
if link := UpstreamLink .try_get_for_block (downstream ):
113
122
return Response (link .to_json ())
114
123
raise ValidationError (detail = f"Block '{ usage_key_string } ' is not linked to an upstream" )
115
124
116
- def put (self , request : Request , usage_key_string : str ) -> Response :
125
+ def put (self , request : _AuthenticatedRequest , usage_key_string : str ) -> Response :
117
126
"""
118
127
Edit an XBlock's link to upstream content.
119
128
"""
120
- assert isinstance (request .user , AbstractUser )
121
129
downstream = _load_accessible_block (request .user , usage_key_string , require_write_access = True )
122
130
new_upstream_ref = request .data .get ("upstream_ref" )
123
131
if not isinstance (new_upstream_ref , str ):
@@ -140,15 +148,13 @@ def put(self, request: Request, usage_key_string: str) -> Response:
140
148
except BadUpstream as exc :
141
149
raise ValidationError ({"upstream_ref" : str (exc )}) from exc
142
150
modulestore ().update_item (downstream , request .user .id )
143
- link = UpstreamLink .get_for_block (downstream )
144
- assert link
151
+ link : UpstreamLink = UpstreamLink .get_for_block (downstream ) # type: ignore[assignment]
145
152
return Response (link .to_json ())
146
153
147
- # def delete(self, request: Request , usage_key_string: str) -> Response:
154
+ # def delete(self, request: _AuthenticatedRequest , usage_key_string: str) -> Response:
148
155
# """
149
156
# Sever an XBlock's link to upstream content.
150
157
# """
151
- # assert isinstance(request.user, AbstractUser)
152
158
# downstream = _load_accessible_block(request.user, usage_key_string, require_write_access=True)
153
159
# _ensure_upstream_ref(downstream)
154
160
# ....
@@ -160,11 +166,10 @@ class SyncFromUpstreamView(DeveloperErrorViewMixin, APIView):
160
166
Accept or decline an opportunity to sync a downstream block from its upstream content.
161
167
"""
162
168
163
- def post (self , request : Request , usage_key_string : str ) -> Response :
169
+ def post (self , request : _AuthenticatedRequest , usage_key_string : str ) -> Response :
164
170
"""
165
171
Pull latest updates to the block at {usage_key_string} from its linked upstream content.
166
172
"""
167
- assert isinstance (request .user , AbstractUser )
168
173
downstream = _load_accessible_block (request .user , usage_key_string , require_write_access = True )
169
174
_ensure_upstream_ref (downstream )
170
175
if not downstream .upstream :
@@ -175,15 +180,13 @@ def post(self, request: Request, usage_key_string: str) -> Response:
175
180
except (BadUpstream , BadDownstream ) as exc :
176
181
raise ValidationError (detail = str (exc )) from exc
177
182
modulestore ().update_item (downstream , request .user .id )
178
- upstream_link = UpstreamLink .get_for_block (downstream )
179
- assert upstream_link
180
- return Response (upstream_link .to_json (), status = 200 )
183
+ link : UpstreamLink = UpstreamLink .get_for_block (downstream ) # type: ignore[assignment]
184
+ return Response (link .to_json (), status = 200 )
181
185
182
- def delete (self , request : Request , usage_key_string : str ) -> Response :
186
+ def delete (self , request : _AuthenticatedRequest , usage_key_string : str ) -> Response :
183
187
"""
184
188
Decline the latest updates to the block at {usage_key_string}.
185
189
"""
186
- assert isinstance (request .user , AbstractUser )
187
190
downstream = _load_accessible_block (request .user , usage_key_string , require_write_access = True )
188
191
_ensure_upstream_ref (downstream )
189
192
try :
@@ -194,7 +197,7 @@ def delete(self, request: Request, usage_key_string: str) -> Response:
194
197
return Response (status = 204 )
195
198
196
199
197
- def _load_accessible_block (user : AbstractUser , usage_key_string : str , * , require_write_access : bool ) -> XBlock :
200
+ def _load_accessible_block (user : User , usage_key_string : str , * , require_write_access : bool ) -> XBlock :
198
201
"""
199
202
Given a logged in-user and a serialized usage key of an upstream-linked XBlock, load it from the ModuleStore,
200
203
raising a DRF-friendly exception if anything goes wrong.
0 commit comments