Django 使用心得 （四）多資料庫

摘要： 部落格原文地址： elfgzp.cn/2019/01/09/… 相信有開發者在專案中可能會有需要將不同的 app 資料庫分離，這樣就需要使用多個數據庫。 網上也有非常多的與 db_router 相關的文章，本篇文章也會簡單介紹一下。 除此之外，還會介紹一下筆...

部落格原文地址： elfgzp.cn/2019/01/09/…

相信有開發者在專案中可能會有需要將不同的 app 資料庫分離，這樣就需要使用多個數據庫。

網上也有非常多的與 db_router 相關的文章，本篇文章也會簡單介紹一下。

除此之外，還會介紹一下筆者在具體專案中使用多資料庫的一些心得和一些 坑 。希望能給讀者帶來一定的幫助，若是讀者們也有相關的心得別忘了留言，可以一起交流學習。

使用 Router 來實現多資料庫

首先我們可以從 Django 的官方文件瞭解到如何使用 routers 來使用多資料庫。

官方文件Using Routers

官方文件中定義了一個 AuthRouter 用於儲存將 Auth app 相關的表結構。

class AuthRouter:
"""
A router to control all database operations on models in the
auth application.
"""
def db_for_read(self, model, **hints):
"""
Attempts to read auth models go to auth_db.
"""
if model._meta.app_label == 'auth':
return 'auth_db'
return None

def db_for_write(self, model, **hints):
"""
Attempts to write auth models go to auth_db.
"""
if model._meta.app_label == 'auth':
return 'auth_db'
return None

def allow_relation(self, obj1, obj2, **hints):
"""
Allow relations if a model in the auth app is involved.
"""
if obj1._meta.app_label == 'auth' or \
obj2._meta.app_label == 'auth':
return True
return None

def allow_migrate(self, db, app_label, model_name=None, **hints):
"""
Make sure the auth app only appears in the 'auth_db'
database.
"""
if app_label == 'auth':
return db == 'auth_db'
return None
複製程式碼

但是我在實際使用中遇到一個問題，在執行 python manage.py test 來進行單元測試時，這個資料庫內依然會生成其他 app 的表結構。

正常情況下是沒什麼問題的，但是我使用了 mysql 與 mongodb 的多資料庫結構，造成了一些異常。

於是我去查閱 Django 單元測試的原始碼發現這樣一段程式碼，他是用於判斷某個 app 的 migrations （資料庫遷移）是否要在某個資料庫執行。

This Github Sample is by django

django/db/utils.py view raw

def allow_migrate(self, db, app_label, **hints):
for router in self.routers:
try:
method = router.allow_migrate
except AttributeError:
# If the router doesn't have a method, skip to the next one.
continue

allow = method(db, app_label, **hints)

if allow is not None:
return allow
return True
複製程式碼

他這個函式相當於是在執行 Router 中的 allow_migrate ，並取其結果來判斷是否要執行資料庫遷移。

也就是官方給的例子：

def allow_migrate(self, db, app_label, model_name=None, **hints):
"""
Make sure the auth app only appears in the 'auth_db'
database.
"""
if app_label == 'auth':
return db == 'auth_db'
return None
複製程式碼

但是這裡有一個問題，假設 app_label 不等於 auth （相當於你設定的 app 名稱），但是 db 卻等於 auth_db ，此時這個函式會返回 None 。

回到 utils.py 的函式中來，可以看到 allow 就得到了這個 None 的返回值，但是他判斷了 is not None 為 假命題 ，那麼迴圈繼續。

這樣導致了所有對於這個資料庫 auth_db 並且 app_label 不為 auth 的結果均返回 None 。最後迴圈結束，返回結果為 True ，這意味著， 所有其他 app_label 的資料庫遷移均會在這個資料庫中執行。

為了解決這個問題，我們需要對官方給出的示例作出修改：

def allow_migrate(self, db, app_label, model_name=None, **hints):
"""
Make sure the auth app only appears in the 'auth_db'
database.
"""
if app_label == 'auth':
return db == 'auth_db'
elif db == 'auth_db':# 若資料庫名稱為 auth_db 但 app_label 不為 auth 直接返回 False
return False
else:
return None
複製程式碼

執行 migrate 時指定 –database

我們定義好 Router 後，在執行 python manage.py migrate 時可以發現，資料庫遷移動作並沒有執行到除預設資料庫以外的資料庫中， 這是因為 migrate 這個 command 必須要指定額外的引數。

官方文件 Synchronizing your databases

閱讀官方文件可以知道，若要將資料庫遷移執行到非預設資料庫中時， 必須 要指定資料庫 --database 。

$ ./manage.py migrate --database=users
$ ./manage.py migrate --database=customers
複製程式碼

但是這樣的話會導致我們使用 CI/CD 部署服務非常的不方便，所以我們可以通過自定義 command 來實現 migrate 指定資料庫。

其實實現方式非常簡單，就是基於 django 預設的 migrate 進行改造，在最外層加一個迴圈，然後在自定義成一個新的命令 multidbmigrate 。

This Github Sample is by elfgzp

multidatabases/management/commands/multidbmigrate.py view raw

...
def handle(self, *args, **options):
self.verbosity = options['verbosity']
self.interactive = options['interactive']

