SqliteTools.py

# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import sqlite3 as sl

class SQLite():
    def __init__(self, database):
        self.client = sl.connect(database, check_same_thread=False)
        self.client.execute("""
            CREATE TABLE IF NOT EXISTS KPI(
                kpi_id INTEGER PRIMARY KEY AUTOINCREMENT,
                kpi_description TEXT,
                kpi_sample_type INTEGER,
                device_id INTEGER,
                endpoint_id INTEGER,
                service_id INTEGER
            );
        """)

    def insert_KPI(self,kpi_description,kpi_sample_type,device_id,endpoint_id,service_id ):
        c = self.client.cursor()
        c.execute("SELECT kpi_id FROM KPI WHERE device_id is ? AND kpi_sample_type is ? AND endpoint_id is ?",(device_id,kpi_sample_type,endpoint_id))
        data=c.fetchone()
        if data is None:
            c.execute("INSERT INTO KPI (kpi_description,kpi_sample_type,device_id,endpoint_id,service_id) VALUES (?,?,?,?,?)", (kpi_description,kpi_sample_type,device_id,endpoint_id,service_id))
            self.client.commit()
            return c.lastrowid
        else:
            return data[0]

    def delete_KPI(self,device_id,kpi_sample_type):
        c = self.client.cursor()
        c.execute("SELECT kpi_id FROM KPI WHERE device_id is ? AND kpi_sample_type is ?",(device_id,kpi_sample_type))       
        data=c.fetchone()
        if data is None:
            return False
        else:
            c.execute("DELETE FROM KPI WHERE device_id is ? AND kpi_sample_type is ?",(device_id,kpi_sample_type))
            self.client.commit()
            return True

    def delete_kpid_id(self,kpi_id):
        c = self.client.cursor()
        c.execute("SELECT * FROM KPI WHERE kpi_id is ?",(kpi_id,))       
        data=c.fetchone()
        if data is None:
            return False
        else:
            c.execute("DELETE FROM KPI WHERE kpi_id is ?",(kpi_id,))
            self.client.commit()
            return True

    def get_KPI(self,kpi_id):
        data = self.client.execute("SELECT * FROM KPI WHERE kpi_id is ?",(kpi_id,))
        return data.fetchone()
        
    def get_KPIS(self):
        data = self.client.execute("SELECT * FROM KPI")
        #print("\n")
        #for row in data:
        #    print(row)
        return data.fetchall()