in microservices/user_management/src/routes/permission.py [0:0]
def get_permissions(skip: int = Query(None, ge=0, le=2000),
limit: int = Query(None, ge=1, le=100),
application_ids: Optional[str] = None,
module_ids: Optional[str] = None,
action_ids: Optional[str] = None,
user_groups: Optional[str] = None,
fetch_tree: Optional[bool] = False,
sort_by: Optional[Literal["application", "module",
"action", "user_groups","created_time"]]
= "created_time",
sort_order: Optional[Literal["ascending", "descending"]]
= "descending"):
"""The get permissions endpoint will fetch the permissions
matching the values in request body from firestore
### Args:
skip (int): Number of objects to be skipped
limit (int): Size of user array to be returned
application_ids (str): Application id which is to be filtered
module_ids (str): Module id which is to be filtered
action_ids (str): Action id which is to be filtered
user_groups (str): User groups to be filtered
fetch_tree (bool): To fetch the entire object
instead of the UUID of the object
sort_by (str): sorting field name
sort_order (str): ascending / descending
### Raises:
Exception: 500 Internal Server Error if something went wrong
### Returns:
AllPermissionResponseModel: Array of Permission Object
"""
try:
collection_manager = Permission.collection
filtered_list = []
if application_ids is not None or module_ids is not None or\
action_ids is not None or user_groups is not None:
input_permission_dict = {
"application_id": application_ids,
"module_id": module_ids,
"action_id": action_ids,
"user_groups": [ug.strip() for ug in user_groups.split(",")]
if user_groups is not None else None
}
fetch_length = skip + limit
permissions = list(Permission.collection.order("-created_time").fetch())
permissions_with_fields = [
i.get_fields(reformat_datetime=True) for i in permissions
]
# filter permissions
for idx, permission in enumerate(permissions_with_fields):
match = True
for key, value in input_permission_dict.items():
if value is not None:
if isinstance(value,list):
if not any(item in permission[key] for item in value):
match = False
break
elif permission[key] not in value:
match = False
break
if match:
filtered_list.append(permissions[idx])
if len(filtered_list) == fetch_length:
break
filtered_list = filtered_list[skip:fetch_length]
else:
filtered_list = collection_manager.order("-created_time").offset(skip).\
fetch(limit)
if fetch_tree:
filtered_list = get_data_for_fetch_tree(filtered_list)
# sort permissions
if sort_by in ["application", "module", "action"]:
filtered_list = sort_records(sort_by="name",sort_order=sort_order,
records=filtered_list, key=sort_by+"_id")
elif sort_by == "user_groups":
reverse = sort_order == "descending"
# Sort user_groups array within each record
for record in filtered_list:
record["user_groups"] = sorted(record["user_groups"],key=itemgetter(
"name"),reverse=reverse)
# Sort user_groups array across all records
filtered_list = sorted(filtered_list, key=lambda x:(itemgetter("name")\
(x["user_groups"][0])) if x["user_groups"] else "" ,reverse=reverse)
else:
filtered_list = [i.get_fields(reformat_datetime=True) for i in
filtered_list]
count = 10000
response = {"records": filtered_list, "total_count": count}
return {
"success": True,
"message": "Data fetched successfully",
"data": response
}
except ResourceNotFoundException as e:
raise ResourceNotFound(str(e)) from e
except ValidationError as e:
raise BadRequest(str(e)) from e
except Exception as e:
raise InternalServerError(str(e)) from e