Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def harvest_resources(request, server_id):
try:
server = Server.objects.get(id=server_id)
except ObjectDoesNotExist:
messages.error(request, "Server Not Found")
user = request.user
harvest_task.delay(server.id, user_id=user.id or None)
index_url = "%s_%s_modeladmin_index" % (Server._meta.app_label, Server._meta.model_name)
messages.success(request, "Server Resources Will Be Collected")
return redirect(index_url)
def delete_invalid_resources(self, request, pk=None):
try:
server = Server.objects.get(pk=pk)
except ObjectDoesNotExist as e:
return Response({"details": str(e)},
status=status.HTTP_500_INTERNAL_SERVER_ERROR)
delete_invalid_resources.delay(server.id)
return Response({"message":
_("Invalid Server Resources Will Be Deleted")},
status=status.HTTP_202_ACCEPTED)
def __init__(self, base_url, server_id, user_id=None):
self.url = base_url
self.server = Server.objects.get(id=server_id)
self.user_id = user_id
self.user = None
if user_id:
USER = get_user_model()
self.user = USER.objects.get(id=user_id)
def serve(self, request, pk, *args, **kwargs):
server = Server.objects.get(pk=pk)
user = request.user
user_id = None if not user else user.id
conn = server.get_user_connections(user_id).first()
allowed = False
params = {}
if conn:
conn = conn.credentials
if request.method in permissions.SAFE_METHODS:
allowed = user.has_perm('use_for_read', conn)
else:
allowed = user.has_perm('use_for_write', conn)
if isinstance(conn, TokenAuthConnection):
params.update({'access_token': conn.token})
if allowed:
logger.info("ALLOWED To USE SESSION")
session = server.handler(user_id=user_id).session
def update_server_resources(self, request, pk=None):
try:
server = Server.objects.get(pk=pk)
except ObjectDoesNotExist as e:
return Response({"details": str(e)},
status=status.HTTP_500_INTERNAL_SERVER_ERROR)
update_server_resources.delay(server.id, user_id=request.user.id or None)
return Response({"message":
_("Server Resources Will Be Updated")},
status=status.HTTP_202_ACCEPTED)