Fixed conflict of model parameter and hints in DefaultRouter.allow_migrate (#46)

Fixed conflict of model parameter and hints in DefaultRouter.allow_migrate

docker-compose.yml

@@ -24,7 +24,7 @@ services:
     build:
       context: .
       args:
-        - PYTHON_VER=latest
+        - PYTHON_IMAGE_TAG=latest
     environment:
       - REDIS_HOST=redis_db
       - PGHOST=postgres_db

docs/routing.md

@@ -25,12 +25,17 @@ Router is a class, defining 3 methods:
   Returns `database alias` to use for given `model` for `SELECT` queries.
 * `def db_for_write(self, model: ClickHouseModel, **hints) -> str`  
   Returns `database alias` to use for given `model` for `INSERT` queries.
-* `def allow_migrate(self, db_alias: str, app_label: str, operation: Operation, model: Optional[ClickHouseModel] = None, **hints: dict) -> bool`
+* `def allow_migrate(self, db_alias: str, app_label: str, operation: Operation, **hints: dict) -> bool`
   Checks if migration `operation` should be applied in django application `app_label` on database `db_alias`.
-  Optional `model` field can be used to determine migrations on concrete model.
+  Optional `hints` help to pass additional info which can be used to test migrations availability on concrete model.
 
 By default [CLICKHOUSE_DATABASE_ROUTER](configuration.md#clickhouse_database_router) is used.
  It gets routing information from model fields, described below.  
+ It also gives ability to use 2 kinds of hints:  
+ * `force_migrate_on_databases: Iterable[str]` - concrete database aliases where migration should be applied
+ * `model: Type[ClickHouseModel]` - a model class, to read routing attributes from. 
+    Can be set as class or its string name. 
+    If name is set, class is searched in current `<app_label>.<config.MODELS_MODULE>` package.  
  
 ## ClickHouseModel routing attributes
 Default database router reads routing settings from model attributes.

setup.py

@@ -13,7 +13,7 @@ with open('requirements.txt') as f:
 
 setup(
     name='django-clickhouse',
-    version='1.2.0',
+    version='1.2.1',
     packages=['django_clickhouse', 'django_clickhouse.management.commands'],
     package_dir={'': 'src'},
     url='https://github.com/carrotquest/django-clickhouse',

src/django_clickhouse/migrations.py

@@ -40,10 +40,9 @@ class Migration:
         database = database or connections[db_alias]
 
         for op in self.operations:
-            model_class = getattr(op, 'model_class', None)
             hints = getattr(op, 'hints', {})
 
-            if db_router.allow_migrate(db_alias, self.__module__, op, model_class, **hints):
+            if db_router.allow_migrate(db_alias, self.__module__, op, **hints):
                 op.apply(database)
 
 

src/django_clickhouse/routers.py

@@ -30,24 +30,26 @@ class DefaultRouter:
         """
         return random.choice(model.write_db_aliases)
 
-    def allow_migrate(self, db_alias: str, app_label: str, operation: Operation,
-                      model=None, **hints) -> bool:
+    def allow_migrate(self, db_alias: str, app_label: str, operation: Operation, **hints) -> bool:
         """
         Checks if migration can be applied to given database
         :param db_alias: Database alias to check
         :param app_label: App from which migration is got
         :param operation: Operation object to perform
-        :param model: Model migration is applied to
         :param hints: Hints to make correct decision
         :return: boolean
         """
         if hints.get("force_migrate_on_databases", None):
             return db_alias in hints["force_migrate_on_databases"]
 
-        if hints.get('model'):
-            model = '%s.%s.%s' % (app_label, config.MODELS_MODULE, hints['model']) \
-                if isinstance(hints['model'], str) else hints['model']
+        model = hints.get('model') or getattr(operation, 'model_class', None)
+        if model is None:
+            raise ValueError('"model_class" attribute is not defined for operation "%s". '
+                             'Please provide "force_migrate_on_databases" or "model" in hints.'
+                             % operation.__class__.__name__)
 
+        model = '%s.%s.%s' % (app_label, config.MODELS_MODULE, model) \
+            if isinstance(model, str) else model
         model = lazy_class_import(model)
 
         if operation.__class__ not in {CreateTable, DropTable}:

tests/test_migrations.py

@@ -13,7 +13,7 @@ from tests.clickhouse_models import ClickHouseTestModel
 
 
 class NoMigrateRouter(DefaultRouter):
-    def allow_migrate(self, db_alias, app_label, operation, model=None, **hints):
+    def allow_migrate(self, db_alias, app_label, operation, **hints):
         return False
 
 

tests/test_routers.py

@@ -0,0 +1,59 @@
+from django.test import SimpleTestCase
+
+from django_clickhouse.migrations import RunSQL, CreateTable
+from django_clickhouse.routers import DefaultRouter
+from tests.clickhouse_models import ClickHouseTestModel
+
+
+class DefaultRouterAllowMigrateTest(SimpleTestCase):
+    def setUp(self):
+        self.router = DefaultRouter()
+        self.operation = RunSQL('SELECT 1')
+
+    def test_hints_model_class(self):
+        hints = {'model': ClickHouseTestModel}
+
+        with self.subTest('Allow migrate'):
+            res = self.router.allow_migrate('default', 'tests', self.operation, **hints)
+            self.assertTrue(res)
+
+        with self.subTest('Reject migrate'):
+            res = self.router.allow_migrate('other', 'tests', self.operation, **hints)
+            self.assertFalse(res)
+
+    def test_hints_model_name(self):
+        hints = {'model': 'ClickHouseTestModel'}
+
+        with self.subTest('Allow migrate'):
+            res = self.router.allow_migrate('default', 'tests', self.operation, **hints)
+            self.assertTrue(res)
+
+        with self.subTest('Reject migrate'):
+            res = self.router.allow_migrate('other', 'tests', self.operation, **hints)
+            self.assertFalse(res)
+
+    def test_hints_force_migrate_on_databases(self):
+        hints = {'force_migrate_on_databases': ['secondary']}
+
+        with self.subTest('Allow migrate'):
+            res = self.router.allow_migrate('secondary', 'apps', self.operation, **hints)
+            self.assertTrue(res)
+
+        with self.subTest('Reject migrate'):
+            res = self.router.allow_migrate('default', 'apps', self.operation, **hints)
+            self.assertFalse(res)
+
+    def test_model_operation(self):
+        with self.subTest('Allow migrate'):
+            operation = CreateTable(ClickHouseTestModel)
+            res = self.router.allow_migrate('default', 'apps', operation)
+            self.assertTrue(res)
+
+        with self.subTest('Reject migrate'):
+            operation = CreateTable(ClickHouseTestModel)
+            res = self.router.allow_migrate('other', 'apps', operation)
+            self.assertFalse(res)
+
+    def test_no_model(self):
+        with self.assertRaises(ValueError):
+            self.router.allow_migrate('default', 'apps', self.operation)