from django.utils.decorators import decorator_from_middleware_with_args from django.utils.decorators import method_decorator from rest_framework import status from rest_framework.views import APIView from authenticate.middleware import AuthorizatorMiddleware from collaborators import defined_roles as dr from collaborators import serializers as collaborators_serializers from collaborators.middleware import CollaborationAccessMiddleware from collaborators.models import Access from institute import conf from institute import serializers as institute_serializers from institute.models import Institute, APIKey, InstituteJWT, InstituteStudent from institute.serializers import GetInstituteStudentSerializer, SetInstituteStudentInputValidator, ListInstituteStudentInputValidator, \ GetInstituteStudentInputValidator from payment.models import InstituteBalanceHistory, InitialBalance from utils import utils from utils.limitations import check_user_limitations from utils.myresponse import MyResponse from wikiazma.middleware import SerializerValidationMiddleware serializer_validation_middleware = decorator_from_middleware_with_args( SerializerValidationMiddleware) user_token_or_server_token_or_server_API_key_validation_and_authorization = decorator_from_middleware_with_args( AuthorizatorMiddleware) check_access_permission_middleware = decorator_from_middleware_with_args( CollaborationAccessMiddleware) @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_user_token=True), name='post') @method_decorator(serializer_validation_middleware(institute_serializers.CreateInstituteInputSerializer), name='post') class CreateInstitute(APIView): """ This API class is responsible for handling incoming requests for creating a institute. it takes name & description from user and checks if a institute with same unique name doesn't exists. if it passes the check, then it creates a institute with incoming name and description """ def post(self, request): # ########################## Block 1 ################################ # receiving user inputs from serializer name = request.middleware_serializer_data.get('name') description = request.middleware_serializer_data.get('description') logo = request.middleware_serializer_data.get('logo') poster = request.middleware_serializer_data.get('poster') info = request.middleware_serializer_data.get('info') # ########################## Block 2 ################################ # first check user limitations and then creates a institute record with incoming parameters # also creates an access record for authenticated user to this institute with __ROLE_OWNER__ # then serializes the institute record to json and passed it to MyResponse if not check_user_limitations(Institute, request.middleware_user): return MyResponse(request, {'status': 'error', 'message': 'Payment Required'}, status=status.HTTP_402_PAYMENT_REQUIRED) institute = Institute() institute.name = name institute.description = description institute.user = request.middleware_user institute.info = info if logo: institute.logo = utils.compress_image( logo, max_side_size=conf.maximum_side_size_compress_for_institute_logo) institute.logo_thumbnail = utils.compress_image(logo, max_side_size=conf.maximum_side_size_compress_for_institute_logo_thumbnail) if poster: institute.poster = utils.compress_image( poster, max_side_size=conf.maximum_side_size_compress_for_institute_poster) institute.poster_thumbnail = utils.compress_image(poster, max_side_size=conf.maximum_side_size_compress_for_institute_poster_thumbnail) institute.save() InstituteBalanceHistory.objects.create(amount=0, reason=InitialBalance, remaining_balance=0, reason_meta_data={}, institute=institute) Access.objects.create( user=request.middleware_user, institute=institute, roles=[dr.__ROLE_OWNER__], institute_name=name, institute_created_at=institute.created_at ) institute_data = institute_serializers.InstituteSerializerForExamAndInstituteAndQuestionBank( institute).data return MyResponse(request, {'status': 'ok', 'message': 'Successful', 'institute': institute_data}, status=status.HTTP_200_OK) @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_user_token=True), name='post') @method_decorator(serializer_validation_middleware(institute_serializers.EditInstituteInputSerializer), name='post') @method_decorator(check_access_permission_middleware([dr.__PERMISSION_EDIT_INSTITUTE__], Institute, 'id', 'institute_id'), name='post') class EditInstitute(APIView): """ This API class is responsible for handling incoming requests for editing a institute. it takes name, description from user. first it checks if a institute with same unique name doesn't exists and then checks if a institute with incoming institute_id exists. if it passes both checks, then it edits desired institute with incoming name and description and if any of above checks fails, user faces with an error. """ def post(self, request): # ########################## Block 1 ################################ # receiving user inputs from serializer name = request.middleware_serializer_data.get('name') description = request.middleware_serializer_data.get('description') info = request.middleware_serializer_data.get('info') logo = request.middleware_serializer_data.get('logo') poster = request.middleware_serializer_data.get('poster') delete_logo = request.middleware_serializer_data.get('delete_logo') delete_poster = request.middleware_serializer_data.get('delete_poster') # ########################## Block 2 ################################ # replace the existing institute name & description with incoming ones # and then serializes the institute record to json and passed it to MyResponse institute = request.middleware_institute institute.name = name if description or description == '': institute.description = description if info or info == '': institute.info = info if delete_logo: institute.logo.delete() institute.logo_thumbnail.delete() if delete_poster: institute.poster.delete() institute.poster_thumbnail.delete() if logo: institute.logo = utils.compress_image( logo, max_side_size=conf.maximum_side_size_compress_for_institute_logo) institute.logo_thumbnail = utils.compress_image(logo, max_side_size=conf.maximum_side_size_compress_for_institute_logo_thumbnail) if poster: institute.poster = utils.compress_image( poster, max_side_size=conf.maximum_side_size_compress_for_institute_poster) institute.poster_thumbnail = utils.compress_image(poster, max_side_size=conf.maximum_side_size_compress_for_institute_poster_thumbnail) institute.save() institute_data = institute_serializers.InstituteSerializerForExamAndInstituteAndQuestionBank( institute).data return MyResponse(request, {'status': 'ok', 'message': 'Successful', 'institute': institute_data}, status=status.HTTP_200_OK) @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_user_token=True), name='post') @method_decorator(serializer_validation_middleware(institute_serializers.DeleteInstituteInputSerializer), name='post') @method_decorator(check_access_permission_middleware([dr.__PERMISSION_DELETE_INSTITUTE__], Institute, 'id', 'institute_id'), name='post') class DeleteInstitute(APIView): """ This API class is responsible for handling incoming requests for deleting a institute. """ def post(self, request): # ########################## Block 1 ################################ # gets the institute from collaborators middleware and deletes it. request.middleware_institute.delete() return MyResponse(request, {'status': 'ok', 'message': 'Successful'}, status=status.HTTP_200_OK) @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_user_token=True), name='post') @method_decorator(serializer_validation_middleware(institute_serializers.GetInstituteInputSerializer), name='post') @method_decorator(check_access_permission_middleware([dr.__PERMISSION_READ_INSTITUTE__], Institute, 'id', 'institute_id'), name='post') class GetInstitute(APIView): def post(self, request): # ########################## Block 1 ################################ # try to get an access record which its user is authenticated user # then serializes it to json & passed it to MyResponse institute_data = institute_serializers.InstituteSerializerForExamAndInstituteAndQuestionBank( request.middleware_institute).data return MyResponse(request, {'status': 'ok', 'message': 'Successful', 'institute': institute_data}, status=status.HTTP_200_OK) # TODO :: skip # @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_user_token=True), # name='post') # @method_decorator(serializer_validation_middleware(institute_serializers.InstituteAccessGetSerializer), name='post') # @method_decorator(check_access_permission_middleware([], Institute, 'id', 'institute_id'), name='post') # class InstituteAccessGet(APIView): # def post(self, request): # # ########################## Block 1 ################################ # # try to get an access record which its user is authenticated user # # then serializes it to json & passed it to MyResponse # access_data = collaborators_serializers.AccessInstituteSerializer( # request.middleware_access).data # return MyResponse(request, {'status': 'ok', 'message': 'Successful', 'access': access_data}, status=status.HTTP_200_OK) # TODO :: skip # @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_user_token=True), # name='post') # @method_decorator(serializer_validation_middleware(institute_serializers.MyAccessSerializer), name='post') # class MyAccess(APIView): # def post(self, request): # institute_id = request.middleware_serializer_data.get('institute_id') # try: # my_access = Access.objects.get(user=request.middleware_user, institute_id=institute_id) # except: # return MyResponse(request, {'status': 'error', 'message': 'you do not have access in this project'}, # status=status.HTTP_403_FORBIDDEN) # my_roles = my_access.roles # data = dr.roles_permission_finder(my_roles) # return MyResponse(request, {'status': 'ok', 'message': 'Successful', 'data': {'roles': data}}, status=status.HTTP_200_OK) @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_user_token=True), name='post') @method_decorator(serializer_validation_middleware(institute_serializers.AllInstitutesInputSerializer), name='post') class AllInstitutes(APIView): """ This API class is responsible for handling incoming requests for getting all institute which the authenticated user has access in them. """ def post(self, request): # ########################## Block 1 ################################ # receiving user inputs from serializer & make sort_by_attr with sort_by search_words = request.middleware_serializer_data.get('search_words') skip = request.middleware_serializer_data.get('skip') take = request.middleware_serializer_data.get('take') sort_by = request.middleware_serializer_data.get('sort_by') sort_by_attr = utils.access_sort_by_converter(sort_by) # ########################## Block 2 ################################ # gets the access list based on two situations: search_words comes or not! access_list = Access.objects.filter(user=request.middleware_user) if search_words: access_list = access_list.filter( institute__name__icontains=search_words) # ########################## Block 3 ################################ # gets the count of access_list and makes an ordered_access_list based on sort_by_attr and skip & take # then serializes it to json and passed it to MyResponse total_count = access_list.count() ordered_access_list = access_list.order_by(sort_by_attr)[ skip:skip + take] access_data = collaborators_serializers.AccessibleInstituteSerializer( ordered_access_list, many=True).data return MyResponse(request, {'status': 'ok', 'message': 'Successful', 'data': { 'total_count': total_count, 'institutes': access_data}}, status=status.HTTP_200_OK) # TODO :: skip # @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_user_token=True), # name='post') # @method_decorator(serializer_validation_middleware(institute_serializers.OwnedInstitutesSerializer), name='post') # class OwnedInstitutes(APIView): # """ # This API class is responsible for handling incoming requests for getting all institutes which the authenticated user is owner of them. # """ # def post(self, request): # # ########################## Block 1 ################################ # # receiving user inputs from serializer & make sort_by_attr with sort_by # search_words = request.middleware_serializer_data.get('search_words') # skip = request.middleware_serializer_data.get('skip') # take = request.middleware_serializer_data.get('take') # sort_by = request.middleware_serializer_data.get('sort_by') # sort_by_attr = utils.institute_sort_by_converter(sort_by) # # ########################## Block 2 ################################ # # gets the institutes list based on two situations: search_words comes or not! # institutes_list = Institute.objects.filter( # user=request.middleware_user) # if search_words: # institutes_list = institutes_list.filter( # name__icontains=search_words) # # ########################## Block 3 ################################ # # gets the count of institutes_list and makes an ordered_institutes_list based on sort_by_attr and skip & take # # then serializes it to json and passed it to MyResponse # total_count = institutes_list.count() # ordered_institutes_list = institutes_list.order_by(sort_by_attr)[ # skip:skip + take] # institutes_data = institute_serializers.InstituteSerializer( # ordered_institutes_list, many=True).data # for item in institutes_data: # # TODO what is this? # del item['redirect_url'] # return MyResponse(request, # {'status': 'ok', 'message': 'Successful', 'data': { # 'total_count': total_count, 'institutes': institutes_data}}, # status=status.HTTP_200_OK) # TODO :: skip # @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_user_token=True), # name='post') # @method_decorator(serializer_validation_middleware(institute_serializers.IntersectionInstitutesSerializer), name='post') # class IntersectionInstitutes(APIView): # """ # This API class is responsible for handling incoming requests for getting itersection accesses between authenticated user & another user. # """ # def post(self, request): # # ########################## Block 1 ################################ # # receiving user inputs from serializer # email = request.middleware_serializer_data.get('email').lower() # skip = request.middleware_serializer_data.get('skip') # take = request.middleware_serializer_data.get('take') # sort_by = request.middleware_serializer_data.get('sort_by') # # ########################## Block 2 ################################ # # checks if incoming email exists in our user database or not # try: # second_user = User.objects.get(email=email) # except: # return MyResponse(request, {'status': 'error', 'message': 'We cant find this user!'}, status=status.HTTP_403_FORBIDDEN) # # ########################## Block 3 ################################ # # getting all access records corresponding to both users # first_user_access_list = Access.objects.filter( # user=request.middleware_user) # second_user_access_list = Access.objects.filter(user=second_user) # # ########################## Block 4 ################################ # # first making union of above records and then find duplicates institute ids. # union_access_list = first_user_access_list | second_user_access_list # duplicates = union_access_list.values('institute').annotate( # institute_count=Count('institute')).exclude(institute_count=1) # # ########################## Block 5 ################################ # # first find access records with those institute ids and gets the count of them. # # and finally ordered them based on sort_by_attr and skip & take and serializes them to json and passed it to MyResponse # total_count = Access.objects.filter(user=request.middleware_user, institute__id__in=[ # item['institute'] for item in duplicates]).count() # sort_by_attr = utils.access_sort_by_converter(sort_by) # intersection_access_list = Access.objects.filter(user=request.middleware_user, institute__id__in=[ # item['institute'] for item in duplicates]).order_by(sort_by_attr)[skip:skip + take] # intersection_access_data = collaborators_serializers.AccessInstituteSerializer( # intersection_access_list, many=True).data # return MyResponse(request, {'status': 'ok', 'message': 'Successful', # 'data': {'total_count': total_count, 'institutes': intersection_access_data}}, # status=status.HTTP_200_OK) # TODO :: skip # @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_user_token=True), # name='post') # @method_decorator(serializer_validation_middleware(institute_serializers.TransferOwnerShipStateSerializer), name='post') # @method_decorator(check_access_permission_middleware([dr.__PERMISSION_TRANSFER_OWNERSHIP__], Institute, 'id', 'institute_id'), name='post') # class TransferOwnerShipState(APIView): # """ # This API class is responsible for handling incoming requests for getting state of the incomplete transfer ownership of authenticated user. # """ # def post(self, request): # # gets the TransferOwnershipHistory record with the incoming institute and based on new_owner_accept_at or old_owner_accept_at are null. # # then serializes it to json (if exists) and then passed it to MyResponse # transfer_ownership_list = TransferOwnershipHistory.objects.filter( # institute=request.middleware_institute).filter(Q(new_owner_accept_at=None) | Q(old_owner_accept_at=None)) # transfer_ownership_record_data = None # if len(transfer_ownership_list) != 0: # transfer_ownership_record = transfer_ownership_list[0] # transfer_ownership_record_data = institute_serializers.TransferOwnershipHistorySerializer( # transfer_ownership_record).data # return MyResponse(request, {'status': 'ok', 'message': 'Successful', 'data': transfer_ownership_record_data}, # status=status.HTTP_200_OK) # TODO :: skip # @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_user_token=True), # name='post') # @method_decorator(serializer_validation_middleware(institute_serializers.TransferOwnerShipRequestSerializer), name='post') # @method_decorator(check_access_permission_middleware([dr.__PERMISSION_TRANSFER_OWNERSHIP__], Institute, 'id', 'institute_id'), name='post') # class TransferOwnerShipRequest(APIView): # """ # This API class is responsible for handling incoming requests to make a request to transfer the ownership of a institute. # """ # def post(self, request): # # ########################## Block 1 ################################ # # receiving user inputs from serializer # new_owner_email = request.middleware_serializer_data.get( # 'new_owner_email').lower() # description = request.middleware_serializer_data.get('description') # # ########################## Block 2 ################################ # # checks if new_owner_email exists in our user database or not! # # and also checks if this email is not equal to token user email # try: # new_owner = User.objects.get(email=new_owner_email) # if new_owner == request.middleware_user: # return MyResponse(request, {'status': 'error', 'message': 'You cant enter your email.'}, status=status.HTTP_403_FORBIDDEN) # except: # return MyResponse(request, {'status': 'error', 'message': 'We cant find this user!'}, status=status.HTTP_403_FORBIDDEN) # # ########################## Block 3 ################################ # # checks if there is an active transfer ownership request for this institute or not! # if TransferOwnershipHistory.objects.filter(institute=request.middleware_institute).filter( # Q(new_owner_accept_at=None) | Q(old_owner_accept_at=None)): # return MyResponse(request, {'status': 'error', 'message': 'There is an active transfer ownership request for this institute!'}, # status=status.HTTP_403_FORBIDDEN) # # ########################## Block 4 ################################ # # checks if there are too many unfinished transfer requests for authenticated user or not! # if TransferOwnershipHistory.objects.filter(old_owner=request.middleware_user).filter(Q(new_owner_accept_at=None) | Q( # old_owner_accept_at=None)).count() >= institute_conf.max_number_of_not_completed_transfer_ownership_requests_per_user: # return MyResponse(request, {'status': 'error', 'message': 'Too many unfinished requests!'}, status=status.HTTP_403_FORBIDDEN) # # ########################## Block 5 ################################ # # generating two random 64 characters string for new_owner_accept_key and old_owner_accept_key and checks if they are unique in database or not! # new_owner_accept_key = ''.join(random.choices( # string.ascii_letters + string.digits, k=64)) # while TransferOwnershipHistory.objects.filter(new_owner_accept_key=new_owner_accept_key): # new_owner_accept_key = ''.join(random.choices( # string.ascii_letters + string.digits, k=64)) # old_owner_accept_key = ''.join(random.choices( # string.ascii_letters + string.digits, k=64)) # while TransferOwnershipHistory.objects.filter(old_owner_accept_key=old_owner_accept_key): # old_owner_accept_key = ''.join(random.choices( # string.ascii_letters + string.digits, k=64)) # # ########################## Block 6 ################################ # # creating a TransferOwnershipHistory record with above values # TransferOwnershipHistory.objects.create( # institute=request.middleware_institute, # old_owner=request.middleware_user, # new_owner=new_owner, # description=description, # new_owner_accept_key=new_owner_accept_key, # old_owner_accept_key=old_owner_accept_key # ) # # ########################## Block 7 ################################ # # email the transfer link to the old owner email # old_owner_link = f'https://panel.wikiazma.com/#institute/transfer/ooc?key={old_owner_accept_key}' # content = institute_pushers.transfer_old_owner_builder( # request.middleware_institute.name, new_owner_email, old_owner_link, request.middleware_user.email) # pushers.send_html_email( # content, request.middleware_user.email) # return MyResponse(request, {'status': 'ok', 'message': 'Successful'}, status=status.HTTP_200_OK) # TODO :: skip # @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_user_token=True), # name='post') # @method_decorator(serializer_validation_middleware(institute_serializers.TransferOOCSerializer), name='post') # class TransferOOC(APIView): # """ # This API class is responsible for handling incoming request from old owner of institute to accept the transfer. # """ # def post(self, request): # # ########################## Block 1 ################################ # # receiving user inputs from serializer # accept_key = request.middleware_serializer_data.get('accept_key') # # ########################## Block 2 ################################ # # try to get a TransferOwnershipHistory record which its old_owner is authenticated user and its old_owner_accept_key is incoming accept_key. # try: # transfer_ownership_record = TransferOwnershipHistory.objects.get( # old_owner=request.middleware_user, old_owner_accept_key=accept_key) # except: # return MyResponse(request, {'status': 'error', 'message': 'Something was wrong.'}, status=status.HTTP_403_FORBIDDEN) # # ########################## Block 3 ################################ # # sets the old_owner_accept_at of transfer_ownership_record to now and old_owner_accept_key to None and saves it # transfer_ownership_record.old_owner_accept_at = timezone.now() # transfer_ownership_record.old_owner_accept_key = None # transfer_ownership_record.save() # # ########################## Block 4 ################################ # # makes a link for new_owner and emails the transfer link to the new owner email # new_owner_link = f'https://panel.wikiazma.com/#institute/transfer/noc?key={transfer_ownership_record.new_owner_accept_key}' # content = institute_pushers.transfer_new_owner_builder( # transfer_ownership_record.institute.name, transfer_ownership_record.new_owner.email, new_owner_link, # transfer_ownership_record.old_owner.email) # pushers.send_html_email( # content, transfer_ownership_record.new_owner.email) # # ########################## Block 5 ################################ # # serializes the transfer_ownership_record to json and passed it to MyResponse # transfer_ownership_record_data = institute_serializers.TransferOwnershipHistorySerializer( # transfer_ownership_record).data # return MyResponse(request, {'status': 'ok', 'message': 'Successful', 'data': transfer_ownership_record_data}, # status=status.HTTP_200_OK) # TODO :: skip # @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_user_token=True), # name='post') # @method_decorator(serializer_validation_middleware(institute_serializers.TransferNOCSerializer), name='post') # class TransferNOC(APIView): # """ # This API class is responsible for handling incoming request from new owner of institute to accept the transfer. # """ # def post(self, request): # # ########################## Block 1 ################################ # # receiving user inputs from serializer # accept_key = request.middleware_serializer_data.get('accept_key') # # ########################## Block 2 ################################ # # try to get a TransferOwnershipHistory record which its new_owner is authenticated user and its new_owner_accept_key is incoming accept_key. # try: # transfer_ownership_record = TransferOwnershipHistory.objects.select_related('institute').get( # new_owner=request.middleware_user, new_owner_accept_key=accept_key) # except: # return MyResponse(request, {'status': 'error', 'message': 'Something was wrong.'}, status=status.HTTP_403_FORBIDDEN) # # ########################## Block 3 ################################ # # check new_owner limitations # if not check_user_limitations(Institute, request.middleware_user): # return MyResponse(request, {'status': 'error', 'message': 'Payment Required'}, status=status.HTTP_402_PAYMENT_REQUIRED) # # ########################## Block 4 ################################ # # sets the new_owner_accept_at of transfer_ownership_record to now and new_owner_accept_key to None and saves it # transfer_ownership_record.new_owner_accept_at = timezone.now() # transfer_ownership_record.new_owner_accept_key = None # transfer_ownership_record.save() # # ########################## Block 5 ################################ # # sets the institute user to new_owner and saves the institute # institute = transfer_ownership_record.institute # institute.user = transfer_ownership_record.new_owner # while Institute.objects.filter(name=institute.name, user=transfer_ownership_record.new_owner): # institute.name += '_' + str(secrets.randbelow(90000) + 10000) # if len(institute.name) > 100: # institute.name = institute.name[:94] + \ # '_' + str(secrets.randbelow(90000) + 10000) # institute.save() # # ########################## Block 6 ################################ # # deletes the old_owner access in this institute # Access.objects.get( # user=transfer_ownership_record.old_owner, institute=institute).delete() # # ########################## Block 7 ################################ # # deletes all new_owner access and invite access records in this institute # Access.objects.filter( # user=transfer_ownership_record.new_owner, institute=institute).delete() # InviteAccess.objects.filter( # invited=transfer_ownership_record.new_owner, institute=institute).delete() # # ########################## Block 8 ################################ # # creates an access record for new_owner in this institute with __ROLE_OWNER__ # Access.objects.create( # user=transfer_ownership_record.new_owner, # institute=institute, # roles=[dr.__ROLE_OWNER__], # institute_name=institute.name, # institute_created_at=institute.created_at # ) # return MyResponse(request, {'status': 'ok', 'message': 'The institute has been transferred to your account.'}, # status=status.HTTP_200_OK) # TODO :: skip # @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_user_token=True), # name='post') # @method_decorator(serializer_validation_middleware(institute_serializers.TransferOwnerShipCancelSerializer), name='post') # @method_decorator(check_access_permission_middleware([dr.__PERMISSION_TRANSFER_OWNERSHIP__], Institute, 'id', 'institute_id'), name='post') # class TransferOwnerShipCancel(APIView): # """ # This API class is responsible for handling incoming request to cancel a transfer ownership of a institute. # """ # def post(self, request): # # ########################## Block 1 ################################ # # receiving user inputs from serializer # transfer_ownership_id = request.middleware_serializer_data.get( # 'transfer_ownership_id') # # ########################## Block 2 ################################ # # gets the TransferOwnershipHistory record with the incoming transfer_ownership_id, institute and which its old_owner is authenticated user # # if exists, then checks if its new_owner_accept_at is null and if so, then deletes the transfer_ownership_record. # # if transfer_ownership_record doesn't exists, user faces an error # # if transfer_ownership_record new_owner_accept_at is not null, first we logs an CRITICAL error # transfer_ownership_list = TransferOwnershipHistory.objects.filter( # id=transfer_ownership_id, institute=request.middleware_institute, old_owner=request.middleware_user) # if len(transfer_ownership_list) != 0: # transfer_ownership_record = transfer_ownership_list[0] # if not transfer_ownership_record.new_owner_accept_at: # transfer_ownership_record.delete() # return MyResponse(request, {'status': 'ok', 'message': 'Transfer canceled.'}, status=status.HTTP_200_OK) # utils.logs_adder( # f'[*CRITICAL*]: TransferOwnerShipCancelAPI: The user {request.middleware_user.email} has a transfer ownership which the new owner accepted it but not transfered successfully!') # return MyResponse(request, {'status': 'error', 'message': 'Something was wrong.'}, status=status.HTTP_403_FORBIDDEN) # TODO :: skip # @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_user_token=True), # name='post') # @method_decorator(serializer_validation_middleware(institute_serializers.TransferEmailCheckSerializer), name='post') # @method_decorator(check_access_permission_middleware([dr.__PERMISSION_TRANSFER_OWNERSHIP__], Institute, 'id', 'institute_id'), name='post') # class TransferEmailCheck(APIView): # """ # This API class is responsible for handling incoming requests to check the existance of new owner email in User database. # """ # def post(self, request): # # ########################## Block 1 ################################ # # receiving user inputs from serializer # new_owner_email = request.middleware_serializer_data.get( # 'new_owner_email').lower() # # ########################## Block 2 ################################ # # checks if incoming new_owner_email exists in User database or not! # try: # user = User.objects.get(email=new_owner_email) # except: # return MyResponse(request, {'status': 'error', 'message': 'We cant find this user'}, status=status.HTTP_403_FORBIDDEN) # # ########################## Block 3 ################################ # # checks if incoming new_owner_email is not as same as authenticated user email # if request.middleware_user.email == new_owner_email: # return MyResponse(request, {'status': 'error', 'message': 'You cant enter your email'}, status=status.HTTP_403_FORBIDDEN) # # ########################## Block 4 ################################ # # serializes the new_owner user to json and passed it to MyResponse # user_data = institute_serializers.NewOwnerSerializer(user).data # return MyResponse(request, {'status': 'ok', 'message': 'Successful', 'user': user_data}, status=status.HTTP_200_OK) @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_user_token=True), name='post') @method_decorator(serializer_validation_middleware(institute_serializers.GenerateAPIKeyInputSerializer), name='post') @method_decorator(check_access_permission_middleware([dr.__PERMISSION_WRITE_APIKEY__], Institute, 'id', 'institute_id'), name='post') class GenerateAPIKey(APIView): def post(self, request): # ########################## Block 1 ################################ # receiving user inputs from serializer expire_at = request.middleware_serializer_data.get('expire_at') roles = request.middleware_serializer_data.get('roles') # ########################## Block 2 ################################ # checks if the institute is not a sanbox one and then creates an api_key record and return it's token if request.middleware_institute.sandbox: return MyResponse(request, {'status': 'error', 'message': 'You can not generate APIKeys for a sandbox institute'}, status=status.HTTP_403_FORBIDDEN) api_key = APIKey.create_record( request.middleware_institute, roles, expire_at) return MyResponse(request, {'status': 'ok', 'message': 'Successful', 'data': {'api_key': api_key}}, status=status.HTTP_200_OK) @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_user_token=True), name='post') @method_decorator(serializer_validation_middleware(institute_serializers.APIKeysListInputSerializer), name='post') @method_decorator(check_access_permission_middleware([dr.__PERMISSION_READ_APIKEY__], Institute, 'id', 'institute_id'), name='post') class APIKeysList(APIView): def post(self, request): if request.middleware_institute.sandbox: return MyResponse(request, {'status': 'error', 'message': 'You can not get APIKeys list for a sandbox institute'}, status=status.HTTP_403_FORBIDDEN) api_keys = APIKey.objects.filter( institute=request.middleware_institute).values('id', 'api_key', 'expire_at') return MyResponse(request, {'status': 'ok', 'message': 'Successful', 'data': {'api_keys': api_keys}}, status=status.HTTP_200_OK) @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_user_token=True), name='post') @method_decorator(serializer_validation_middleware(institute_serializers.DeleteAPIKeyInputSerializer), name='post') @method_decorator(check_access_permission_middleware([dr.__PERMISSION_WRITE_APIKEY__], APIKey, 'id', 'api_key_id'), name='post') class DeleteAPIKey(APIView): def post(self, request): request.middleware_model_record.delete() return MyResponse(request, {'status': 'ok', 'message': 'Successful'}, status=status.HTTP_200_OK) @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_server_api_key=True), name='post') @method_decorator(serializer_validation_middleware(institute_serializers.GetTokenInputSerializer), name='post') class GetToken(APIView): def post(self, request): token = InstituteJWT.create_record(request.middleware_api_key) return MyResponse(request, {'status': 'ok', 'message': 'Successful', 'data': {'token': token}}, status=status.HTTP_200_OK) @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_server_api_key=True), name='post') @method_decorator(serializer_validation_middleware(SetInstituteStudentInputValidator), name='post') class SetInstituteStudent(APIView): """ This API class is responsible for handling incoming requests for adding or editing an InstituteStudent in the database """ def post(self, request): # Receive data from serializer referer_institute_student_identity = request.middleware_serializer_data.get( 'referer_institute_student_identity') first_name = request.middleware_serializer_data.get('first_name') last_name = request.middleware_serializer_data.get('last_name') email = request.middleware_serializer_data.get('email') phone = request.middleware_serializer_data.get('phone') info = request.middleware_serializer_data.get('info') # institute and referer_institute_student_identity are unique together institute_student, created = InstituteStudent.objects.update_or_create( institute=request.middleware_institute, referer_identity=referer_institute_student_identity, defaults={ 'first_name': first_name, 'last_name': last_name, 'email': email, 'phone': phone, 'info': info } ) institute_student_data = GetInstituteStudentSerializer( institute_student).data return MyResponse(request, {'status': 'ok', 'message': 'Successful', 'data': {'institute_student': institute_student_data}}, status=status.HTTP_200_OK) @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_server_api_key=True), name='post') @method_decorator(serializer_validation_middleware(GetInstituteStudentInputValidator), name='post') class GetInstituteStudent(APIView): """ This API class is responsible for handling incoming requests for getting an InstituteStudent """ def post(self, request): # Receive data from serializer wikiazma_institute_student_id = request.middleware_serializer_data.get( 'wikiazma_institute_student_id') referer_institute_student_identity = request.middleware_serializer_data.get( 'referer_institute_student_identity') if wikiazma_institute_student_id: try: institute_student = InstituteStudent.objects.get( institute=request.middleware_institute, id=wikiazma_institute_student_id) except: return MyResponse(request, {'status': 'error', 'message': 'Institute Student not found'}, status=status.HTTP_404_NOT_FOUND) elif referer_institute_student_identity: try: institute_student = InstituteStudent.objects.get(institute=request.middleware_institute, referer_identity=referer_institute_student_identity) except: return MyResponse(request, {'status': 'error', 'message': 'Institute Student not found'}, status=status.HTTP_404_NOT_FOUND) else: return MyResponse(request, {'status': 'error', 'message': 'either wikiazma_institute_student_id or referer_institute_student_identity should be provided'}, status=status.HTTP_400_BAD_REQUEST) institute_student_data = GetInstituteStudentSerializer( institute_student).data return MyResponse(request, {'status': 'ok', 'message': 'Successful', 'data': {'institute_student': institute_student_data}}, status=status.HTTP_200_OK) @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_server_api_key=True), name='post') @method_decorator(serializer_validation_middleware(ListInstituteStudentInputValidator), name='post') class ListInstituteStudents(APIView): """ This API class is responsible for handling incoming requests for listing InstituteStudents """ def post(self, request): # Receive data from serializer skip = request.middleware_serializer_data.get('skip') take = request.middleware_serializer_data.get('take') sort_by = request.middleware_serializer_data.get('sort_by') sort_by_attr = utils.institute_student_sort_by_converter(sort_by) institute_student = InstituteStudent.objects.filter( institute=request.middleware_institute) if sort_by: institute_student = institute_student.order_by(sort_by_attr) institute_student = institute_student[skip:skip + take] institute_student_data = GetInstituteStudentSerializer( institute_student, many=True).data return MyResponse(request, {'status': 'ok', 'message': 'Successful', 'data': {'institute_students': institute_student_data}}, status=status.HTTP_200_OK) @method_decorator(user_token_or_server_token_or_server_API_key_validation_and_authorization(accept_user_token=True), name='post') @method_decorator(serializer_validation_middleware(institute_serializers.InstituteBalanceInputSerialize), name='post') @method_decorator(check_access_permission_middleware([dr.__PERMISSION_READ_INSTITUTE_BILLING__], Institute, 'id', 'institute_id'), name='post') class InstituteBalance(APIView): def post(self, request): institute = request.middleware_model_record balance = institute.balance return MyResponse(request, {'status': 'ok', 'message': 'Successful', 'data': {'balance': balance}}, status=status.HTTP_200_OK)