Django DRF Post for Multiple Database Connections

Question

Fairly new to DRF and I am working on a project that connects to multiple databases. I am able to use get to retrieve data from whichever specified database however when using post I get an error related to my database connection and table name. I think I need to reference the database in the post method but I am striking out researching the issue. Anyone know how to reference the specific database for self.create ?

class AdminView(
    mixins.ListModelMixin, 
    mixins.CreateModelMixin,
    generics.GenericAPIView):
        serializer_class = AdminMeterLakeSerializer
        queryset = AdminMeterLake.objects.using('testdb').all()

        def get(self, request, *args, **kwargs):
            return self.list(request, *args, **kwargs)

        def post(self, request, *args, **kwargs):
            return self.create(request.data, *args, **kwargs)

Answer 1

I recommend to create context manager. That will allow to effectively control DB connections in future:

• Add DB router to config/settings.py

...
DATABASE_ROUTERS = [
    'config.routers.DynamicDatabaseRouter',
]
...

• Create DynamicDatabaseRouter in config/routers.py:

from contextvars import ContextVar
active_db = ContextVar("DB to use", default=None)

def get_active_db():
    # return default connection if not set
    db = active_db.get(None)
    return db if db else 'default'

def set_active_db(connection_name):
    return active_db.set(connection_name)

class DynamicDatabaseRouter:

    @staticmethod
    def _get_db(*args, **kwargs):
        db = get_active_db()
        return db

    db_for_read = _get_db
    db_for_write = _get_db

• Add context manager into config/context.py:

from contextlib import ContextDecorator
from config.routers import set_active_db


class db(ContextDecorator):

    def __init__(self, connection_name):
        self.connection_name = connection_name

    def __enter__(self):
        set_active_db(connection_name)
        return self

    def __exit__(self, *exc):
        set_active_db(None)
        return False

And your class should be updated to:

from config.context import db

class AdminView(
    mixins.ListModelMixin, 
    mixins.CreateModelMixin,
    generics.GenericAPIView):
        serializer_class = AdminMeterLakeSerializer
        queryset = AdminMeterLake.objects.using('testdb').all()

        def get(self, request, *args, **kwargs):
            return self.list(request, *args, **kwargs)

        def post(self, request, *args, **kwargs):
            with db('testdb'):
                return self.create(request.data, *args, **kwargs)

The flow is:

1. When you run with db('testdb') the __enter__ method is called
2. active_db context variable value will be updated to testdb
3. database router read active_db value and use testdb connection
4. When operation is complete __exit__ will be called and context variable value will be reverted to None *

• if active_db value is None router will return default connection

OR 1

You may simply use django-dynamic-db-router

Also this information may be useful

OR 2

You may set custom Manager class for database Model:

from django.db import models

class AdminMeterLake(models.Model):
    ...
    col_name_1 = models.CharField(max_length=50)
    col_name_2 = models.CharField(max_length=50)
    ...

    objects = models.Manager().using('testdb')

In this case testdb will be used by default and you will be able to set queryset = AdminMeterLake.objects.all() instead of queryset = AdminMeterLake.objects.using('testdb').all()

• models.Manager().using('testdb') way is didn't tested by me and it is just theoretical solution (but it should work I guess...)

Answer 2

Working off of the absolute boss rzlvmp's solution one, I was able to get a working solution capable of routing databases on demand.

Project settings:

'''
    DATABASE_ROUTERS = [
    'config.routers.DynamicDatabaseRouter',
    ]
'''

config/routers.py which I placed into my root directory with manage.py

   from contextvars import ContextVar
   active_db = ContextVar("DB to use", default=None)

    def get_active_db():
        # return default connection if not set
        db = active_db.get(None)
        return db if db else 'default'

    def set_active_db(connection_name):
        return active_db.set(connection_name)

    class DynamicDatabaseRouter:

        @staticmethod
        def _get_db(*args, **kwargs):
            db = get_active_db()
            return db

        db_for_read = _get_db
        db_for_write = _get_db

config/context.py

from contextlib import ContextDecorator
from config.routers import set_active_db


class db(ContextDecorator):

    def __init__(self, connection_name):
        self.connection_name = connection_name

    def __enter__(self):
        
        set_active_db(self.connection_name)
        return self

    def __exit__(self, *exc):
        set_active_db(None)
        return False

for my serilizers.py

from .models import AdminMeterLake
from rest_framework import request, serializers

class AdminMeterLakeSerializer(serializers.ModelSerializer):

    class Meta:
        model = AdminMeterLake
        fields = ['testa', 'testb', 'testc']

for my views.py, please note I did not include all my imports

from config.context import db

class AdminView(
    mixins.ListModelMixin, 
    mixins.CreateModelMixin,
    generics.GenericAPIView):
    serializer_class = AdminMeterLakeSerializer
    queryset = AdminMeterLake.objects.using('testdb').all()

    def get(self, request, *args, **kwargs):
        return self.list(request, *args, **kwargs)

    def post(self, request, *args, **kwargs):
        with db('testdb'):
            serializer = AdminMeterLakeSerializer(data=request.data)
            if serializer.is_valid():
                serializer.save()
                return Response(serializer.data)
            return Response(serializer.errors)