from appservices.common.util import * from appservices.common.form_schemas import * api_security = Blueprint("api_security",__name__) def fetching_api_security_details(api_security_queryset): apiDict = {} try: # Collecting all IP addresses and filtering out None or empty values ip_addresses = filter(None, [ api_security_queryset.ipAddress, api_security_queryset.ipAddress2, api_security_queryset.ipAddress3, api_security_queryset.ipAddress4 ]) # Constructing the dictionary apiDict = { "id": str(api_security_queryset.id), "userId": str(api_security_queryset.userId.id), "fullName": api_security_queryset.userId.fullName, "createdOn": api_security_queryset.createdOn.strftime("%d-%m-%Y"), "validStartDate": api_security_queryset.validStartDate.strftime("%d-%m-%Y"), "apiType": api_security_queryset.apiType, "ipAddress": api_security_queryset.ipAddress, "ipAddress2": api_security_queryset.ipAddress2, "ipAddress3": api_security_queryset.ipAddress3, "ipAddress4": api_security_queryset.ipAddress4, "apiKey": api_security_queryset.apiKey, "secretKey": api_security_queryset.secretKey, "callBackUrl": api_security_queryset.callBackUrl, "returnUrl": api_security_queryset.returnUrl, "validFor": api_security_queryset.validFor, # Combine all non-empty IPs into a single string "ipAddressList": ", ".join(ip_addresses) if ip_addresses else "" } # Handling validEndDate apiDict["validEndDate"] = ( api_security_queryset.validEndDate.strftime("%d-%m-%Y") if api_security_queryset.validEndDate else "always" ) # Handling status apiDict["actionText"] = "Active" if api_security_queryset.status == 1 else "Deactive" # Adjusting date format apiDict["createdOn"] = ( api_security_queryset.createdOn.strftime("%m-%d-%Y") if api_security_queryset.createdOn else "" ) except Exception as e: app.logger.error(traceback.format_exc()) return apiDict @api_security.route("/get_api_security_list",methods=["POST","GET"]) @adminid_access_token_required def get_api_security_list(): if not session.get("adminId"): return redirect("admin_login") apiSecurityList = [] userList = [] pagination = "" search_element="" snoCount=0 apiSecurityPage=None adminId = session.get("adminId") permissionsList = check_permissions(session.get("adminId"),"apiSecurityPermissions") form = SearchApiType(request.args) if "view" in permissionsList: try: redirectTo = request.args.get("redirectTo","apiSecurity") if redirectTo: redirectval = redirectTo else: redirectval = "apiSecurity" search_element = request.args.get("search_element","").strip() if form.validate(): if search_element: search_element = form.search_element.data.strip() apiSecurityPage = request.args.get(get_page_parameter("apiSecurityPage"), type=int, default=1) per_page = 20 start = (apiSecurityPage - 1) * per_page total_count=0 filters = Q(status__in=[0, 1]) print("(((((((((((((((((((search_element)))))))))))))))))))",search_element) if search_element and search_element.strip(): filters &= Q(apiType__icontains=search_element.strip()) total_count = ApiSecurities.objects(filters).count() api_security_queryset =( ApiSecurities.objects(filters) .order_by("-id") .skip(start) .limit(per_page) ) snoCount = start apiSecurityList = list(api_security_queryset) user_queryset = Users.objects(status__in=[0,1]).only("id","fullName").order_by("-id").all() # for each_user in user_queryset: # userDict = fetching_user_details(each_user) # userList.append(userDict) userList=list(user_queryset) # Pagination object for rendering pagination controls in the template pagination = Pagination(apiSecurityPage=apiSecurityPage, total=total_count, per_page=per_page, page_parameter ="apiSecurityPage", alignment="right", record_name="apiSecurity") return render_template("super_admin_templates/api_security_list.html", pagination=pagination, apiSecurityList=apiSecurityList, userList=userList, redirectval=redirectval, search_element=search_element, snoCount = snoCount, apiSecurityPage=apiSecurityPage, form = form ) # api_security_queryset = ApiSecurities.objects(status__in=[0,1]).order_by("-id") # if search_element: # api_security_queryset = api_security_queryset.filter(Q(apiType__icontains=search_element)) # # Get the current page from the query parameters # page = request.args.get(get_page_parameter(), type=int, default=1) # per_page = 20 # Number of items per page # # Query the database for the current page's data # total_count = api_security_queryset.count() # start = (page - 1) * per_page # end = min(start + per_page, total_count) # total_api_security = api_security_queryset[start:end] # snoCount = start # for each_api_security in total_api_security: # snoCount +=1 # apiDict = fetching_api_security_details(each_api_security) # apiDict["snoCount"]=snoCount # apiSecurityList.append(apiDict) # # Pagination object for rendering pagination controls in the template # pagination = Pagination(page=page, total=total_count, per_page=per_page, alignment="right", record_name="apiSecurity") # user_queryset = Users.objects(status__in=[0,1]).order_by("-id").all() # for each_user in user_queryset: # userDict = fetching_user_details(each_user) # userList.append(userDict) # return render_template("super_admin_templates/api_security_list.html", # pagination=pagination, # apiSecurityList=apiSecurityList, # userList=userList, # redirectval=redirectval, # search_element=search_element # ) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to fetch api security details!!" return render_template("super_admin_templates/api_security_list.html", error=error, pagination=pagination, apiSecurityList=apiSecurityList, userList=userList, redirectval=redirectval, search_element=search_element, snoCount = snoCount, apiSecurityPage=apiSecurityPage, form = form ) else: flash("The staff member does not have permission to view api security !", "danger") return redirect(url_for("admin.dashboard")) # return render_template("super_admin_templates/bank_masters_list.html") # @api_security.route("/create_api_security",methods=["POST"]) # def create_api_security(): # try: # if not session.get("adminId"): # return redirect("admin_login") # adminId = session.get("adminId") # if request.method == "POST": # userId = request.form.get("userId","") # apiType = request.form.get("apiType","") # "payout","auto_collect","payment_gateway" # # ipAddress = request.form.get("ipAddress","") # validFor = request.form.get("validFor","") # "one_month","one_year","always" # # ipAddressList = request.form.get("ipAddressList") # ip_addresses = [ip.strip() for ip in request.form.getlist('ipAddress[]') if ip.strip()] # ip_address_list = ", ".join(ip_addresses) # if userId and apiType and validFor and ip_addresses: # token = secrets.token_hex(16) # apiKey = userId+"-"+str(token) # secretKey = str(secrets.token_hex(16)) # ipAddressList = ", ".join(ip_addresses) # api_security_table = ApiSecurities( # userId=userId, # apiType=apiType, # ipAddressList=ip_address_list, # apiKey=apiKey, # secretKey=secretKey, # createdOn=datetime.datetime.now(), # status=1 # ) # save_table=api_security_table.save() # # Get the current date and time # current_date = datetime.datetime.now() # if validFor == "one_month": # # Add one month to the current date and time # current_date_plus_one_month = current_date + relativedelta(months=1) # save_table.update(validFor=validFor,validStartDate=current_date,validEndDate=current_date_plus_one_month) # elif validFor == "one_year": # # Add one year to the current date and time # current_date_plus_one_year = current_date + relativedelta(years=1) # save_table.update(validFor=validFor,validStartDate=current_date,validEndDate=current_date_plus_one_year) # else: # save_table.update(validFor=validFor,validStartDate=current_date,validEndDate=None) # flash("API generated successfully!") # return redirect(url_for("api_security.get_api_security_list")) # else: # flash("Required fields are missing!!") # return redirect(url_for("api_security.get_api_security_list")) # except Exception as e: # app.logger.error(traceback.format_exc()) # error = "Unable to generate api!!" # return render_template("super_admin_templates/api_security_list.html",error=error) @api_security.route("/create_api_security", methods=["POST","GET"]) @adminid_access_token_required @csrf_protect def create_api_security(): data_status={"responseStatus":0,"result":""} try: if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") csrf_token = request.form.get("csrf_token") print(csrf_token,"((((((((((((((csrf_token))))))))))))))") latitude = request.form.get("latitude", "") longitude = request.form.get("longitude", "") loginBrowser = request.headers.get("Sec-Ch-Ua") if loginBrowser: loginBrowseData = loginBrowser.split(";") browser = loginBrowseData[0] else: loginBrowseData = request.headers.get('User-Agent').split(";") browser = loginBrowseData[0] client_ip=0 # Extracting client IP address if request.headers.getlist("X-Forwarded-For"): client_ip = request.headers.getlist("X-Forwarded-For")[0] else: client_ip = request.remote_addr actionDate = datetime.datetime.now() ipAddressList=[] permissionsList = check_permissions(adminId,"apiSecurityPermissions") if "add" in permissionsList: if request.method == "POST": userId = request.form.get("userId", "") apiType = request.form.get("apiType", "") ipAddress = request.form.get("ipAddress", "") ipAddress2 = request.form.get("ipAddress2", "") ipAddress3 = request.form.get("ipAddress3", "") ipAddress4 = request.form.get("ipAddress4", "") validFor = request.form.get("validFor", "") callBackUrl = request.form.get("callBackUrl", "") returnUrl = request.form.get("returnUrl", "") print(ipAddress,"((((((((((ipAddress))))))))))") jsonData = request.form.to_dict(flat=True) requestData = [jsonData] updatedrequestData = [jsonData] form = CreateApiSecurityList(request.form) if form.validate_on_submit(): # Combine non-empty IPs into a comma-separated list # ipAddressList = ", ".join(filter(None, [ipAddress, ipAddress2, ipAddress3, ipAddress4])) ipAddressList = list(filter(None, [ipAddress, ipAddress2, ipAddress3, ipAddress4])) if userId and apiType and validFor and ipAddress: try: api_security_queryset = ApiSecurities.objects(userId=userId,apiType=apiType).first() if api_security_queryset: error = "Api type already exists with selected user" data_status["responseStatus"] = 2 data_status["result"]= error return data_status user_queryet = Users.objects(id=userId).only("fullName").first() if not user_queryet: error = "Invalid user" data_status["responseStatus"] = 2 data_status["result"]= error return data_status admin_queryset = SuperAdmin.objects(id=adminId,status=1).first() if admin_queryset: message = admin_queryset.userName+" "+user_queryet.fullName+" API Security generated successfully!" save_admin_log_table = save_admin_logs_data(adminId,None,None,"create_api_security","create",actionDate,client_ip,browser,message,requestData,updatedrequestData,latitude,longitude) else: message = "No api found" token = secrets.token_hex(16) apiKey = userId + "-" + str(token) secretKey = str(secrets.token_hex(16)) api_security_table = ApiSecurities( userId=userId, apiType=apiType, ipAddress=ipAddress, ipAddress2=ipAddress2, ipAddress3=ipAddress3, ipAddress4=ipAddress4, ipAddressList=ipAddressList, apiKey=apiKey, secretKey=secretKey, callBackUrl=callBackUrl, returnUrl=returnUrl, createdOn=datetime.datetime.now(), status=1 ) save_table = api_security_table.save() # Get the current date and time current_date = datetime.datetime.now() if validFor == "one_month": current_date_plus_one_month = current_date + relativedelta(months=1) save_table.update(validFor=validFor, validStartDate=current_date, validEndDate=current_date_plus_one_month) elif validFor == "one_year": current_date_plus_one_year = current_date + relativedelta(years=1) save_table.update(validFor=validFor, validStartDate=current_date, validEndDate=current_date_plus_one_year) else: save_table.update(validFor=validFor, validStartDate=current_date, validEndDate=None) flash("API generated successfully!") data_status['responseStatus']=1 return data_status except Exception as e: flash("Unable to save ticket category!!") app.logger.error(traceback.format_exc()) data_status['responseStatus']=4 print("=====================================",traceback.format_exc()) return data_status else: data_status['responseStatus']=2 data_status['result']="Required fields are missing!!" return data_status # # Combine non-empty IPs into a comma-separated list # # ipAddressList = ", ".join(filter(None, [ipAddress, ipAddress2, ipAddress3, ipAddress4])) # ipAddressList = list(filter(None, [ipAddress, ipAddress2, ipAddress3, ipAddress4])) # if ipAddress and not is_valid_numeric_allow_muldots(ipAddress): # flash("Ip Address contain numeric characters and allows multiple dots.") # return redirect(url_for("api_security.get_api_security_list")) # if ipAddress2 and not is_valid_numeric_allow_muldots(ipAddress2): # flash("Ip Address contain numeric characters and allows multiple dots.") # return redirect(url_for("api_security.get_api_security_list")) # if ipAddress3 and not is_valid_numeric_allow_muldots(ipAddress3): # flash("Ip Address contain numeric characters and allows multiple dots.") # return redirect(url_for("api_security.get_api_security_list")) # if ipAddress4 and not is_valid_numeric_allow_muldots(ipAddress4): # flash("Ip Address contain numeric characters and allows multiple dots.") # return redirect(url_for("api_security.get_api_security_list")) # if userId and apiType and validFor and ipAddress: # try: # admin_queryset = SuperAdmin.objects(id=adminId,status=1).first() # if admin_queryset: # message = admin_queryset.userName+" API generated successfully!" # save_admin_log_table = save_admin_logs_data(adminId,None,None,"create_api_security","create",actionDate,client_ip,browser,message,requestData,updatedrequestData) # token = secrets.token_hex(16) # apiKey = userId + "-" + str(token) # secretKey = str(secrets.token_hex(16)) # api_security_table = ApiSecurities( # userId=userId, # apiType=apiType, # ipAddress=ipAddress, # ipAddress2=ipAddress2, # ipAddress3=ipAddress3, # ipAddress4=ipAddress4, # ipAddressList=ipAddressList, # apiKey=apiKey, # secretKey=secretKey, # callBackUrl=callBackUrl, # returnUrl=returnUrl, # createdOn=datetime.datetime.now(), # status=1 # ) # save_table = api_security_table.save() # # Get the current date and time # current_date = datetime.datetime.now() # if validFor == "one_month": # current_date_plus_one_month = current_date + relativedelta(months=1) # save_table.update(validFor=validFor, validStartDate=current_date, validEndDate=current_date_plus_one_month) # elif validFor == "one_year": # current_date_plus_one_year = current_date + relativedelta(years=1) # save_table.update(validFor=validFor, validStartDate=current_date, validEndDate=current_date_plus_one_year) # else: # save_table.update(validFor=validFor, validStartDate=current_date, validEndDate=None) # flash("API generated successfully!") # return redirect(url_for("api_security.get_api_security_list")) # except Exception as e: # flash("Unable to save ticket category!!") # app.logger.error(traceback.format_exc()) # return redirect(url_for("api_security.get_api_security_list")) # else: # flash("Required fields are missing!!") # return redirect(url_for("api_security.get_api_security_list")) else: data_status['result']=form.errors return data_status else: return redirect(url_for("api_security.get_api_security_list")) else: flash("The staff member does not have permission to create a API Security.", "danger") data_status['responseStatus']=4 return data_status except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to generate API!!" flash("Unable to generate API!!", "danger") data_status['responseStatus']=4 print("=====================================",traceback.format_exc()) return data_status # @api_security.route("/update_api_security",methods=["POST"]) # def update_api_security(): # if not session.get("adminId"): # return redirect("admin_login") # try: # if request.method == "POST": # apiId = request.form.get("apiId","") # userId = request.form.get("userId","") # apiType = request.form.get("apiType","") # # ipAddress = request.form.get("ipAddress","") # validFor = request.form.get("validFor") # # ipAddressList = request.form.get("ipAddressList") # ip_addresses = [ip.strip() for ip in request.form.getlist("ipAddress[]") if ip.strip()] # ip_address_list = ", ".join(ip_addresses) # if apiId and userId and ipAddress: # api_security_queryset = ApiSecurities.objects(id=apiId,userId=userId,status__in=[1]).first() # if not api_security_queryset: # flash("Invaild api id!!") # return redirect(url_for("api_security.get_api_security_list")) # api_security_queryset.update( # apiType=apiType, # ipAddress=ipAddress, # ipAddressList=ip_address_list, # validFor=validFor # ) # # Get the current date and time # current_date = datetime.datetime.now() # if validFor == "one_month": # # Add one month to the current date and time # current_date_plus_one_month = current_date + relativedelta(months=1) # api_security_queryset.update(validFor=validFor,validStartDate=current_date,validEndDate=current_date_plus_one_month) # elif validFor == "one_year": # # Add one year to the current date and time # current_date_plus_one_year = current_date + relativedelta(years=1) # api_security_queryset.update(validFor=validFor,validStartDate=current_date,validEndDate=current_date_plus_one_year) # else: # api_security_queryset.update(validFor=validFor,validStartDate=current_date,validEndDate=None) # flash("API updated successfully!") # return redirect(url_for("api_security.get_api_security_list")) # else: # flash("Required fields are missing!!") # return redirect(url_for("api_security.get_api_security_list")) # except Exception as e: # app.logger.error(traceback.format_exc()) # error = "Unable to update api!!" # flash(error) # return redirect(url_for("api_security.get_api_security_list")) @api_security.route("/update_api_security", methods=["POST","GET"]) @adminid_access_token_required @csrf_protect def update_api_security(): data_status={"responseStatus":0,"result":""} if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") csrf_token = request.form.get("csrf_token") print(csrf_token,"((((((((((((((csrf_token))))))))))))))") latitude = request.form.get("latitude", "") longitude = request.form.get("longitude", "") loginBrowser = request.headers.get("Sec-Ch-Ua") if loginBrowser: loginBrowseData = loginBrowser.split(";") browser = loginBrowseData[0] else: loginBrowseData = request.headers.get('User-Agent').split(";") browser = loginBrowseData[0] existing_record = "" client_ip=0 # Extracting client IP address if request.headers.getlist("X-Forwarded-For"): client_ip = request.headers.getlist("X-Forwarded-For")[0] else: client_ip = request.remote_addr actionDate=datetime.datetime.now() ipAddressList=[] try: permissionsList = check_permissions(session.get("adminId"),"apiSecurityPermissions") if "edit" in permissionsList: apiId = request.args.get("apiId", "") if request.method == "POST": userId = request.form.get("userId", "") apiType = request.form.get("apiType", "") ipAddress = request.form.get("ipAddress", "") ipAddress2 = request.form.get("ipAddress2", "") ipAddress3 = request.form.get("ipAddress3", "") ipAddress4 = request.form.get("ipAddress4", "") validFor = request.form.get("validFor", "") callBackUrl = request.form.get("callBackUrl", "") returnUrl = request.form.get("returnUrl", "") jsonData = request.form.to_dict(flat=True) form = CreateApiSecurityList(request.form, current_id=apiId) if form.validate_on_submit(): try: # Fetch the existing API security record admin_queryset = SuperAdmin.objects(id=adminId,status=1).first() api_security_queryset = ApiSecurities.objects(userId=userId, status__in=[1]).first() existing_record = api_security_queryset.to_json() message = admin_queryset.userName+" "+api_security_queryset.userId.fullName+" API Security updated successfully!" requestData = [api_security_queryset] updatedrequestData=[jsonData] save_admin_log_table = save_admin_logs_data(adminId,None,None,"update_api_security","update",actionDate,client_ip,browser,message,requestData,updatedrequestData,latitude,longitude) if not api_security_queryset: flash("Invalid API ID!!") data_status["responseStatus"]=4 return data_status # Combine IPs into a single list ipAddressList = list(filter(None, [ipAddress, ipAddress2, ipAddress3, ipAddress4])) # Update the record api_security_queryset.update( apiType=apiType, ipAddress=ipAddress, ipAddress2=ipAddress2, ipAddress3=ipAddress3, ipAddress4=ipAddress4, ipAddressList=ipAddressList, validFor=validFor, callBackUrl=callBackUrl, returnUrl=returnUrl, ) # Handle Validity Dates current_date = datetime.datetime.now() if validFor == "one_month": # Add one month to the current date and time current_date_plus_one_month = current_date + relativedelta(months=1) api_security_queryset.update(validFor=validFor,validStartDate=current_date,validEndDate=current_date_plus_one_month) elif validFor == "one_year": # Add one year to the current date and time current_date_plus_one_year = current_date + relativedelta(years=1) api_security_queryset.update(validFor=validFor,validStartDate=current_date,validEndDate=current_date_plus_one_year) else: api_security_queryset.update(validFor=validFor,validStartDate=current_date,validEndDate=None) flash("API updated successfully!") data_status["responseStatus"]=1 return data_status except Exception as e: flash("Unable to update ticket category!!") app.logger.error(traceback.format_exc()) data_status['responseStatus']=4 return data_status # try: # # Fetch the existing API security record # admin_queryset = SuperAdmin.objects(id=adminId,status=1).first() # api_security_queryset = ApiSecurities.objects(userId=userId, status__in=[1]).first() # existing_record = api_security_queryset.to_json() # message = admin_queryset.userName+" API updated successfully!" # requestData = [api_security_queryset] # updatedrequestData=[jsonData] # save_admin_log_table = save_admin_logs_data(adminId,None,None,"update_api_security","update",actionDate,client_ip,browser,message,requestData,updatedrequestData) # if not api_security_queryset: # flash("Invalid API ID!!") # return redirect(url_for("api_security.get_api_security_list")) # # Combine IPs into a single list # ipAddressList = list(filter(None, [ipAddress, ipAddress2, ipAddress3, ipAddress4])) # # Update the record # api_security_queryset.update( # apiType=apiType, # ipAddress=ipAddress, # ipAddress2=ipAddress2, # ipAddress3=ipAddress3, # ipAddress4=ipAddress4, # ipAddressList=ipAddressList, # validFor=validFor, # callBackUrl=callBackUrl, # returnUrl=returnUrl, # ) # # Handle Validity Dates # current_date = datetime.datetime.now() # if validFor == "one_month": # # Add one month to the current date and time # current_date_plus_one_month = current_date + relativedelta(months=1) # api_security_queryset.update(validFor=validFor,validStartDate=current_date,validEndDate=current_date_plus_one_month) # elif validFor == "one_year": # # Add one year to the current date and time # current_date_plus_one_year = current_date + relativedelta(years=1) # api_security_queryset.update(validFor=validFor,validStartDate=current_date,validEndDate=current_date_plus_one_year) # else: # api_security_queryset.update(validFor=validFor,validStartDate=current_date,validEndDate=None) # flash("API updated successfully!") # return redirect(url_for("api_security.get_api_security_list")) # except Exception as e: # app.logger.error(traceback.format_exc()) # error = "Unable to update ticket category!!" # return redirect(url_for("api_security.get_api_security_list")) else: data_status['result']=form.errors return data_status else: return redirect(url_for("api_security.get_api_security_list")) else: flash("The staff member does not have permission to update api security.", "danger") data_status['responseStatus']=4 return data_status except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to update API security details!") data_status['responseStatus']=4 return data_status @api_security.route("/activate_or_deactivate_api_security_status",methods=["POST","GET"]) @adminid_access_token_required def activate_or_deactivate_api_security_status(): if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") latitude = request.args.get("latitude", "") longitude = request.args.get("longitude", "") loginBrowser = request.headers.get("Sec-Ch-Ua") if loginBrowser: loginBrowseData = loginBrowser.split(";") browser = loginBrowseData[0] else: loginBrowseData = request.headers.get("User-Agent").split(";") browser = loginBrowseData[0] client_ip=0 # Extracting client IP address if request.headers.getlist("X-Forwarded-For"): client_ip = request.headers.getlist("X-Forwarded-For")[0] else: client_ip = request.remote_addr actionDate = datetime.datetime.now() jsonData = request.form.to_dict(flat=True) existing_record = "" updatedrequestData = [jsonData] apiSecurityPage = request.args.get("apiSecurityPage") search_element = request.args.get("search_element", "").strip() permissionsList = check_permissions(session.get("adminId"),"apiSecurityPermissions") if "edit" in permissionsList: apiId = request.args.get("apiId","") if apiId: try: message="" admin_queryset = SuperAdmin.objects(id=adminId,status=1).first() api_security_queryset = ApiSecurities.objects(id=apiId,status__nin=[2]).first() existing_record = api_security_queryset.to_json() requestData = [existing_record] if api_security_queryset: if api_security_queryset.status == 0: api_security_queryset.update(status=1) flash("API status activated successfully!") message = admin_queryset.userName+" "+api_security_queryset.userId.fullName+" API Security status activated successfully!" elif api_security_queryset.status == 1: api_security_queryset.update(status=0) flash("API status deactivated successfully!") message = admin_queryset.userName+" "+api_security_queryset.userId.fullName+" API Security status deactivated successfully!" save_admin_log_table = save_admin_logs_data(adminId,None,None,"activate_or_deactivate_api_security_status","updatestatus",actionDate,client_ip,browser,message,requestData,updatedrequestData,latitude,longitude) return redirect(url_for("api_security.get_api_security_list",redirectTo="apiSecurity",search_element=search_element,page_parameter ="apiSecurityPage",apiSecurityPage=apiSecurityPage,)) else: flash("Invaild id!!") return redirect(url_for("api_security.get_api_security_list",redirectTo="apiSecurity",search_element=search_element,page_parameter ="apiSecurityPage",apiSecurityPage=apiSecurityPage)) except Exception as e: app.logger.error(traceback.format_exc()) return redirect(url_for("api_security.get_api_security_list",redirectTo="apiSecurity",search_element=search_element,page_parameter ="apiSecurityPage",apiSecurityPage=apiSecurityPage)) else: flash("Required field is missing!!") return redirect(url_for("api_security.get_api_security_list",redirectTo="apiSecurity",search_element=search_element,page_parameter ="apiSecurityPage",apiSecurityPage=apiSecurityPage)) else: flash("The staff member does not have permission to update api security status.", "danger") return redirect(url_for("api_security.get_api_security_list",redirectTo="apiSecurity",search_element=search_element,page_parameter ="apiSecurityPage",apiSecurityPage=apiSecurityPage)) @api_security.route("/delete_api_security",methods=["POST","GET"]) @adminid_access_token_required def delete_api_security(): try: if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") apiId = request.args.get("apiId","") latitude = request.args.get("latitude", "") longitude = request.args.get("longitude", "") loginBrowser = request.headers.get("Sec-Ch-Ua") if loginBrowser: loginBrowseData = loginBrowser.split(";") browser = loginBrowseData[0] else: loginBrowseData = request.headers.get("User-Agent").split(";") browser = loginBrowseData[0] client_ip=0 # Extracting client IP address if request.headers.getlist("X-Forwarded-For"): client_ip = request.headers.getlist("X-Forwarded-For")[0] else: client_ip = request.remote_addr actionDate = datetime.datetime.now() jsonData = request.form.to_dict(flat=True) existing_record ="" updatedrequestData = [jsonData] apiSecurityPage = 1 search_element = "" permissionsList = check_permissions(session.get("adminId"),"apiSecurityPermissions") if "delete" in permissionsList: try: if apiId: apiSecurityPage = request.args.get(get_page_parameter("apiSecurityPage"), type=int, default=1) search_element = request.args.get("search_element","") admin_queryset = SuperAdmin.objects(id=adminId,status=1).first() api_security_queryset = ApiSecurities.objects(id=apiId,status__in=[0,1]).first() existing_record = api_security_queryset.to_json() requestData = [existing_record] api_security_queryset.update(status=2) flash("API deleted successfully!") message = admin_queryset.userName+" "+api_security_queryset.userId.fullName+" API Security deleted successfully!" save_admin_log_table = save_admin_logs_data(adminId,None,None,"delete_api_security","delete",actionDate,client_ip,browser,message,requestData,updatedrequestData,latitude,longitude) return redirect(url_for("api_security.get_api_security_list",apiSecurityPage=apiSecurityPage, search_element = search_element)) else: flash("Required fields are missing!!") return redirect(url_for("api_security.get_api_security_list",apiSecurityPage=apiSecurityPage, search_element = search_element)) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to delete api!!" return redirect(url_for("api_security.get_api_security_list",apiSecurityPage=apiSecurityPage, search_element = search_element)) else: flash("The staff member does not have permission to delete api security.", "danger") return redirect(url_for("api_security.get_api_security_list",apiSecurityPage=apiSecurityPage, search_element = search_element)) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to update API security!!" return redirect(url_for("api_security.get_api_security_list",error=error))