|
6 | 6 |
|
7 | 7 | from alembic import context |
8 | 8 | from sqlalchemy.util import await_fallback |
9 | | -from alembic.migration import MigrationContext |
10 | | -from alembic.operations import MigrateOperation |
11 | 9 | from alembic.operations.ops import MigrationScript |
12 | 10 | from sqlalchemy.ext.asyncio import AsyncEngine, AsyncConnection |
13 | 11 | from sqlalchemy import MetaData, Connection, TwoPhaseTransaction |
@@ -68,19 +66,12 @@ def run_migrations_offline() -> None: |
68 | 66 | config.print_stdout(f"将输出写入到 {file_}") |
69 | 67 |
|
70 | 68 |
|
71 | | -def process_revision_directives( |
72 | | - context: MigrationContext, |
73 | | - revision: tuple[str, str], |
74 | | - directives: list[MigrateOperation], |
75 | | -) -> None: |
| 69 | +def process_revision_directives(_, __, directives: list[MigrationScript]) -> None: |
76 | 70 | # 此回调用于防止在模型没有更改时生成自动迁移。 |
77 | 71 | # 参见:https://alembic.sqlalchemy.org/en/latest/cookbook.html#don-t-generate-empty-migrations-with-autogenerate |
78 | 72 |
|
79 | 73 | if getattr(config.cmd_opts, "autogenerate", False) and all( |
80 | | - filter( |
81 | | - methodcaller("is_empty"), |
82 | | - cast(MigrationScript, directives[0]).upgrade_ops_list, |
83 | | - ) |
| 74 | + filter(methodcaller("is_empty"), directives[0].upgrade_ops_list) |
84 | 75 | ): |
85 | 76 | directives[:] = [] |
86 | 77 | config.print_stdout("未检测到模型更改") |
|
0 commit comments