Django 多数据库支持

Posted 2021-11-03 wuzdandz

很多时候,需要去其他数据库查询数据,都将会面临多数据库支持问题.

1.在settings文件内添加多数据库连接

DATABASES = {
    ‘default‘: {
        ‘ENGINE‘: ‘django.db.backends.sqlite3‘,
        ‘NAME‘: os.path.join(BASE_DIR, ‘db.sqlite3‘),
    },
    ‘test‘: {
        ‘ENGINE‘: ‘django.db.backends.mysql‘,
        ‘NAME‘: ‘pro_control‘,
        ‘USER‘: ‘root‘,
        ‘PASSWORD‘: ‘root‘,
        ‘HOST‘: ‘10.0.11.11‘,
        ‘PORT‘: ‘3306‘,
    }
}

如果默认数据库的概念在项目上下文中没有意义，则需要始终小心地指定要使用的数据库。Django需要一个default数据库,如果不需要使用,可以设置为空字典.

举个栗子

DATABASES = {
    ‘default‘: {},
    ‘users‘: {
        ‘NAME‘: ‘user_data‘,
        ‘ENGINE‘: ‘django.db.backends.mysql‘,
        ‘USER‘: ‘mysql_user‘,
        ‘PASSWORD‘: ‘superS3cret‘
    },
    ‘customers‘: {
        ‘NAME‘: ‘customer_data‘,
        ‘ENGINE‘: ‘django.db.backends.mysql‘,
        ‘USER‘: ‘mysql_cust‘,
        ‘PASSWORD‘: ‘[email protected]‘
    }
}

用例

如果你试图连接一个没有在DATABASES内定义的数据库django将产生django.db.utils.ConnectionDoesNotExist异常

将模型同步到数据库

./manage.py migrate  # 操作将同步到default数据库
./manage.py migrate --database=test  # 同步到test数据库

如果不是想所有的模型都同步到一个数据库上你可以定义一个database router,新建db_router.py文件

class AuthRouter:
    def db_for_read(self, model, **hits):
        if model._meta.app_label == ‘test01‘:
            return ‘test‘
        return None

    def db_for_write(self, model, **hits):
        if model._meta.app_label == ‘test01‘:
            return ‘test‘
        return None

    def allow_relation(self, obj1, obj2, **hits):
        if obj1._meta.app_label == ‘test01‘ or obj2._meta.app_label == ‘test01‘:
            return True
        return None

    def allow_migrate(self, db, app_label, model_name=None, **hits):
        if app_label == ‘test01‘:
            return db == ‘test‘
        return None

在setting文件内添加DATABASE_ROUTERS参数

DATABASE_ROUTERS = [‘test01.db_router.AuthRouter‘, ]

为测试model指定,在test01的models文件内写入

from django.db import models
import shortuuid

# Create your models here.


def createuuid():
    return shortuuid.uuid()


class CourseManage(models.Model):
    """课程管控表"""
    uuid = models.CharField(
        ‘ID‘,
        max_length=22,
        primary_key=True,
        default=createuuid,
        editable=False)
    add_time = models.DateTimeField(‘创建时间‘, auto_now_add=True)

    del_state_type = ((0, ‘已删除‘), (1, ‘默认‘))
    modified_time = models.DateTimeField(‘修改时间‘, auto_now=True)
    del_state = models.IntegerField(
        ‘删除状态‘, choices=del_state_type, default=1, db_index=True)
    config_status = ((1, ‘进行中‘), (2, ‘完结‘), )
    remote_id = models.CharField(verbose_name=‘远端产品id‘, max_length=30)
    status = models.PositiveSmallIntegerField(choices=config_status, default=1)

    class Meta:
        verbose_name = ‘课程‘
        verbose_name_plural = verbose_name
        db_table = ‘backend_coursemanage‘

db_table指定此模型在远端的数据库内的表名称.不然会报错

按照上述操作,便能连接并查询.

 

标签:当项目需要执行python manage.py makemigrations时,会发现,test01项目的migrations文件夹会产生migrations文件.并且终端也会产生提示,

可以尝试删除此migrations,来避免这个问题,allow_migrate的判断只针对真正的migrate操作.

 

如何选择数据库? 

1.模型层

User.objects.using(‘legacy_users‘).get(username=‘fred‘)
user_obj.save(using=‘new_users‘)
user_obj.delete(using=‘legacy_users‘)

2.原生光标

from django.db import connections
with connections[‘my_db_alias‘].cursor() as cursor:
    ...

 

关于replication

DATABASES = {
    ‘default‘: {},
    ‘auth_db‘: {
        ‘NAME‘: ‘auth_db‘,
        ‘ENGINE‘: ‘django.db.backends.mysql‘,
        ‘USER‘: ‘mysql_user‘,
        ‘PASSWORD‘: ‘swordfish‘,
    },
    ‘primary‘: {
        ‘NAME‘: ‘primary‘,
        ‘ENGINE‘: ‘django.db.backends.mysql‘,
        ‘USER‘: ‘mysql_user‘,
        ‘PASSWORD‘: ‘spam‘,
    },
    ‘replica1‘: {
        ‘NAME‘: ‘replica1‘,
        ‘ENGINE‘: ‘django.db.backends.mysql‘,
        ‘USER‘: ‘mysql_user‘,
        ‘PASSWORD‘: ‘eggs‘,
    },
    ‘replica2‘: {
        ‘NAME‘: ‘replica2‘,
        ‘ENGINE‘: ‘django.db.backends.mysql‘,
        ‘USER‘: ‘mysql_user‘,
        ‘PASSWORD‘: ‘bacon‘,
    },
}

import random

class PrimaryReplicaRouter:
    def db_for_read(self, model, **hints):
        """
        Reads go to a randomly-chosen replica.
        """
        return random.choice([‘replica1‘, ‘replica2‘])  # 这里是重点

    def db_for_write(self, model, **hints):
        """
        Writes always go to primary.
        """
        return ‘primary‘

    def allow_relation(self, obj1, obj2, **hints):
        """
        Relations between objects are allowed if both objects are
        in the primary/replica pool.
        """
        db_list = (‘primary‘, ‘replica1‘, ‘replica2‘)
        if obj1._state.db in db_list and obj2._state.db in db_list:
            return True
        return None

    def allow_migrate(self, db, app_label, model_name=None, **hints):
        """
        All non-auth models end up in this pool.
        """
        return True