# By: Riasat Ullah # Contributor: Md. Fahim Bin Amin # This file contains functions related to external people views. from dbqueries.itsm import db_people from exceptions.user_exceptions import InvalidRequest, UnauthorizedRequest from rest_framework.decorators import api_view from rest_framework.response import Response from translators import label_translator as _lt from utils import errors, info, logging, permissions, times, tokenizer, var_names from utils.db_connection import CONN_POOL from validations import request_validator import jwt @api_view(['POST']) def create_person(request, conn=None): ''' Create new people. Always expects a list of person objects. :param request: HttpRequest - expects data field with list of person objects :param conn: db connection :return: HttpResponse -> list of created ref_ids ''' if request.method == 'POST': lang = request_validator.get_user_language(request) expected_fields = [var_names.data] try: conn = CONN_POOL.get_db_conn() if conn is None else conn request_validator.validate_fields(request, expected_fields) user_id, org_id, user_perm, org_perm = tokenizer.authorize_request(request) people_details = request.data[var_names.data] # Check permissions once for all operations if not permissions.has_org_permission(org_perm, permissions.ORG_PEOPLE_PERMISSION): return Response(_lt.get_label(errors.err_subscription_rights, lang), status=403) if not permissions.has_user_permission(user_perm, permissions.USER_COMPONENTS_EDIT_PERMISSION): return Response(_lt.get_label(errors.err_user_rights, lang), status=403) # Check if any person has group_ref_id has_groups = any(var_names.group_ref_id in person and person[var_names.group_ref_id] is not None for person in people_details) if has_groups and not permissions.has_org_permission(org_perm, permissions.ORG_GROUPS_PERMISSION): return Response(_lt.get_label(errors.err_subscription_rights, lang), status=403) do_adv_check, do_comp_check, do_team_check = permissions.get_user_advanced_check_status(user_perm, org_perm) current_time = times.get_current_timestamp() # Process in batches of 100 batch_size = 100 all_ref_ids = [] for i in range(0, len(people_details), batch_size): batch = people_details[i:i + batch_size] ref_ids = db_people.create_person( conn, current_time, org_id, user_id, batch, check_adv_perm=do_adv_check, has_comp_perm=do_comp_check, has_team_perm=do_team_check ) all_ref_ids.extend(ref_ids) return Response(all_ref_ids) except AssertionError as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_invalid_request, lang), status=400) except PermissionError as e: logging.exception(str(e)) return Response(_lt.get_label(str(e), lang), status=403) except (UnauthorizedRequest, jwt.ExpiredSignatureError, jwt.InvalidSignatureError) as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_authorization, lang), status=401) except Exception as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_system_error, lang), status=500) finally: CONN_POOL.put_db_conn(conn) @api_view(['POST']) def edit_person(request, conn=None): ''' Edit a person. :param request: HttpRequest :param conn: db connection :return: HttpResponse -> str ''' if request.method == 'POST': lang = request_validator.get_user_language(request) expected_fields = [var_names.people_ref_id, var_names.first_name, var_names.last_name] optional_fields = [var_names.title, var_names.group_ref_id, var_names.email, var_names.iso_country_code, var_names.phone, var_names.language] try: conn = CONN_POOL.get_db_conn() if conn is None else conn request_validator.validate_fields(request, expected_fields, optional_fields) user_id, org_id, user_perm, org_perm = tokenizer.authorize_request(request) group_ref_id = request.data[var_names.group_ref_id]\ if var_names.group_ref_id in request.data and request.data[var_names.group_ref_id] is not None\ else None if not permissions.has_org_permission(org_perm, permissions.ORG_PEOPLE_PERMISSION): return Response(_lt.get_label(errors.err_subscription_rights, lang), status=403) if group_ref_id is not None and \ not permissions.has_org_permission(org_perm, permissions.ORG_GROUPS_PERMISSION): return Response(_lt.get_label(errors.err_subscription_rights, lang), status=403) if not permissions.has_user_permission(user_perm, permissions.USER_COMPONENTS_EDIT_PERMISSION): return Response(_lt.get_label(errors.err_user_rights, lang), status=403) do_adv_check, do_comp_check, do_team_check = permissions.get_user_advanced_check_status(user_perm, org_perm) db_people.edit_person( conn, times.get_current_timestamp(), org_id, user_id, request.data[var_names.people_ref_id], request.data[var_names.first_name], request.data[var_names.last_name], title=request.data[var_names.title] if var_names.title in request.data else None, email=request.data[var_names.email] if var_names.email in request.data else None, iso_code=request.data[var_names.iso_country_code] if var_names.iso_country_code in request.data else None, phone=request.data[var_names.phone] if var_names.phone in request.data else None, person_language=request.data[var_names.language] if var_names.language in request.data else None, group_ref_id=group_ref_id, check_adv_perm=do_adv_check, has_comp_perm=do_comp_check, has_team_perm=do_team_check ) return Response(_lt.get_label(info.msg_person_edited, lang)) except PermissionError as e: logging.exception(str(e)) return Response(_lt.get_label(str(e), lang), status=403) except InvalidRequest as e: logging.exception(str(e)) return Response(_lt.get_label(str(e), lang), status=400) except (UnauthorizedRequest, jwt.ExpiredSignatureError, jwt.InvalidSignatureError) as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_authorization, lang), status=401) except Exception as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_system_error, lang), status=500) finally: CONN_POOL.put_db_conn(conn) @api_view(['POST']) def delete_person(request, conn=None): ''' Delete a person. :param request: HttpRequest :param conn: db connection :return: HttpResponse -> str ''' if request.method == 'POST': lang = request_validator.get_user_language(request) expected_fields = [var_names.people_ref_id] try: conn = CONN_POOL.get_db_conn() if conn is None else conn request_validator.validate_fields(request, expected_fields) user_id, org_id, user_perm, org_perm = tokenizer.authorize_request(request) if not permissions.has_org_permission(org_perm, permissions.ORG_PEOPLE_PERMISSION): return Response(_lt.get_label(errors.err_subscription_rights, lang), status=403) if not permissions.has_user_permission(user_perm, permissions.USER_COMPONENTS_EDIT_PERMISSION): return Response(_lt.get_label(errors.err_user_rights, lang), status=403) do_adv_check = permissions.get_user_advanced_check_status(user_perm, org_perm)[0] db_people.delete_people(conn, times.get_current_timestamp(), org_id, user_id, request.data[var_names.people_ref_id], check_adv_perm=do_adv_check) return Response(_lt.get_label(info.msg_person_deleted, lang)) except InvalidRequest as e: logging.exception(str(e)) return Response(_lt.get_label(str(e), lang), status=400) except (UnauthorizedRequest, jwt.ExpiredSignatureError, jwt.InvalidSignatureError) as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_authorization, lang), status=401) except Exception as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_system_error, lang), status=500) finally: CONN_POOL.put_db_conn(conn) @api_view(['POST']) def list_people(request, conn=None): ''' Get the list of people associated with the organization. :param request: HttpRequest :param conn: db connection :return: HttpResponse -> str ''' if request.method == 'POST': lang = request_validator.get_user_language(request) optional_fields = [var_names.group_ref_id, var_names.keywords, var_names.row_limit, var_names.row_offset] try: conn = CONN_POOL.get_db_conn() if conn is None else conn request_validator.validate_fields(request, [], optional_fields) user_id, org_id, user_perm, org_perm = tokenizer.authorize_request(request) group_ref_id = request.data[var_names.group_ref_id]\ if var_names.group_ref_id in request.data and request.data[var_names.group_ref_id] is not None\ else None if not permissions.has_org_permission(org_perm, permissions.ORG_PEOPLE_PERMISSION): return Response(_lt.get_label(errors.err_subscription_rights, lang), status=403) if group_ref_id is not None and \ not permissions.has_org_permission(org_perm, permissions.ORG_GROUPS_PERMISSION): return Response(_lt.get_label(errors.err_subscription_rights, lang), status=403) if not permissions.has_user_permission(user_perm, permissions.USER_COMPONENTS_VIEW_PERMISSION): return Response(_lt.get_label(errors.err_user_rights, lang), status=403) do_adv_check = permissions.get_user_advanced_check_status(user_perm, org_perm)[0] people_list = db_people.list_people( conn, times.get_current_timestamp(), org_id, user_id, group_ref_id=group_ref_id, keywords=request.data[var_names.keywords] if var_names.keywords in request.data else None, row_limit=request.data[var_names.row_limit] if var_names.row_limit in request.data else None, row_offset=request.data[var_names.row_offset] if var_names.row_offset in request.data else None, check_adv_perm=do_adv_check ) return Response(people_list) except InvalidRequest as e: logging.exception(str(e)) return Response(_lt.get_label(str(e), lang), status=400) except (UnauthorizedRequest, jwt.ExpiredSignatureError, jwt.InvalidSignatureError) as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_authorization, lang), status=401) except Exception as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_system_error, lang), status=500) finally: CONN_POOL.put_db_conn(conn)