Django 使用心得（四）多數據庫

時間 2019-11-08

原文原文鏈接

博客原文地址：elfgzp.cn/2019/01/09/…html

相信有開發者在項目中可能會有須要將不一樣的 app 數據庫分離，這樣就須要使用多個數據庫。
網上也有很是多的與 db_router 相關的文章，本篇文章也會簡單介紹一下。
除此以外，還會介紹一下筆者在具體項目中使用多數據庫的一些心得和一些坑。但願能給讀者帶來必定的幫助，如果讀者們也有相關的心得別忘了留言，能夠一塊兒交流學習。python

使用 Router 來實現多數據庫

首先咱們能夠從 Django 的官方文檔瞭解到如何使用 routers 來使用多數據庫。mysql

官方文檔 Using Routersgit

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

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 的多數據庫結構，形成了一些異常。sql

因而我去查閱 Django 單元測試的源碼發現這樣一段代碼，他是用於判斷某個 app 的 migrations（數據庫遷移）是否要在某個數據庫執行。mongodb

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。django

回到 utils.py 的函數中來，能夠看到 allow 就獲得了這個 None 的返回值，可是他判斷了 is not None 爲假命題，那麼循環繼續。bash

這樣致使了全部對於這個數據庫 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()
    ... 
複製代碼