46 lines
2.0 KiB
Python
46 lines
2.0 KiB
Python
from typing import Type
|
|
|
|
from cpl_core.database.context import DatabaseContextABC
|
|
from cpl_core.dependency_injection import ServiceProviderABC
|
|
from cpl_core.logging import LoggerABC
|
|
|
|
from gismo_data.abc.migration_abc import MigrationABC
|
|
from gismo_data.model.migration_history import MigrationHistory
|
|
|
|
|
|
class MigrationService:
|
|
|
|
def __init__(self, logger: LoggerABC, services: ServiceProviderABC, db: DatabaseContextABC):
|
|
self._logger = logger
|
|
self._services = services
|
|
|
|
self._db = db
|
|
self._cursor = db.cursor
|
|
|
|
self._migrations: list[Type[MigrationABC]] = MigrationABC.__subclasses__()
|
|
|
|
def migrate(self):
|
|
self._logger.info(__name__, f"Running Migrations")
|
|
for migration in self._migrations:
|
|
migration_id = migration.__name__
|
|
try:
|
|
# check if table exists
|
|
self._cursor.execute("SHOW TABLES LIKE 'MigrationHistory'")
|
|
result = self._cursor.fetchone()
|
|
if result:
|
|
# there is a table named "tableName"
|
|
self._logger.trace(__name__, f"Running SQL Command: {MigrationHistory.get_select_by_id_string(migration_id)}")
|
|
migration_from_db = self._db.select(MigrationHistory.get_select_by_id_string(migration_id))
|
|
self._logger.trace(__name__, str(migration_from_db))
|
|
if migration_from_db is not None and len(migration_from_db) > 0:
|
|
continue
|
|
|
|
self._logger.debug(__name__, f"Running Migration {migration}")
|
|
migration_as_service: MigrationABC = self._services.get_service(migration)
|
|
migration_as_service.upgrade()
|
|
self._cursor.execute(MigrationHistory(migration_id).insert_string)
|
|
self._db.save_changes()
|
|
|
|
except Exception as e:
|
|
self._logger.error(__name__, f'Cannot get migration with id {migration}', e)
|