You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
46 lines
1.8 KiB
Python
46 lines
1.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
from typing import List, Dict, Any, Union
|
|
|
|
from sqlalchemy import text
|
|
|
|
from website.db_mysql import get_session, to_json_list, to_json
|
|
|
|
|
|
class EnterpriseNodeAlertRepository(object):
|
|
|
|
def get_all(self) -> List[Dict]:
|
|
with get_session() as session:
|
|
sql = text(
|
|
"""SELECT * FROM `enterprise_alert` WHERE `enterprise_node_id`=:enterprise_node_id ORDER BY `created_at` DESC;"""
|
|
)
|
|
cursor = session.execute(
|
|
sql, {"enterprise_node_id": self.enterprise_node_id}
|
|
)
|
|
|
|
def get_one(
|
|
self, enterprise_suid: str, enterprise_node_id: int
|
|
) -> Union[Dict, None]:
|
|
with get_session() as session:
|
|
sql = text(
|
|
"""SELECT * FROM `enterprise_alert` WHERE `enterprise_suid`=:enterprise_suid and `node_id`=:node_id;"""
|
|
)
|
|
cursor = session.execute(
|
|
sql, {"enterprise_suid": enterprise_suid, "node_id": enterprise_node_id}
|
|
)
|
|
return to_json(cursor)
|
|
|
|
def update_or_inert(self, data: Dict) -> None:
|
|
with get_session() as session:
|
|
sql = text(
|
|
"""
|
|
INSERT INTO `enterprise_alert` (`enterprise_suid`, `node_id`, `node_suid`, `is_sms`, `sms_to`, `is_email`, `email_to`, `freq`)
|
|
VALUES (:enterprise_suid, :node_id, :node_suid, :is_sms, :sms_to, :is_email, :email_to, :freq)
|
|
ON DUPLICATE KEY UPDATE
|
|
`node_suid`=:node_suid, `is_sms`=:is_sms, `sms_to`=:sms_to, `is_email`=:is_email, `email_to`=:email_to, `freq`=:freq
|
|
WHERE `enterprise_suid` = :enterprise_suid AND `node_id` = :node_id;
|
|
"""
|
|
)
|
|
session.execute(sql, data)
|
|
session.commit()
|