Untitled
unknown
python
a year ago
2.3 kB
5
Indexable
class AssetsV3View(GenericAPIView): def get(self, request, path, *args, **kwargs): import boto3 maybe_schema = path.split('/', 1)[0] code = request.GET.get('code') state = request.GET.get('state') error = request.GET.get('error') keycloak = Keycloak() redirect_uri = request.build_absolute_uri().split('?')[0] login_url = keycloak.generate_link(state if not error else None, redirect_uri) if not code: return HttpResponseRedirect(login_url) else: data = keycloak.check_token(code, redirect_uri) token = data.get('access_token', None) refresh_token = data.get('refresh_token', None) if not token: return HttpResponseRedirect(login_url) decode = keycloak.parse_token(token) if not decode: return HttpResponseRedirect(login_url) email = decode.get('email', None) if not email: return HttpResponseRedirect(login_url) user = User.objects.get(email=email) is_access = False # Check schema already schema = Account.objects.filter(schema_name=maybe_schema).first() if not schema: return HttpResponseRedirect(login_url) for account in user.accounts.all(): if account.schema_name != 'public' and account.schema_name == maybe_schema: is_access = True if not is_access: if keycloak.logout(token, refresh_token): return HttpResponseRedirect(login_url + '&errorCode=403') s3 = boto3.resource( service_name='s3', aws_access_key_id=settings.AWS_ACCESS_KEY_ID, aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY, ) url = s3.meta.client.generate_presigned_url( ClientMethod="get_object", Params={ "Bucket": settings.AWS_STORAGE_BUCKET_NAME_FILES_MS, "Key": path, }, ) r = requests.get(url=url, stream=True) r.raise_for_status() content_type = r.headers['content-type'] response = HttpResponse(r.raw, content_type=content_type) # response['Content-Disposition'] = 'inline; filename=music.mp3' return response
Editor is loading...
Leave a Comment