-from peewee import CharField, TextField, FloatField, Model, AutoField
+from peewee import CharField, TextField, FloatField, Model, AutoField, Proxy
from playhouse.db_url import connect
from osm_mon.core.settings import Config
log = logging.getLogger(__name__)
from playhouse.db_url import connect
from osm_mon.core.settings import Config
log = logging.getLogger(__name__)
- def get_credentials(self, vim_uuid) -> VimCredentials:
- return VimCredentials.get_or_none(VimCredentials.uuid == vim_uuid)
+ def get_credentials(self, vim_uuid: str = None) -> VimCredentials:
+ with db.atomic():
+ return VimCredentials.get_or_none(VimCredentials.uuid == vim_uuid)
def save_credentials(self, vim_credentials) -> VimCredentials:
"""Saves vim credentials. If a record with same uuid exists, overwrite it."""
def save_credentials(self, vim_credentials) -> VimCredentials:
"""Saves vim credentials. If a record with same uuid exists, overwrite it."""
- exists = VimCredentials.get_or_none(VimCredentials.uuid == vim_credentials.uuid)
- if exists:
- vim_credentials.id = exists.id
- vim_credentials.save()
- return vim_credentials
+ with db.atomic():
+ exists = VimCredentials.get_or_none(VimCredentials.uuid == vim_credentials.uuid)
+ if exists:
+ vim_credentials.id = exists.id
+ vim_credentials.save()
+ return vim_credentials
def save_alarm(self, name, threshold, operation, severity, statistic, metric_name, vdur_name,
vnf_member_index, nsr_id) -> Alarm:
"""Saves alarm."""
# TODO: Add uuid optional param and check if exists to handle updates (see self.save_credentials)
def save_alarm(self, name, threshold, operation, severity, statistic, metric_name, vdur_name,
vnf_member_index, nsr_id) -> Alarm:
"""Saves alarm."""
# TODO: Add uuid optional param and check if exists to handle updates (see self.save_credentials)
- alarm = Alarm()
- alarm.uuid = str(uuid.uuid4())
- alarm.name = name
- alarm.threshold = threshold
- alarm.operation = operation
- alarm.severity = severity
- alarm.statistic = statistic
- alarm.monitoring_param = metric_name
- alarm.vdur_name = vdur_name
- alarm.vnf_member_index = vnf_member_index
- alarm.nsr_id = nsr_id
- alarm.save()
- return alarm
+ with db.atomic():
+ alarm = Alarm()
+ alarm.uuid = str(uuid.uuid4())
+ alarm.name = name
+ alarm.threshold = threshold
+ alarm.operation = operation
+ alarm.severity = severity
+ alarm.statistic = statistic
+ alarm.monitoring_param = metric_name
+ alarm.vdur_name = vdur_name
+ alarm.vnf_member_index = vnf_member_index
+ alarm.nsr_id = nsr_id
+ alarm.save()
+ return alarm
- alarm = (Alarm.select()
- .where(Alarm.uuid == alarm_uuid)
- .get())
- alarm.delete_instance()
+ with db.atomic():
+ alarm = (Alarm.select()
+ .where(Alarm.uuid == alarm_uuid)
+ .get())
+ alarm.delete_instance()
def get_vim_type(self, vim_account_id) -> str:
"""Get the vim type that is required by the message."""
def get_vim_type(self, vim_account_id) -> str:
"""Get the vim type that is required by the message."""