feat: support skip table migration by set managed=False (#397)

.github/workflows/ci.yml

@@ -26,6 +26,8 @@ jobs:
           - tortoise022
           - tortoise023
           - tortoise024
+          # TODO: add dev back when drop python3.8 support
+          # - tortoisedev
     steps:
       - name: Start MySQL
         run: sudo systemctl start mysql.service

CHANGELOG.md

@@ -7,6 +7,7 @@
 #### Added
 - feat: support command `python -m aerich`. ([#417])
 - feat: add --fake to upgrade/downgrade. ([#398])
+- Support ignore table by settings `managed=False` in `Meta` class. ([#397])
 
 #### Fixed
 - fix: aerich migrate raises tortoise.exceptions.FieldError when `index.INDEX_TYPE` is not empty. ([#415])
@@ -17,6 +18,7 @@
 ### Changed
 - Refactored version management to use `importlib.metadata.version(__package__)` instead of hardcoded version string ([#412])
 
+[#397]: https://github.com/tortoise/aerich/pull/397
 [#398]: https://github.com/tortoise/aerich/pull/398
 [#401]: https://github.com/tortoise/aerich/pull/401
 [#404]: https://github.com/tortoise/aerich/pull/404

README.md

@@ -295,6 +295,16 @@ aerich downgrade --fake -v 2
 aerich --app models downgrade --fake -v 2
 ```
 
+### Ignore tables
+
+You can tell aerich to ignore table by setting `managed=False` in the `Meta` class, e.g.:
+```py
+class MyModel(Model):
+    class Meta:
+        managed = False
+```
+**Note** `managed=False` does not recognized by `tortoise-orm` and `aerich init-db`, it is only for `aerich migrate`.
+
 ## License
 
 This project is licensed under the

aerich/migrate.py

@@ -264,7 +264,11 @@ class Migrate:
     ) -> None:
         old_m2m_fields = cast("list[dict]", old_model_describe.get("m2m_fields", []))
         new_m2m_fields = cast("list[dict]", new_model_describe.get("m2m_fields", []))
-        new_tables: dict[str, dict] = {field["table"]: field for field in new_models.values()}
+        new_tables: dict[str, dict] = {
+            field["table"]: field
+            for field in new_models.values()
+            if field.get("managed") is not False
+        }
         for action, option, change in get_dict_diff_by_key(old_m2m_fields, new_m2m_fields):
             if (option and option[-1] == "nullable") or change[0][0] == "db_constraint":
                 continue
@@ -387,6 +391,8 @@ class Migrate:
         models_with_rename_field: set[str] = set()  # models that trigger the click.prompt
 
         for new_model_str, new_model_describe in new_models.items():
+            if upgrade and new_model_describe.get("managed") is False:
+                continue
             model = cls._get_model(new_model_describe["name"].split(".")[1])
             if new_model_str not in old_models:
                 if upgrade:
@@ -397,6 +403,8 @@ class Migrate:
                     pass
             else:
                 old_model_describe = cast(dict, old_models.get(new_model_str))
+                if not upgrade and old_model_describe.get("managed") is False:
+                    continue
                 # rename table
                 new_table = cast(str, new_model_describe.get("table"))
                 old_table = cast(str, old_model_describe.get("table"))
@@ -593,6 +601,8 @@ class Migrate:
                     )
 
         for old_model in old_models.keys() - new_models.keys():
+            if not upgrade and old_models[old_model].get("managed") is False:
+                continue
             cls._add_operator(cls.drop_model(old_models[old_model]["table"]), upgrade)
 
     @classmethod

aerich/utils.py

@@ -34,15 +34,11 @@ def get_app_connection_name(config, app_name: str) -> str:
     get connection name
     :param config:
     :param app_name:
-    :return:
+    :return: the default connection name (Usally it is 'default')
     """
-    app = config.get("apps").get(app_name)
-    if not app:
-        raise BadOptionUsage(
-            option_name="--app",
-            message=f'Can\'t get app named "{app_name}"',
-        )
-    return app.get("default_connection", "default")
+    if app := config.get("apps").get(app_name):
+        return app.get("default_connection", "default")
+    raise BadOptionUsage(option_name="--app", message=f"Can't get app named {app_name!r}")
 
 
 def get_app_connection(config, app) -> BaseDBAsyncClient:
@@ -89,8 +85,9 @@ def get_models_describe(app: str) -> dict:
     """
     ret = {}
     for model in Tortoise.apps[app].values():
+        managed = getattr(model.Meta, "managed", None)
         describe = model.describe()
-        ret[describe.get("name")] = describe
+        ret[describe.get("name")] = dict(describe, managed=managed)
     return ret
 
 

tests/models.py

@@ -102,6 +102,7 @@ class Product(Model):
     class Meta:
         unique_together = (("name", "type"),)
         indexes = (("name", "type"),)
+        managed = True
 
 
 class Config(Model):
@@ -118,6 +119,21 @@ class Config(Model):
 
     email: fields.OneToOneRelation[Email]
 
+    class Meta:
+        managed = True
+
+
+class DontManageMe(Model):
+    name = fields.CharField(max_length=50)
+
+    class Meta:
+        managed = False
+
+
+class Ignore(Model):
+    class Meta:
+        managed = False
+
 
 class NewModel(Model):
     name = fields.CharField(max_length=50)

tests/old_models.py

@@ -89,3 +89,40 @@ class Config(Model):
 
     class Meta:
         table = "configs"
+
+
+class DontManageMe(Model):
+    name = fields.CharField(max_length=50)
+
+    class Meta:
+        table = "dont_manage"
+
+
+class Ignore(Model):
+    name = fields.CharField(max_length=50)
+
+    class Meta:
+        managed = True
+
+
+def main() -> None:
+    """Generate a python file for the old_models_describe"""
+    from pathlib import Path
+
+    from tortoise import run_async
+    from tortoise.contrib.test import init_memory_sqlite
+
+    from aerich.utils import get_models_describe
+
+    @init_memory_sqlite
+    async def run() -> None:
+        old_models_describe = get_models_describe("models")
+        p = Path("old_models_describe.py")
+        p.write_text(f"{old_models_describe = }", encoding="utf-8")
+        print(f"Write value to {p}\nYou can reformat it by `ruff format {p}`")
+
+    run_async(run())
+
+
+if __name__ == "__main__":
+    main()