# Import the 'management' module within each installed app, to register
# dispatcher events.
for app_config in apps.get_app_configs():
if module_has_submodule(app_config.module, "management"):
import_module('.management', app_config.name)

db_routers = [import_string(router)() for router in conf.settings.DATABASE_ROUTERS] # 對所有的 routers 進行 migrate 操作
for connection in connections.all():
# Hook for backends needing any database preparation
connection.prepare_database()
# Work out which apps have migrations and which do not
executor = MigrationExecutor(connection, self.migration_progress_callback)

# Raise an error if any migrations are applied before their dependencies.
executor.loader.check_consistent_history(connection)

# Before anything else, see if there's conflicting apps and drop out
# hard if there are any
conflicts = executor.loader.detect_conflicts()
...
複製程式碼

由於程式碼過長，這裡就不全部 copy 出來，只放出其中最關鍵部分，完整部分可以參閱 elfgzp/django_experience 倉庫。

在支援事務資料庫與不支援事務資料庫混用在單元測試遇到的問題

在筆者使用 Mysql 和 Mongodb 時，遇到了個問題。

總所周知，Mysql 是支援事務的資料庫，而 Mongodb 是不支援的。在專案中筆者同時使用了這兩個資料庫，並且運行了單元測試。

發現在執行完某一個單元測試後，我在 Mysql 資料庫所生成的初始化資料（即筆者在 migrate 中使用 RunPython 生成了一些 demo 資料）全部被清除了，導致其他單元測試測試失敗。

通過 TestCase 類的特性可以知道，單元測試在執行完後會去執行 tearDown 來做清除垃圾的操作。於是順著這個函式，筆者去閱讀了 Django 中對應函式的原始碼，發現有一段這樣的邏輯。

...
def connections_support_transactions():# 判斷是否所有資料庫支援事務
"""Return True if all connections support transactions."""
return all(conn.features.supports_transactions for conn in connections.all())
...

class TransactionTestCase(SimpleTestCase):
...
multi_db = False
...
@classmethod
def _databases_names(cls, include_mirrors=True):
# If the test case has a multi_db=True flag, act on all databases,
# including mirrors or not. Otherwise, just on the default DB.
if cls.multi_db:
return [
alias for alias in connections
if include_mirrors or not connections[alias].settings_dict['TEST']['MIRROR']
]
else:
return [DEFAULT_DB_ALIAS]
...
def _fixture_teardown(self):
# Allow TRUNCATE ... CASCADE and don't emit the post_migrate signal
# when flushing only a subset of the apps
for db_name in self._databases_names(include_mirrors=False):
# Flush the database
inhibit_post_migrate = (
self.available_apps is not None or
(# Inhibit the post_migrate signal when using serialized
# rollback to avoid trying to recreate the serialized data.
self.serialized_rollback and
hasattr(connections[db_name], '_test_serialized_contents')
)
)
call_command('flush', verbosity=0, interactive=False,# 清空資料庫表
database=db_name, reset_sequences=False,
allow_cascade=self.available_apps is not None,
inhibit_post_migrate=inhibit_post_migrate)
...

class TestCase(TransactionTestCase):
...
def _fixture_teardown(self):
if not connections_support_transactions():# 判斷是否所有資料庫支援事務
return super()._fixture_teardown()
try:
for db_name in reversed(self._databases_names()):
if self._should_check_constraints(connections[db_name]):
connections[db_name].check_constraints()
finally:
self._rollback_atomics(self.atomics)
...
複製程式碼

看到這段程式碼後筆者都快氣死了，這個單元測試明明只是只對單個數據庫起作用， multi_db 這個屬性預設也是為 False ，這個單元測試作用在 Mysql 跟 Mongodb 有什麼關係呢！？正確的邏輯應應該是判斷 _databases_names 即這個單元測試所涉及的資料庫支不支援事務才對。

於是需要對 TestCase 進行了改造，並且將單元測試繼承的 TestCase 修改為新的 TestCase。修改結果如下：

This Github Sample is by elfgzp

multidatabases/testcases.py view raw

class TestCase(TransactionTestCase):
"""
此類修復 Django TestCase 中由於使用了多資料庫，但是 multi_db 並未指定多資料庫，單元測試依然只是在一個數據庫上執行。
但是原始碼中的 connections_support_transactions 將所有資料庫都包含進來了，導致在同時使用 MangoDB 和 MySQL 資料庫時，
MySQL 資料庫無法回滾，清空了所有的初始化資料，導致單元測試無法使用初始化的資料。
"""

@classmethod
def _databases_support_transactions(cls):
return all(
conn.features.supports_transactions
for conn in connections.all()
if conn.alias in cls._databases_names()
)
...

def _fixture_setup(self):
if not self._databases_support_transactions():
# If the backend does not support transactions, we should reload
# class data before each test
self.setUpTestData()
return super()._fixture_setup()

assert not self.reset_sequences, 'reset_sequences cannot be used on TestCase instances'
self.atomics = self._enter_atomics()
... 
複製程式碼

除了 _fixture_setup 以外還有其他成員函式需要將判斷函式改為 _databases_support_transactions ，完整程式碼參考 elfgzp/django_experience 倉庫

總結

踩過這些坑，筆者更加堅信不能太相信官方文件和原始碼，要自己去學習研究原始碼的實現，才能找到解決問題的辦法。