diff --git a/src/telemetry/database/TelemetryDB.py b/src/telemetry/database/TelemetryDB.py
new file mode 100644
index 0000000000000000000000000000000000000000..b5b0c4c7e31f0616c91503c6f4e1dbd7abead5c9
--- /dev/null
+++ b/src/telemetry/database/TelemetryDB.py
@@ -0,0 +1,137 @@
+# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
+#
+# 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 logging
+import sqlalchemy_utils
+from sqlalchemy.orm import sessionmaker
+from telemetry.database.TelemetryModel import Collector as CollectorModel
+from telemetry.database.TelemetryEngine import TelemetryEngine
+from common.method_wrappers.ServiceExceptions import (
+ OperationFailedException, AlreadyExistsException )
+
+LOGGER = logging.getLogger(__name__)
+DB_NAME = "tfs_telemetry"
+
+class TelemetryDBmanager:
+ def __init__(self):
+ self.db_engine = TelemetryEngine.get_engine()
+ if self.db_engine is None:
+ LOGGER.error('Unable to get SQLAlchemy DB Engine...')
+ return False
+ self.db_name = DB_NAME
+ self.Session = sessionmaker(bind=self.db_engine)
+
+ def create_database(self):
+ if not sqlalchemy_utils.database_exists(self.db_engine.url):
+ LOGGER.debug("Database created. {:}".format(self.db_engine.url))
+ sqlalchemy_utils.create_database(self.db_engine.url)
+
+ def drop_database(self) -> None:
+ if sqlalchemy_utils.database_exists(self.db_engine.url):
+ sqlalchemy_utils.drop_database(self.db_engine.url)
+
+ def create_tables(self):
+ try:
+ CollectorModel.metadata.create_all(self.db_engine) # type: ignore
+ LOGGER.debug("Tables created in the database: {:}".format(self.db_name))
+ except Exception as e:
+ LOGGER.debug("Tables cannot be created in the database. {:s}".format(str(e)))
+ raise OperationFailedException ("Tables can't be created", extra_details=["unable to create table {:}".format(e)])
+
+ def verify_tables(self):
+ try:
+ with self.db_engine.connect() as connection:
+ result = connection.execute("SHOW TABLES;")
+ tables = result.fetchall()
+ LOGGER.info("Tables in DB: {:}".format(tables))
+ except Exception as e:
+ LOGGER.info("Unable to fetch Table names. {:s}".format(str(e)))
+
+# ----------------- CURD METHODs ---------------------
+
+ def add_row_to_db(self, row):
+ session = self.Session()
+ try:
+ session.add(row)
+ session.commit()
+ LOGGER.debug(f"Row inserted into {row.__class__.__name__} table.")
+ return True
+ except Exception as e:
+ session.rollback()
+ if "psycopg2.errors.UniqueViolation" in str(e):
+ LOGGER.error(f"Unique key voilation: {row.__class__.__name__} table. {str(e)}")
+ raise AlreadyExistsException(row.__class__.__name__, row,
+ extra_details=["Unique key voilation: {:}".format(e)] )
+ else:
+ LOGGER.error(f"Failed to insert new row into {row.__class__.__name__} table. {str(e)}")
+ raise OperationFailedException ("Deletion by column id", extra_details=["unable to delete row {:}".format(e)])
+ finally:
+ session.close()
+
+ def search_db_row_by_id(self, model, col_name, id_to_search):
+ session = self.Session()
+ try:
+ entity = session.query(model).filter_by(**{col_name: id_to_search}).first()
+ if entity:
+ # LOGGER.debug(f"{model.__name__} ID found: {str(entity)}")
+ return entity
+ else:
+ LOGGER.debug(f"{model.__name__} ID not found, No matching row: {str(id_to_search)}")
+ print("{:} ID not found, No matching row: {:}".format(model.__name__, id_to_search))
+ return None
+ except Exception as e:
+ session.rollback()
+ LOGGER.debug(f"Failed to retrieve {model.__name__} ID. {str(e)}")
+ raise OperationFailedException ("search by column id", extra_details=["unable to search row {:}".format(e)])
+ finally:
+ session.close()
+
+ def delete_db_row_by_id(self, model, col_name, id_to_search):
+ session = self.Session()
+ try:
+ record = session.query(model).filter_by(**{col_name: id_to_search}).first()
+ if record:
+ session.delete(record)
+ session.commit()
+ LOGGER.debug("Deleted %s with %s: %s", model.__name__, col_name, id_to_search)
+ else:
+ LOGGER.debug("%s with %s %s not found", model.__name__, col_name, id_to_search)
+ return None
+ except Exception as e:
+ session.rollback()
+ LOGGER.error("Error deleting %s with %s %s: %s", model.__name__, col_name, id_to_search, e)
+ raise OperationFailedException ("Deletion by column id", extra_details=["unable to delete row {:}".format(e)])
+ finally:
+ session.close()
+
+ def select_with_filter(self, model, filter_object):
+ session = self.Session()
+ try:
+ query = session.query(CollectorModel)
+ # Apply filters based on the filter_object
+ if filter_object.kpi_id:
+ query = query.filter(CollectorModel.kpi_id.in_([k.kpi_id.uuid for k in filter_object.kpi_id]))
+ result = query.all()
+
+ if result:
+ LOGGER.debug(f"Fetched filtered rows from {model.__name__} table with filters: {filter_object}") # - Results: {result}
+ else:
+ LOGGER.debug(f"No matching row found in {model.__name__} table with filters: {filter_object}")
+ return result
+ except Exception as e:
+ LOGGER.error(f"Error fetching filtered rows from {model.__name__} table with filters {filter_object} ::: {e}")
+ raise OperationFailedException ("Select by filter", extra_details=["unable to apply the filter {:}".format(e)])
+ finally:
+ session.close()
+
diff --git a/src/telemetry/database/TelemetryDBmanager.py b/src/telemetry/database/TelemetryDBmanager.py
deleted file mode 100644
index b558180a9e1fbf85bf523c7faededf58f57e2264..0000000000000000000000000000000000000000
--- a/src/telemetry/database/TelemetryDBmanager.py
+++ /dev/null
@@ -1,248 +0,0 @@
-# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
-#
-# 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 logging, time
-import sqlalchemy
-from sqlalchemy import inspect, MetaData, Table
-from sqlalchemy.orm import sessionmaker
-from telemetry.database.TelemetryModel import Collector as CollectorModel
-from telemetry.database.TelemetryModel import Kpi as KpiModel
-from sqlalchemy.ext.declarative import declarative_base
-from telemetry.database.TelemetryEngine import TelemetryEngine
-from common.proto.kpi_manager_pb2 import KpiDescriptor, KpiId
-from common.proto.telemetry_frontend_pb2 import Collector, CollectorId
-from sqlalchemy.exc import SQLAlchemyError
-from telemetry.database.TelemetryModel import Base
-
-LOGGER = logging.getLogger(__name__)
-DB_NAME = "telemetryfrontend"
-
-class TelemetryDBmanager:
- def __init__(self):
- self.db_engine = TelemetryEngine.get_engine()
- if self.db_engine is None:
- LOGGER.error('Unable to get SQLAlchemy DB Engine...')
- return False
- self.db_name = DB_NAME
- self.Session = sessionmaker(bind=self.db_engine)
-
- def create_database(self):
- try:
- # with self.db_engine.connect() as connection:
- # connection.execute(f"CREATE DATABASE {self.db_name};")
- TelemetryEngine.create_database(self.db_engine)
- LOGGER.info('TelemetryDBmanager initalized DB Name: {:}'.format(self.db_name))
- return True
- except Exception as e: # pylint: disable=bare-except # pragma: no cover
- LOGGER.exception('Failed to check/create the database: {:s}'.format(str(e)))
- return False
-
- def create_tables(self):
- try:
- Base.metadata.create_all(self.db_engine) # type: ignore
- LOGGER.info("Tables created in database ({:}) the as per Models".format(self.db_name))
- except Exception as e:
- LOGGER.info("Tables cannot be created in the TelemetryFrontend database. {:s}".format(str(e)))
-
- def verify_tables(self):
- try:
- with self.db_engine.connect() as connection:
- result = connection.execute("SHOW TABLES;")
- tables = result.fetchall()
- LOGGER.info("Tables in DB: {:}".format(tables))
- except Exception as e:
- LOGGER.info("Unable to fetch Table names. {:s}".format(str(e)))
-
- def drop_table(self, table_to_drop: str):
- try:
- inspector = inspect(self.db_engine)
- existing_tables = inspector.get_table_names()
- if table_to_drop in existing_tables:
- table = Table(table_to_drop, MetaData(), autoload_with=self.db_engine)
- table.drop(self.db_engine)
- LOGGER.info("Tables delete in the DB Name: {:}".format(self.db_name))
- else:
- LOGGER.warning("No table {:} in database {:} ".format(table_to_drop, DB_NAME))
- except Exception as e:
- LOGGER.info("Tables cannot be deleted in the {:} database. {:s}".format(DB_NAME, str(e)))
-
- def list_databases(self):
- query = "SHOW DATABASES"
- with self.db_engine.connect() as connection:
- result = connection.execute(query)
- databases = [row[0] for row in result]
- LOGGER.info("List of available DBs: {:}".format(databases))
-
-# ------------------ INSERT METHODs --------------------------------------
-
- def inser_kpi(self, request: KpiDescriptor):
- session = self.Session()
- try:
- # Create a new Kpi instance
- kpi_to_insert = KpiModel()
- kpi_to_insert.kpi_id = request.kpi_id.kpi_id.uuid
- kpi_to_insert.kpi_description = request.kpi_description
- kpi_to_insert.kpi_sample_type = request.kpi_sample_type
- kpi_to_insert.device_id = request.service_id.service_uuid.uuid
- kpi_to_insert.endpoint_id = request.device_id.device_uuid.uuid
- kpi_to_insert.service_id = request.slice_id.slice_uuid.uuid
- kpi_to_insert.slice_id = request.endpoint_id.endpoint_uuid.uuid
- kpi_to_insert.connection_id = request.connection_id.connection_uuid.uuid
- # kpi_to_insert.link_id = request.link_id.link_id.uuid
- # Add the instance to the session
- session.add(kpi_to_insert)
- session.commit()
- LOGGER.info("Row inserted into kpi table: {:}".format(kpi_to_insert.kpi_id))
- except Exception as e:
- session.rollback()
- LOGGER.info("Failed to insert new kpi. {:s}".format(str(e)))
- finally:
- # Close the session
- session.close()
-
- # Function to insert a row into the Collector model
- def insert_collector(self, request: Collector):
- session = self.Session()
- try:
- # Create a new Collector instance
- collector_to_insert = CollectorModel()
- collector_to_insert.collector_id = request.collector_id.collector_id.uuid
- collector_to_insert.kpi_id = request.kpi_id.kpi_id.uuid
- collector_to_insert.collector = "Test collector description"
- collector_to_insert.sampling_duration_s = request.duration_s
- collector_to_insert.sampling_interval_s = request.interval_s
- collector_to_insert.start_timestamp = time.time()
- collector_to_insert.end_timestamp = time.time()
-
- session.add(collector_to_insert)
- session.commit()
- LOGGER.info("Row inserted into collector table: {:}".format(collector_to_insert.collector_id))
- except Exception as e:
- session.rollback()
- LOGGER.info("Failed to insert new collector. {:s}".format(str(e)))
- finally:
- # Close the session
- session.close()
-
-# ------------------ GET METHODs --------------------------------------
-
- def get_kpi_descriptor(self, request: KpiId):
- session = self.Session()
- try:
- kpi_id_to_search = request.kpi_id.uuid
- kpi = session.query(KpiModel).filter_by(kpi_id=kpi_id_to_search).first()
- if kpi:
- LOGGER.info("kpi ID found: {:s}".format(str(kpi)))
- return kpi
- else:
- LOGGER.warning("Kpi ID not found {:s}".format(str(kpi_id_to_search)))
- return None
- except Exception as e:
- session.rollback()
- LOGGER.info("Failed to retrieve KPI ID. {:s}".format(str(e)))
- raise
- finally:
- session.close()
-
- def get_collector(self, request: CollectorId):
- session = self.Session()
- try:
- collector_id_to_search = request.collector_id.uuid
- collector = session.query(CollectorModel).filter_by(collector_id=collector_id_to_search).first()
- if collector:
- LOGGER.info("collector ID found: {:s}".format(str(collector)))
- return collector
- else:
- LOGGER.warning("collector ID not found{:s}".format(str(collector_id_to_search)))
- return None
- except Exception as e:
- session.rollback()
- LOGGER.info("Failed to retrieve collector ID. {:s}".format(str(e)))
- raise
- finally:
- session.close()
-
- # ------------------ SELECT METHODs --------------------------------------
-
- def select_kpi_descriptor(self, **filters):
- session = self.Session()
- try:
- query = session.query(KpiModel)
- for column, value in filters.items():
- query = query.filter(getattr(KpiModel, column) == value)
- result = query.all()
- if len(result) != 0:
- LOGGER.info("Fetched filtered rows from KPI table with filters : {:s}".format(str(result)))
- else:
- LOGGER.warning("No matching row found : {:s}".format(str(result)))
- return result
- except SQLAlchemyError as e:
- LOGGER.error("Error fetching filtered rows from KPI table with filters {:}: {:}".format(filters, e))
- return []
- finally:
- session.close()
-
- def select_collector(self, **filters):
- session = self.Session()
- try:
- query = session.query(CollectorModel)
- for column, value in filters.items():
- query = query.filter(getattr(CollectorModel, column) == value)
- result = query.all()
- if len(result) != 0:
- LOGGER.info("Fetched filtered rows from KPI table with filters : {:s}".format(str(result)))
- else:
- LOGGER.warning("No matching row found : {:s}".format(str(result)))
- return result
- except SQLAlchemyError as e:
- LOGGER.error("Error fetching filtered rows from KPI table with filters {:}: {:}".format(filters, e))
- return []
- finally:
- session.close()
-
-# ------------------ DELETE METHODs --------------------------------------
-
- def delete_kpi_descriptor(self, request: KpiId):
- session = self.Session()
- try:
- kpi_id_to_delete = request.kpi_id.uuid
- kpi = session.query(KpiModel).filter_by(kpi_id=kpi_id_to_delete).first()
- if kpi:
- session.delete(kpi)
- session.commit()
- LOGGER.info("Deleted KPI with kpi_id: %s", kpi_id_to_delete)
- else:
- LOGGER.warning("KPI with kpi_id %s not found", kpi_id_to_delete)
- except SQLAlchemyError as e:
- session.rollback()
- LOGGER.error("Error deleting KPI with kpi_id %s: %s", kpi_id_to_delete, e)
- finally:
- session.close()
-
- def delete_collector(self, request: CollectorId):
- session = self.Session()
- try:
- collector_id_to_delete = request.collector_id.uuid
- collector = session.query(CollectorModel).filter_by(collector_id=collector_id_to_delete).first()
- if collector:
- session.delete(collector)
- session.commit()
- LOGGER.info("Deleted collector with collector_id: %s", collector_id_to_delete)
- else:
- LOGGER.warning("collector with collector_id %s not found", collector_id_to_delete)
- except SQLAlchemyError as e:
- session.rollback()
- LOGGER.error("Error deleting collector with collector_id %s: %s", collector_id_to_delete, e)
- finally:
- session.close()
\ No newline at end of file
diff --git a/src/telemetry/database/TelemetryEngine.py b/src/telemetry/database/TelemetryEngine.py
index a563fa09f94c812aed07d0aa3cbd5bc988737fc4..bd7cda5996779858b8819b7f67a223803c2933c1 100644
--- a/src/telemetry/database/TelemetryEngine.py
+++ b/src/telemetry/database/TelemetryEngine.py
@@ -12,34 +12,28 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import logging, sqlalchemy, sqlalchemy_utils
-# from common.Settings import get_setting
+import logging, sqlalchemy
+from common.Settings import get_setting
LOGGER = logging.getLogger(__name__)
-APP_NAME = 'tfs'
-ECHO = False # False: No dump SQL commands and transactions executed
CRDB_URI_TEMPLATE = 'cockroachdb://{:s}:{:s}@127.0.0.1:{:s}/{:s}?sslmode={:s}'
# CRDB_URI_TEMPLATE = 'cockroachdb://{:s}:{:s}@cockroachdb-public.{:s}.svc.cluster.local:{:s}/{:s}?sslmode={:s}'
class TelemetryEngine:
- # def __init__(self):
- # self.engine = self.get_engine()
@staticmethod
def get_engine() -> sqlalchemy.engine.Engine:
- CRDB_NAMESPACE = "crdb"
- CRDB_SQL_PORT = "26257"
- CRDB_DATABASE = "telemetryfrontend"
- CRDB_USERNAME = "tfs"
- CRDB_PASSWORD = "tfs123"
- CRDB_SSLMODE = "require"
- crdb_uri = CRDB_URI_TEMPLATE.format(
- CRDB_USERNAME, CRDB_PASSWORD, CRDB_SQL_PORT, CRDB_DATABASE, CRDB_SSLMODE)
- # crdb_uri = CRDB_URI_TEMPLATE.format(
- # CRDB_USERNAME, CRDB_PASSWORD, CRDB_NAMESPACE, CRDB_SQL_PORT, CRDB_DATABASE, CRDB_SSLMODE)
+ crdb_uri = get_setting('CRDB_URI', default=None)
+ if crdb_uri is None:
+ CRDB_NAMESPACE = "crdb"
+ CRDB_SQL_PORT = "26257"
+ CRDB_DATABASE = "telemetryfrontend"
+ CRDB_USERNAME = "tfs"
+ CRDB_PASSWORD = "tfs123"
+ CRDB_SSLMODE = "require"
+ crdb_uri = CRDB_URI_TEMPLATE.format(
+ CRDB_USERNAME, CRDB_PASSWORD, CRDB_SQL_PORT, CRDB_DATABASE, CRDB_SSLMODE)
try:
- # engine = sqlalchemy.create_engine(
- # crdb_uri, connect_args={'application_name': APP_NAME}, echo=ECHO, future=True)
engine = sqlalchemy.create_engine(crdb_uri, echo=False)
LOGGER.info(' TelemetryDBmanager initalized with DB URL: {:}'.format(crdb_uri))
except: # pylint: disable=bare-except # pragma: no cover
@@ -47,13 +41,4 @@ class TelemetryEngine:
return None # type: ignore
return engine # type: ignore
- @staticmethod
- def create_database(engine : sqlalchemy.engine.Engine) -> None:
- if not sqlalchemy_utils.database_exists(engine.url):
- LOGGER.info("Database created. {:}".format(engine.url))
- sqlalchemy_utils.create_database(engine.url)
- @staticmethod
- def drop_database(engine : sqlalchemy.engine.Engine) -> None:
- if sqlalchemy_utils.database_exists(engine.url):
- sqlalchemy_utils.drop_database(engine.url)
diff --git a/src/telemetry/database/TelemetryModel.py b/src/telemetry/database/TelemetryModel.py
index be4f0969c86638520cf226b8e42db90426165804..95f692e4b80d5a3a1a453315c0f56c13abb2fc60 100644
--- a/src/telemetry/database/TelemetryModel.py
+++ b/src/telemetry/database/TelemetryModel.py
@@ -14,9 +14,7 @@
import logging
from sqlalchemy.dialects.postgresql import UUID
-from sqlalchemy import Column, Integer, String, Float, Text, ForeignKey
-from sqlalchemy.ext.declarative import declarative_base
-from sqlalchemy.orm import sessionmaker, relationship
+from sqlalchemy import Column, String, Float
from sqlalchemy.orm import registry
logging.basicConfig(level=logging.INFO)
@@ -24,22 +22,23 @@ LOGGER = logging.getLogger(__name__)
# Create a base class for declarative models
Base = registry().generate_base()
-# Base = declarative_base()
class Collector(Base):
__tablename__ = 'collector'
collector_id = Column(UUID(as_uuid=False), primary_key=True)
- kpi_id = Column(UUID(as_uuid=False))
- collector_decription = Column(String)
- sampling_duration_s = Column(Float)
- sampling_interval_s = Column(Float)
- start_timestamp = Column(Float)
- end_timestamp = Column(Float)
-
+ kpi_id = Column(UUID(as_uuid=False), nullable=False)
+ collector_decription = Column(String , nullable=False)
+ sampling_duration_s = Column(Float , nullable=False)
+ sampling_interval_s = Column(Float , nullable=False)
+ start_timestamp = Column(Float , nullable=False)
+ end_timestamp = Column(Float , nullable=False)
+ # helps in logging the information
def __repr__(self):
return (f"<Collector(collector_id='{self.collector_id}', kpi_id='{self.kpi_id}', "
f"collector='{self.collector_decription}', sampling_duration_s='{self.sampling_duration_s}', "
f"sampling_interval_s='{self.sampling_interval_s}', start_timestamp='{self.start_timestamp}', "
- f"end_timestamp='{self.end_timestamp}')>")
\ No newline at end of file
+ f"end_timestamp='{self.end_timestamp}')>")
+
+# add method to convert gRPC requests to rows if necessary...
diff --git a/src/telemetry/database/managementDB.py b/src/telemetry/database/managementDB.py
deleted file mode 100644
index f79126f279d7bbece6c08ae5eb1cd74e340d1c7d..0000000000000000000000000000000000000000
--- a/src/telemetry/database/managementDB.py
+++ /dev/null
@@ -1,138 +0,0 @@
-# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
-#
-# 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 logging, time
-import sqlalchemy
-import sqlalchemy_utils
-from sqlalchemy.orm import sessionmaker
-from sqlalchemy.ext.declarative import declarative_base
-from telemetry.database.TelemetryEngine import TelemetryEngine
-from telemetry.database.TelemetryModel import Base
-
-LOGGER = logging.getLogger(__name__)
-DB_NAME = "telemetryfrontend"
-
-# # Create a base class for declarative models
-# Base = declarative_base()
-
-class managementDB:
- def __init__(self):
- self.db_engine = TelemetryEngine.get_engine()
- if self.db_engine is None:
- LOGGER.error('Unable to get SQLAlchemy DB Engine...')
- return False
- self.db_name = DB_NAME
- self.Session = sessionmaker(bind=self.db_engine)
-
- @staticmethod
- def create_database(engine : sqlalchemy.engine.Engine) -> None:
- if not sqlalchemy_utils.database_exists(engine.url):
- LOGGER.info("Database created. {:}".format(engine.url))
- sqlalchemy_utils.create_database(engine.url)
-
- @staticmethod
- def drop_database(engine : sqlalchemy.engine.Engine) -> None:
- if sqlalchemy_utils.database_exists(engine.url):
- sqlalchemy_utils.drop_database(engine.url)
-
- # def create_database(self):
- # try:
- # with self.db_engine.connect() as connection:
- # connection.execute(f"CREATE DATABASE {self.db_name};")
- # LOGGER.info('managementDB initalizes database. Name: {self.db_name}')
- # return True
- # except:
- # LOGGER.exception('Failed to check/create the database: {:s}'.format(str(self.db_engine.url)))
- # return False
-
- @staticmethod
- def create_tables(engine : sqlalchemy.engine.Engine):
- try:
- Base.metadata.create_all(engine) # type: ignore
- LOGGER.info("Tables created in the DB Name: {:}".format(DB_NAME))
- except Exception as e:
- LOGGER.info("Tables cannot be created in the TelemetryFrontend database. {:s}".format(str(e)))
-
- def verify_tables(self):
- try:
- with self.db_engine.connect() as connection:
- result = connection.execute("SHOW TABLES;")
- tables = result.fetchall() # type: ignore
- LOGGER.info("Tables verified: {:}".format(tables))
- except Exception as e:
- LOGGER.info("Unable to fetch Table names. {:s}".format(str(e)))
-
- @staticmethod
- def add_row_to_db(self, row):
- session = self.Session()
- try:
- session.add(row)
- session.commit()
- LOGGER.info(f"Row inserted into {row.__class__.__name__} table.")
- except Exception as e:
- session.rollback()
- LOGGER.error(f"Failed to insert new row into {row.__class__.__name__} table. {str(e)}")
- finally:
- session.close()
-
- def search_db_row_by_id(self, model, col_name, id_to_search):
- session = self.Session()
- try:
- entity = session.query(model).filter_by(**{col_name: id_to_search}).first()
- if entity:
- LOGGER.info(f"{model.__name__} ID found: {str(entity)}")
- return entity
- else:
- LOGGER.warning(f"{model.__name__} ID not found: {str(id_to_search)}")
- return None
- except Exception as e:
- session.rollback()
- LOGGER.info(f"Failed to retrieve {model.__name__} ID. {str(e)}")
- raise
- finally:
- session.close()
-
- def delete_db_row_by_id(self, model, col_name, id_to_search):
- session = self.Session()
- try:
- record = session.query(model).filter_by(**{col_name: id_to_search}).first()
- if record:
- session.delete(record)
- session.commit()
- LOGGER.info("Deleted %s with %s: %s", model.__name__, col_name, id_to_search)
- else:
- LOGGER.warning("%s with %s %s not found", model.__name__, col_name, id_to_search)
- except Exception as e:
- session.rollback()
- LOGGER.error("Error deleting %s with %s %s: %s", model.__name__, col_name, id_to_search, e)
- finally:
- session.close()
-
- def select_with_filter(self, model, **filters):
- session = self.Session()
- try:
- query = session.query(model)
- for column, value in filters.items():
- query = query.filter(getattr(model, column) == value) # type: ignore
- result = query.all()
- if result:
- LOGGER.info(f"Fetched filtered rows from {model.__name__} table with filters: {filters}") # - Results: {result}
- else:
- LOGGER.warning(f"No matching row found in {model.__name__} table with filters: {filters}")
- return result
- except Exception as e:
- LOGGER.error(f"Error fetching filtered rows from {model.__name__} table with filters {filters} ::: {e}")
- return []
- finally:
- session.close()
\ No newline at end of file