def _post(self, request):
raise NotImplementedError("You have to define this method in children View class")
python类View()的实例源码
def get_view_name(view_cls, suffix=None):
"""
Given a view class, return a textual name to represent the view.
This name is used in the browsable API, and in OPTIONS responses.
This function is the default for the `VIEW_NAME_FUNCTION` setting.
"""
name = view_cls.__name__
name = formatting.remove_trailing_string(name, 'View')
name = formatting.remove_trailing_string(name, 'ViewSet')
name = formatting.camelcase_to_spaces(name)
if suffix:
name += ' ' + suffix
return name
def dispatch(self, request, *args, **kwargs):
return generic.View.dispatch(self, request, *args, **kwargs)
def dispatch(self, request, *args, **kwargs):
return generic.View.dispatch(self, request, *args, **kwargs)
def _handle_api_callback(self, request):
"""handle redirect from instagram when user denied access"""
global HOST
logger.debug("Dropbox handling get-request: "
"identified as API callback")
try:
redirect_dropbox = HOST + "/dropbox/authenticate"
access_token, userid, url_state = DropboxOAuth2Flow(APP_KEY,
APP_SECRET,
redirect_dropbox,
request.session,
"dropbox-auth-csrf-token"
).finish(request.GET)
return self._save_user(access_token, userid, url_state, request)
except DropboxOAuth2Flow.BadRequestException as e:
logger.error("[Dropbox - View - auth-finish] BadRequestException")
return self.error(request)
except DropboxOAuth2Flow.BadStateException as e:
return redirect(reverse("dropbox:connect"))
except DropboxOAuth2Flow.CsrfException as e:
logger.error("[Dropbox - View - auth-finish] ... CsrfException")
return self.error(request)
except DropboxOAuth2Flow.NotApprovedException as e:
logger.error("[Dropbox - View - auth-finish] ... 403")
return self.error(request)
except DropboxOAuth2Flow.ProviderException as e:
logger.error("[Dropbox - View - auth-finish] ... NotApprovedException")
return self.error(request)
def _save_user(self, access_token, userid, url_state, request):
dbx = dropbox.Dropbox(access_token)
account = dbx.users_get_current_account()
disk = dbx.users_get_space_usage()
if disk.allocation.is_individual() is False:
return self.error(request)
init_dropbox_folder = dbx.files_list_folder(path='',
recursive=True,
include_media_info=True,
include_deleted=True)
# to convert the bytes to megabytes
MEGA_BYTE = 1000000
used = (disk.used / MEGA_BYTE)
allocated = (disk.allocation.get_individual().allocated / MEGA_BYTE)
dropbox_account = DropboxAccount(user = request.user,
access_token = access_token,
cursor = init_dropbox_folder.cursor)
dropbox_account.save()
dropbox_user = DropboxUser(dropbox_account = dropbox_account,
dropbox_userid = userid,
display_name = account.name.display_name,
email = account.email,
profile_photo_url = account.profile_photo_url,
disk_used = used,
disk_allocated = allocated)
dropbox_user.save()
logger.debug('[Dropbox - View - auth-finish] \
DropboxUser added to user: {}'.format(request.user.username))
return self.success(request)
def get(self, request):
logger.debug("DropBox handling GET webhook")
if request.method == "GET":
logger.debug('[Dropbox - View - dropbox_call_webhook] \
DropBox - send Hook-Test')
return HttpResponse(request.GET.get('challenge'))
def post(self, request):
"""Receive a list of changed user IDs from Dropbox and process each."""
# Make sure this is a valid request from Dropbox
logger.debug('[Dropbox - View - Webhook] received a webhook')
signature = request.environ.get('HTTP_X_DROPBOX_SIGNATURE')
try:
request_body_size = int(request.environ.get('CONTENT_LENGTH', 0))
except (ValueError):
request_body_size = 0
pass
request_body = request.environ['wsgi.input'].read(request_body_size)
request_hashed = hmac.new( APP_SECRET.encode("utf-8"),
request_body,
sha256).hexdigest()
if not hmac.compare_digest(signature, request_hashed):
error_url = '{}?status=error'.format(self.get_redirect(request))
return redirect(error_url)
try:
data = json.loads(request_body.decode('utf-8'))
for user in data['delta']['users']:
logger.debug('[Dropbox - View - webhook] \
firedTrigger for user with id:{}'.format(user))
tasks.fireTrigger(user)
except json.decoder.JSONDecodeError:
pass
return HttpResponse()
def dispatch(self, request, *args, **kwargs):
return generic.View.dispatch(self, request, *args, **kwargs)
def get(self, request):
"""
Handling GET method
:param request: Request to View
:return: rendered registration page
"""
return render(request, "register.html")
def get(self, request):
"""
:param request: Request to View
:return: user profile page
"""
user = request.user
data = {}
data['AdviserUser'] = model_to_dict(AdviserUser.objects.get(user_id=user.id))
data['User'] = model_to_dict(User.objects.get(id=user.id))
return HttpResponse(json.dumps(data, default=self.datetime_handler))
def test_fallback_class_based_view(self):
class OtherTestView(View):
def get(self, request, *args, **kwargs):
return HttpResponse('fallback cbv')
view = TestView.as_view(
flag_name=self.flag_name,
condition=True,
fallback=OtherTestView.as_view()
)
response = view(self.request())
self.assertContains(response, 'fallback cbv')
def project_detail_view():
class FakeProjectDetailView(mixins.PhaseDispatchMixin, View):
model = models.Project
def get(self, request, *args, **kwargs):
return HttpResponse('project_detail')
return FakeProjectDetailView.as_view()
def dispatch(self, request, *args, **kwargs):
return generic.View.dispatch(self, request, *args, **kwargs)
# Post function to handle Facebook messages
def dispatch(self, request, *args, **kwargs):
return generic.View.dispatch(self, request, *args, **kwargs)
def dispatch(self, request, *args, **kwargs):
return generic.View.dispatch(self, request, *args, **kwargs)
# Post function to handle Facebook messages
def get_permission_denied_message(self):
logger.info("Access Denied - Cannot View")
return "You do not have access to view this Lawn! Please go back, and create your own."
def index(request):
logger.debug("Index View (homepage)")
return render(request, "planner/index.html", {})
def clear_session_view(request):
"""
View that clears the session and restarts the user flow.
@param request: the HTTP request
"""
request.session.flush()
return redirect(build_view_url(request, PaymentMethodChoiceView.url_name))
def dispatch(self, request, *args, **kwargs):
return generic.View.dispatch(self, request, *args, **kwargs)
# Post function to handle Facebook messages
