56 lines
2.0 KiB
Python
56 lines
2.0 KiB
Python
from cpl_cli.abc.generate_schematic_abc import GenerateSchematicABC
|
|
|
|
|
|
class Migration(GenerateSchematicABC):
|
|
def __init__(self, *args: str):
|
|
GenerateSchematicABC.__init__(self, *args)
|
|
|
|
def get_code(self) -> str:
|
|
import textwrap
|
|
|
|
code = textwrap.dedent(
|
|
"""\
|
|
from bot_core.logging.database_logger import DatabaseLogger
|
|
from bot_data.abc.migration_abc import MigrationABC
|
|
from bot_data.db_context import DBContext
|
|
|
|
|
|
class $ClassName(MigrationABC):
|
|
name = "1.0_$ClassName"
|
|
|
|
def __init__(self, logger: DatabaseLogger, db: DBContext):
|
|
MigrationABC.__init__(self)
|
|
self._logger = logger
|
|
self._db = db
|
|
self._cursor = db.cursor
|
|
|
|
def upgrade(self):
|
|
self._logger.debug(__name__, "Running upgrade")
|
|
|
|
self._cursor.execute(
|
|
str(
|
|
f\"""
|
|
CREATE TABLE IF NOT EXISTS `$TableName` (
|
|
`Id` BIGINT NOT NULL AUTO_INCREMENT,
|
|
`CreatedAt` DATETIME(6),
|
|
`LastModifiedAt` DATETIME(6),
|
|
PRIMARY KEY(`Id`)
|
|
);
|
|
\"""
|
|
)
|
|
)
|
|
|
|
def downgrade(self):
|
|
self._cursor.execute("DROP TABLE `$TableName`;")
|
|
"""
|
|
)
|
|
return self.build_code_str(
|
|
code,
|
|
ClassName=self._class_name,
|
|
TableName=self._class_name.split("Migration")[0],
|
|
)
|
|
|
|
@classmethod
|
|
def register(cls):
|
|
GenerateSchematicABC.register(cls, "migration", [])
|