+ return self.common_db.get_one("projects", {"_id": project_id})
+
+ def get_k8sclusters(self):
+ return self.common_db.get_list("k8sclusters")
+
+ def create_alarm(self, alarm: Alarm):
+ action_data = {"uuid": alarm.uuid, "action": alarm.action}
+ self.common_db.create("alarms_action", action_data)
+ return self.common_db.create("alarms", alarm.to_dict())
+
+ def delete_alarm(self, alarm_uuid: str):
+ self.common_db.del_one("alarms_action", {"uuid": alarm_uuid})
+ return self.common_db.del_one("alarms", {"uuid": alarm_uuid})
+
+ def get_alarms(self) -> List[Alarm]:
+ alarms = []
+ alarm_dicts = self.common_db.get_list("alarms")
+ for alarm_dict in alarm_dicts:
+ alarms.append(Alarm.from_dict(alarm_dict))
+ return alarms
+
+ def update_alarm_status(self, alarm_state: str, uuid):
+ modified_count = self.common_db.set_one("alarms", {"uuid": uuid}, {"alarm_status": alarm_state})
+ return modified_count
+
+ def get_alarm_by_uuid(self, uuid: str):
+ return self.common_db.get_one("alarms", {"uuid": uuid})
+
+ def get_user(self, username: str):
+ return self.common_db.get_one("users", {"username": username})
+
+ def get_user_by_id(self, userid: str):
+ return self.common_db.get_one("users", {"_id": userid})
+
+ def get_role_by_name(self, name: str):
+ return self.common_db.get_one("roles", {"name": name})
+
+ def get_role_by_id(self, role_id: str):
+ return self.common_db.get_one("roles", {"_id": role_id})