Adds pylint to tox test pipeline and fixes related errors
[osm/MON.git] / osm_mon / core / database.py
1 # -*- coding: utf-8 -*-
2
3 # Copyright 2018 Whitestack, LLC
4 # *************************************************************
5
6 # This file is part of OSM Monitoring module
7 # All Rights Reserved to Whitestack, LLC
8
9 # Licensed under the Apache License, Version 2.0 (the "License"); you may
10 # not use this file except in compliance with the License. You may obtain
11 # a copy of the License at
12
13 # http://www.apache.org/licenses/LICENSE-2.0
14
15 # Unless required by applicable law or agreed to in writing, software
16 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
17 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
18 # License for the specific language governing permissions and limitations
19 # under the License.
20
21 # For those usages not covered by the Apache License, Version 2.0 please
22 # contact: bdiaz@whitestack.com or glavado@whitestack.com
23 ##
24
25 import json
26 import logging
27 import os
28 import uuid
29
30 from peewee import CharField, TextField, FloatField, Model, AutoField, Proxy
31 from peewee_migrate import Router
32 from playhouse.db_url import connect
33
34 from osm_mon import migrations
35 from osm_mon.core.config import Config
36
37 log = logging.getLogger(__name__)
38
39 db = Proxy()
40
41
42 class BaseModel(Model):
43 id = AutoField(primary_key=True)
44
45 class Meta:
46 database = db
47
48
49 class VimCredentials(BaseModel):
50 uuid = CharField(unique=True)
51 name = CharField()
52 type = CharField()
53 url = CharField()
54 user = CharField()
55 password = CharField()
56 tenant_name = CharField()
57 config = TextField()
58
59
60 class Alarm(BaseModel):
61 uuid = CharField(unique=True)
62 name = CharField()
63 severity = CharField()
64 threshold = FloatField()
65 operation = CharField()
66 statistic = CharField()
67 monitoring_param = CharField()
68 vdur_name = CharField()
69 vnf_member_index = CharField()
70 nsr_id = CharField()
71
72
73 class DatabaseManager:
74 def __init__(self, config: Config):
75 db.initialize(connect(config.get('sql', 'database_uri')))
76
77 def create_tables(self) -> None:
78 db.connect()
79 with db.atomic():
80 router = Router(db, os.path.dirname(migrations.__file__))
81 router.run()
82 db.close()
83
84 def get_credentials(self, vim_uuid: str = None) -> VimCredentials:
85 db.connect()
86 try:
87 with db.atomic():
88 vim_credentials = VimCredentials.get_or_none(VimCredentials.uuid == vim_uuid)
89 return vim_credentials
90 finally:
91 db.close()
92
93 def save_credentials(self, vim_credentials) -> VimCredentials:
94 """Saves vim credentials. If a record with same uuid exists, overwrite it."""
95 db.connect()
96 try:
97 with db.atomic():
98 exists = VimCredentials.get_or_none(VimCredentials.uuid == vim_credentials.uuid)
99 if exists:
100 vim_credentials.id = exists.id
101 vim_credentials.save()
102 return vim_credentials
103 finally:
104 db.close()
105
106 def get_alarm(self, alarm_id) -> Alarm:
107 db.connect()
108 try:
109 with db.atomic():
110 alarm = (Alarm.select()
111 .where(Alarm.alarm_id == alarm_id)
112 .get())
113 return alarm
114 finally:
115 db.close()
116
117 def save_alarm(self, name, threshold, operation, severity, statistic, metric_name, vdur_name,
118 vnf_member_index, nsr_id) -> Alarm:
119 """Saves alarm."""
120 # TODO: Add uuid optional param and check if exists to handle updates (see self.save_credentials)
121 db.connect()
122 try:
123 with db.atomic():
124 alarm = Alarm()
125 alarm.uuid = str(uuid.uuid4())
126 alarm.name = name
127 alarm.threshold = threshold
128 alarm.operation = operation
129 alarm.severity = severity
130 alarm.statistic = statistic
131 alarm.monitoring_param = metric_name
132 alarm.vdur_name = vdur_name
133 alarm.vnf_member_index = vnf_member_index
134 alarm.nsr_id = nsr_id
135 alarm.save()
136 return alarm
137 finally:
138 db.close()
139
140 def delete_alarm(self, alarm_uuid) -> None:
141 db.connect()
142 with db.atomic():
143 alarm = (Alarm.select()
144 .where(Alarm.uuid == alarm_uuid)
145 .get())
146 alarm.delete_instance()
147 db.close()
148
149 def get_vim_type(self, vim_account_id) -> str:
150 """Get the vim type that is required by the message."""
151 credentials = self.get_credentials(vim_account_id)
152 config = json.loads(credentials.config)
153 if 'vim_type' in config:
154 vim_type = config['vim_type']
155 return str(vim_type.lower())
156 else:
157 return str(credentials.type)