✨ 优化数据库迁移脚本

nonebot_plugin_tetris_stats/config/migrations/3c25a5a8c050_refactor_historical.py

@@ -47,6 +47,9 @@ def migrate_old_data() -> None:
             TimeRemainingColumn(),
         ) as progress,
     ):
+        if session.query(OldHistoricalData).count() == 0:
+            logger.info('空表, 跳过')
+            return
         task_id = progress.add_task('[cyan]Migrating:', total=session.query(OldHistoricalData).count())
         pointer = 0
         while pointer < session.query(OldHistoricalData).order_by(desc(OldHistoricalData.id)).limit(1).one().id:

nonebot_plugin_tetris_stats/config/migrations/8a91210ce14d_correct_the_data_in_historicaldata.py

@@ -28,12 +28,6 @@ depends_on: str | Sequence[str] | None = None
 def upgrade(name: str = '') -> None:  # noqa: C901
     if name:
         return
-    from nonebot_plugin_tetris_stats.version import __version__
-
-    if __version__ != '1.0.3':
-        msg = '本迁移需要1.0.3版本, 请先锁定版本至1.0.3版本再执行本迁移'
-        logger.critical(msg)
-        raise RuntimeError(msg)
 
     from nonebot.compat import PYDANTIC_V2, type_validate_json
     from pydantic import BaseModel, ValidationError
@@ -46,10 +40,6 @@ def upgrade(name: str = '') -> None:  # noqa: C901
         TimeRemainingColumn,
     )
 
-    from nonebot_plugin_tetris_stats.game_data_processor.schemas import (  # type: ignore[import-untyped]
-        BaseProcessedData,
-    )
-
     Base = automap_base()  # noqa: N806
     Base.prepare(autoload_with=op.get_bind())
     HistoricalData = Base.classes.nonebot_plugin_tetris_stats_historicaldata  # noqa: N806
@@ -62,6 +52,23 @@ def upgrade(name: str = '') -> None:  # noqa: C901
         def model_to_json(value: BaseModel) -> str:
             return value.json(by_alias=True)
 
+    with Session(op.get_bind()) as session:
+        count = session.query(HistoricalData).count()
+        if count == 0:
+            logger.info('空表, 跳过')
+            return
+
+        from nonebot_plugin_tetris_stats.version import __version__
+
+        if __version__ != '1.0.3':
+            msg = '本迁移需要1.0.3版本, 请先锁定版本至1.0.3版本再执行本迁移'
+            logger.critical(msg)
+            raise RuntimeError(msg)
+
+        from nonebot_plugin_tetris_stats.game_data_processor.schemas import (  # type: ignore[import-untyped]
+            BaseProcessedData,
+        )
+
         models = BaseProcessedData.__subclasses__()
 
         def json_to_model(value: str) -> BaseModel:
@@ -72,8 +79,6 @@ def upgrade(name: str = '') -> None:  # noqa: C901
                     ...
             raise ValueError
 
-    with Session(op.get_bind()) as session:
-        count = session.query(HistoricalData).count()
         with Progress(
             TextColumn('[progress.description]{task.description}'),
             BarColumn(),

nonebot_plugin_tetris_stats/config/migrations/b7fbdafc339a_add_user_unique_identifier_field_to_.py

@@ -26,12 +26,7 @@ depends_on: str | Sequence[str] | None = None
 def upgrade(name: str = '') -> None:
     if name:
         return
-    from nonebot_plugin_tetris_stats.version import __version__
 
-    if __version__ != '1.0.4':
-        msg = '本迁移需要1.0.4版本, 请先锁定版本至1.0.4版本再执行本迁移'
-        logger.critical(msg)
-        raise RuntimeError(msg)
     from nonebot.compat import type_validate_json
     from pydantic import ValidationError
     from rich.progress import (
@@ -46,8 +41,6 @@ def upgrade(name: str = '') -> None:
     from sqlalchemy.ext.automap import automap_base
     from sqlalchemy.orm import Session
 
-    from nonebot_plugin_tetris_stats.game_data_processor.schemas import BaseUser  # type: ignore[import-untyped]
-
     with op.batch_alter_table('nonebot_plugin_tetris_stats_historicaldata', schema=None) as batch_op:
         batch_op.add_column(sa.Column('user_unique_identifier', sa.String(length=32), nullable=True))
         batch_op.create_index(
@@ -60,6 +53,19 @@ def upgrade(name: str = '') -> None:
     Base.prepare(autoload_with=connection)
     HistoricalData = Base.classes.nonebot_plugin_tetris_stats_historicaldata  # noqa: N806
 
+    with Session(op.get_bind()) as session:
+        count = session.query(HistoricalData).count()
+        if count == 0:
+            logger.info('空表, 跳过')
+        else:
+            from nonebot_plugin_tetris_stats.version import __version__
+
+            if __version__ != '1.0.4':
+                msg = '本迁移需要1.0.4版本, 请先锁定版本至1.0.4版本再执行本迁移'
+                logger.critical(msg)
+                raise RuntimeError(msg)
+            from nonebot_plugin_tetris_stats.game_data_processor.schemas import BaseUser  # type: ignore[import-untyped]
+
             models: list[type[BaseUser]] = BaseUser.__subclasses__()
 
             def json_to_model(value: str) -> BaseUser:
@@ -70,8 +76,6 @@ def upgrade(name: str = '') -> None:
                         ...
                 raise ValueError
 
-    with Session(op.get_bind()) as session:
-        count = session.query(HistoricalData).count()
             with Progress(
                 TextColumn('[progress.description]{task.description}'),
                 BarColumn(),