【Django Rest Framework】アプリ毎にDB接続先を切り替える構成 (SimpleJWT)

概要

https://gyazo.com/6cccd5c4340ade1d202bc36400f0964e

設定する項目

■ 設定項目

2. アプリ毎のマッピング設定 (settings.py / db_router.py)

3. (connection.cursor()を使用している場合)生SQLの実行先コネクション切替 (views.py など)

5. TokenObtainPairView 等のトークン発行・検証処理のカスタマイズ

------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ --------------

① DATABASESでの各種DB接続先設定

アプリ毎で設定するDatabase設定を以下のように記載

code: (edit) settings.py

DATABASES = {

'default': {

'ENGINE': 'django.db.backends.sqlite3',

'NAME': BASE_DIR / 'db_app1.sqlite3',

},

'app2': {

'ENGINE': 'django.db.backends.sqlite3',

'NAME': BASE_DIR / 'db_app2.sqlite3',

}

}

------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ --------------

② アプリ毎マッピング設定

①にて設定したデータベース設定に対してアプリ毎のマッピング設定を設定

※config(プロジェクト名)/db_router.pyの追加、config(プロジェクト名)/setting.pyへの追記

code: (new) config/db_router.py

from config.settings import DATABASE_APPS_MAPPING

class DatabaseRouter(object):

def db_for_read(self, model, **hints):

return DATABASE_APPS_MAPPING.get(model._meta.app_label)

def db_for_write(self, model, **hints):

return DATABASE_APPS_MAPPING.get(model._meta.app_label)

def allow_relation(self, obj1, obj2, **hints):

db1 = DATABASE_APPS_MAPPING.get(obj1._meta.app_label)

db2 = DATABASE_APPS_MAPPING.get(obj2._meta.app_label)

return db1 == db2 if db1 and db2 else None

def allow_migrate(self, db, app_label, model=None, **hints):

if db in DATABASE_APPS_MAPPING.values():

return DATABASE_APPS_MAPPING.get(app_label) == db

elif app_label in DATABASE_APPS_MAPPING:

return False

code: (edit) settings.py

DATABASE_APPS_MAPPING = {

'admin' : 'default',

'auth' : 'default',

'contenttypes': 'default',

'sessions' : 'default',

'messages' : 'default',

'staticfiles' : 'default',

'app1' : 'default',

'app2' : 'app2',

}

DATABASE_ROUTERS = [

'config.db_router.DatabaseRouter',

]

------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ --------------

③ 生SQLの実行先コネクション切替

django.dbのconnection.cursor()で生SQLを実行している場合は、①②の設定だけでは読み込み先DBの設定がデフォルトから切り替わらない

なので追加で以下のようにアプリ毎にdjango.dbのconnectionsからコネクションの指定をした上でcursorが使われるように更新

code: (before) views.py

from django.db import connection

with connection.cursor() as cursor:

cursor.execute(sql, params)

code: (after) views.py

from django.db import connections

cursor.execute(sql, params)

------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ --------------

④ JWTAuthenticationのカスタマイズ

SimpleJWTの認証時にデフォルトDBを参照してしまうため、

アプリ毎のユーザモデルに切り替える必要があります。

code: (new) app1/common/JWTAuthentication.py

# カスタムJWTAuthenticationクラスを元々のJWTAuthentication.pyを元に作成

# ここでアプリ毎のDB接続を意識して認証

code: views.py

from app1.common.JWTAuthentication import JWTAuthentication

class ContentListViewSet(ModelViewSet):

------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ --------------

⑤ TokenObtainPairView等のカスタマイズ

トークン発行時もデフォルトDBを参照するため、View/Serializerをサブクラス化して

アプリ毎DBに切り替えます。

※大事なのはアプリ毎に定義しておくことで、アプリ毎のスキーマに対するモデル設定を見に行くようになること

code: app1/views/auth/token_obtain_pair.py

# TokenObtainPairViewをカスタマイズしてDB指定

from rest_framework_simplejwt import views as jwt_views

from rest_framework_simplejwt import serializers as jwt_serializers

from rest_framework_simplejwt import exceptions as jwt_exp

from rest_framework.response import Response

from rest_framework import status

class TokenObtainSerializer(jwt_serializers.TokenObtainPairSerializer):

@classmethod

def get_token(cls, user):

token = super().get_token(user)

# ペイロードへの値の設定する場合(しない場合はこのTokenObtainSerializer自体オーバーライド不要)

return token

class TokenObtainView(jwt_views.TokenObtainPairView):

# Tokenペアの発行

def post(self, request, *args, **kwargs):

# serializer = self.get_serializer(data=request.data)

# ※JWTトークンにsettings.pyで指定しているユーザモデルのpk以外を含ませないのであれば以下でなく上の設定でOK

serializer = TokenObtainSerializer(data=request.data)

# 検証

try:

serializer.is_valid(raise_exception=True)

# エラーハンドリング

except jwt_exp.TokenError as e:

res = Response(serializer.validated_data, status=status.HTTP_200_OK)

return res

------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ ------------------ --------------

まとめ

・JWT認証やトークン発行もアプリ毎DBを参照するようカスタマイズ

・これでアプリ毎にDB接続先を安全に切り替え可能

参考

・ Qiita: Django データベース接続先をログインユーザ別に切り替える

・ 個人ブログ: Djangoでデータベースを分割して利用