34 lines
1.1 KiB
Python
34 lines
1.1 KiB
Python
from cpl_core.database import DatabaseSettings
|
|
from cpl_core.database.context import DatabaseContext
|
|
from cpl_core.logging import LoggerABC
|
|
|
|
|
|
class DBContext(DatabaseContext):
|
|
|
|
def __init__(self, logger: LoggerABC):
|
|
|
|
self._logger = logger
|
|
|
|
DatabaseContext.__init__(self)
|
|
|
|
|
|
def connect(self, database_settings: DatabaseSettings):
|
|
try:
|
|
self._logger.debug(__name__, "Connecting to database")
|
|
self._db.connect(database_settings)
|
|
|
|
self.save_changes()
|
|
self._logger.info(__name__, "Connected to database")
|
|
except Exception as e:
|
|
self._logger.fatal(__name__, "Connecting to database failed", e)
|
|
|
|
def save_changes(self):
|
|
try:
|
|
self._logger.trace(__name__, "Save changes")
|
|
super(DBContext, self).save_changes()
|
|
self._logger.debug(__name__, "Saved changes")
|
|
except Exception as e:
|
|
self._logger.error(__name__, "Saving changes failed", e)
|
|
|
|
def select(self, statement: str) -> list[tuple]:
|
|
return super(DBContext, self).select(statement) |