前情提要 这篇文章基于上一篇文章Django多数据库历险记(一) ,将继续讲述关于Django多数据库的历险记。
在上一篇文章中,我创建了一个Django项目:multi_db
,在这个项目里指定了两个app:app_1
和app_2
,每个app下各自创建了一个Model
:Model1
和Model2
,并为这两个app各自分配了独立的数据库db_1
和db_2
。历险继续~
第三关:TestCase 在multi_db
这个Django项目能够运行起来之后,下一步让我们来运行一些TestCase(对,即使这个项目里一个视图函数都还没有,毕竟是TDD😂)。编辑TestCase文件并运行:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 from django.test import TestCasefrom app_1.models import Model1from app_2.models import Model2class TestDB1 (TestCase ): databases = ["db_1" ] def test_query (self ): count = Model1.objects.count() self.assertEqual(count, 0 )class TestDB2 (TestCase ): databases = ["db_2" ] def test_query (self ): count = Model2.objects.count() self.assertEqual(count, 0 )
1 2 3 4 5 6 7 8 9 10 11 12 $ time python manage.py test tests/ Creating test database for alias 'db_1'... Creating test database for alias 'db_2'... System check identified no issues (0 silenced). .. ---------------------------------------------------------------------- Ran 2 tests in 0.006s OK Destroying test database for alias 'db_1'... Destroying test database for alias 'db_2'... python manage.py test tests/ 1.02s user 0.04s system 14% cpu 7.288 total
很好,两个TestCase都测试通过(毕竟没有比这更简单的TestCase了)。一切都看起来非常美好,除了这惊人的命令耗时:为什么创建&销毁两个测试数据库会消耗7秒以上的时间!如果你看过上一篇文章,那么你可能已经隐约猜到原因了。没错,幕后黑手还是migrate
命令。
从Django测试模块源码(runner.py#L614 )中可以看出,对于TestCase里所有的databases
(比如上面的db_1
、db_2
),Django都会创建对应的测试数据库(test_db_1
、test_db_2
),并为每个测试数据库执行一次migrate
命令(creation.py#L67 )。在上一篇文章中,我们就已经知道migrate
命令会一股脑地将所有迁移方案都应用到数据库,这里也不例外。用test
命令的--keepdb
选项验证一下:
1 2 $ python manage.py test tests/ --keepdb # ... 略
1 2 3 4 5 6 7 8 mariadb root@127 .0 .0 .1 :test_db_1> show tables+ | Tables_in_test_db_1 | + | app_1_model1 | | app_2_model2 | | auth_group | # ... 略
1 2 3 4 5 6 7 8 mariadb root@127 .0 .0 .1 :test_db_2> show tables+ | Tables_in_test_db_2 | + | app_1_model1 | | app_2_model2 | | auth_group | # ... 略
那么,有没有办法在对db_1
进行测试时,只在test_db_1
这个测试数据库上创建只属于app_1
的Model
呢?答案是肯定的。
allow_migrate 在上一篇文章的DBRouter
数据库路由类中,我将allow_migrate()
方法的返回值固定为True
,意味着Django可以在任何数据库上、对任何app进行数据库迁移。这当然是一种有点不负责任的做法,所以,只要严格限制在db_1
上只能进行app_1
的迁移、在db_2
上只能进行app_2
的迁移,就可以达成我上文提到的需求了。
1 2 3 4 5 6 7 8 from django.conf import settingsfrom django.db import DEFAULT_DB_ALIASclass DBRouter : def allow_migrate (self, db, app_label, model_name=None , **hints ): return db == settings.DB_ROUTING.get(app_label, DEFAULT_DB_ALIAS)
大功告成 再次执行test
命令,可以很明显地感受到效果:
1 2 3 4 5 $ time python manage.py test tests/ Creating test database for alias 'db_1'... Creating test database for alias 'db_2'...# ... 略 python manage.py test tests/ 0.74s user 0.02s system 57% cpu 1.320 total
总耗时由7秒下降至1秒。再用--keepdb
选项验证一下:
1 2 3 4 5 6 7 mariadb root@127.0.0.1:test_db_1> show tables +---------------------+ | Tables_in_test_db_1 | +---------------------+ | app_1_model1 | | django_ migrations |+---------------------+
完美。
第四关:跨库外键约束
必须要说的是,Django(至少在文档里)并不支持跨数据库的外键约束。当然,这里指的是物理约束,即数据库层面的CONSTRAINT
字段约束。但在实际业务中,由Django管理的、逻辑上的跨库外键约束却并不少见,这给程序员带来了巨大的便利——但代价是什么呢?代价往往是抛弃Django的数据迁移机制。
定义 在app_1
里创建一个新Model(省略内容见上一篇文章 ),并生成对应迁移方案:
1 2 3 4 5 6 7 class ChildModel1 (models.Model): name = models.CharField(max_length=255 ) parent1 = models.ForeignKey("Model1" , on_delete=models.CASCADE) parent2 = models.ForeignKey("app_2.Model2" , on_delete=models.CASCADE)
1 2 3 4 $ python manage.py makemigrations Migrations for 'app_1': app_1/migrations/0002_childmodel1.py - Create model ChildModel1
迁移 1 2 3 4 5 6 7 8 $ python manage.py migrate app_1 --database db_1 Operations to perform: Apply all migrations: app_1 Running migrations: Applying app_2.0001_initial... OK Applying app_1.0002_childmodel1...# ... 略 django.db.utils.OperationalError: (1005, 'Can\'t create table `db_1`.`app_1_childmodel1` (errno: 150 "Foreign key constraint is incorrectly formed")')
果不其然,报错了。Django想要在db_1
上应用app_2
的迁移方案0001_initial
来为ChildModel1
的迁移铺路,但在上一关的allow_migrate()
中我已经严格限制了数据库和app的对应关系,所以实际上app_2.0001_initial
被跳过了(虽然还是在迁移进度里留下了成功标记)。
于是到了物理约束部分,Django想要在db1
上创建从ChildModel1
到Model2
的CONSTRAINT
约束就不可能成功:因为app_2.0001_inital
这个前提并没有实际应用。打开数据库验证一下:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 mariadb root@127 .0 .0 .1 :db_1> show tables + | Tables_in_db_1 | + | app_1_childmodel1 | | app_1_model1 | | django_migrations | + mariadb root@127 .0 .0 .1 :db_1> select * from django_migrations+ | id | app | name | applied | + | 1 | app_1 | 0001 _initial | 2020 -04 -27 xx:xx:xx.xxxxxx | | 2 | app_2 | 0001 _initial | 2020 -04 -27 xx:xx:xx.xxxxxx | | 3 | app_1 | 0002 _childmodel1 | 2020 -04 -27 xx:xx:xx.xxxxxx | + mariadb root@127 .0 .0 .1 :db_1> show create table app_1_childmodel1CREATE TABLE `app_1_childmodel1` ( `id` int (11 ) NOT NULL AUTO_INCREMENT, `name` varchar (255 ) COLLATE utf8mb4_unicode_ci NOT NULL , `parent1_id` int (11 ) NOT NULL , `parent2_id` int (11 ) NOT NULL , PRIMARY KEY (`id`), KEY `xxx` (`parent1_id`), CONSTRAINT `xxx` FOREIGN KEY (`parent1_id`) REFERENCES `app_1_model1` (`id`) ) ENGINE= InnoDB DEFAULT CHARSET= utf8mb4 COLLATE = utf8mb4_unicode_ci
应用 当然,即使Django在应用迁移方案的时候报错了,但缺失的只有一个数据库层面的约束字段,并不影响逻辑层面的外键约束。所以,我们可以直接在Django里使用这个Model了:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from django.http import JsonResponsefrom app_1.models import ChildModel1, Model1from app_2.models import Model2def test (request ): child = ChildModel1.objects.create( name="carl" , parent1=Model1.objects.create(name="alice" ), parent2=Model2.objects.create(name="bob" ), ) return JsonResponse({ "name" : child.name, "parent1" : child.parent1.name, "parent2" : child.parent2.name })
1 2 3 4 5 6 7 8 from app_1.views import test urlpatterns = [ path('admin/' , admin.site.urls), path('' , test) ]
1 2 3 4 5 6 7 8 9 10 $ python manage.py runserver 8000 Performing system checks... Watching for file changes with StatReloader System check identified no issues (0 silenced). April 27, 2020 - xx:xx:xx# 新termial $ curl 127.0.0.1:8000 {"name": "carl", "parent1": "alice", "parent2": "bob"}
TestCase 在上一节中可以这样蒙混过关,是因为migrate
命令的失败并不会影响runserver
命令的执行,但在TestCase里这样就行不通了。在第三关中提到,Django在进行TestCase前会对测试数据库执行migrate
命令,而migrate
的失败会导致TestCase直接结束。
解决方案主要有取消allow_migrate()
对app&数据库对应关系的限制、在运行TestCase时使用sqlite3(默认禁用外键约束)作为测试数据库。但这些都不能从根本上解决问题,这也就是为什么我在本关开头提到,跨库外键的代价往往是抛弃Django的数据迁移机制。
要想从根本上解决这个问题,就只有避免出现物理上的外键约束,只由Django进行逻辑上的外键约束管理。这个话题,就留到下篇文章再讲吧。(我怎么感觉我要被打了……)