Dynamically instantiate database django

In a django web application, when logging in I am saving the name of the database in the session. I need a method to dynamically instantiate the database using this name that is saved in the user session. I dont't want to use "using()" because my aplication is already done, and it would give me a lot of work, neither by configuring multiple databases in your settings (because my databases are being createded dinamically) if there is another way i would like to know. Maybe it's not possible, i understand if it's the case...

I tried middlewares, routes, threading, decorators but maybe a think i would have done something wrong. I exhausted my attempts. As far as I could go i get logged in the right database until the second screen of my application, when i am browsing in the application the 3 screen i pass the database goes back to default.

middleware:

import re
from django.utils.deprecation import MiddlewareMixin
from django.db import connections
from .thread import set_banco_academia

class DatabaseMiddleware(MiddlewareMixin):
    def process_request(self, request):
        """
        Middleware que define dinamicamente o banco de dados a ser usado
        com base na sessão do usuário e adiciona a conexão ao Django.
        """
        if not hasattr(request, "session"):  # Garante que a sessão foi carregada
            return

        # Obtém o banco armazenado na sessão ou usa "default"
        banco_academia = request.session.get("banco_academia", "default")

        print(f"🔍 Banco selecionado na sessão: {banco_academia}")

        # Valida o nome do banco para evitar SQL Injection
        if self.is_valid_database_name(banco_academia):
            set_banco_academia(banco_academia)
        else:
            set_banco_academia("default")
            banco_academia = "default"

        # Se o banco ainda não estiver em connections, adicionamos dinamicamente
        if banco_academia not in connections.databases:
            connections.databases[banco_academia] = {
                'ENGINE': 'django.db.backends.mysql',  # Ou outro engine
                'NAME': banco_academia,
                'USER': 'root',
                'PASSWORD': '',
                'HOST': 'localhost',
                'PORT': '3306',
                'ATOMIC_REQUESTS': False,
                'AUTOCOMMIT': True,
                'CONN_MAX_AGE': 0,
                'CONN_HEALTH_CHECKS': False,
                'OPTIONS': {},
                'TIME_ZONE': None,
                'TEST': {
                    'CHARSET': None,
                    'COLLATION': None,
                    'MIGRATE': True,
                    'MIRROR': None,
                    'NAME': None,
                },
            }
            print(f"🔗 Conexão adicionada dinamicamente para o banco: {banco_academia}")

    def is_valid_database_name(self, name):
        """Valida se o nome do banco é seguro e segue um padrão esperado"""
        return bool(re.match(r"^[a-zA-Z0-9_]+$", name))

threading:

import threading

_local = threading.local()

def set_banco_academia(banco):
    """Define o banco de dados na thread local"""
    _local.banco_academia = banco

def get_banco_academia():
    """Obtém o banco de dados armazenado na thread local"""
    return getattr(_local, "banco_academia", "default")

router:

from django.db import connections
from .thread import get_banco_academia

class ClubGymRouter:
    def db_for_read(self, model, **hints):
        """Define o banco de leitura dinamicamente"""
        banco_academia = get_banco_academia()
        if banco_academia in connections.databases:
            return banco_academia
        return "default"

    def db_for_write(self, model, **hints):
        """Define o banco de escrita dinamicamente"""
        banco_academia = get_banco_academia()
        if banco_academia in connections.databases:
            return banco_academia
        return "default"

    def allow_relation(self, obj1, obj2, **hints):
        """Permite relações entre objetos do mesmo banco"""
        db_set = {get_banco_academia(), "default"}
        return obj1._state.db in db_set and obj2._state.db in db_set

    def allow_migrate(self, db, app_label, model_name=None, **hints):
        """Define se a migração pode ser feita no banco correto"""
        return db == get_banco_academia() or db == "default"

this is how i save the name of the database in session

request.session["banco_academia"] = nome_banco
request.session.modified = True  # 🔥 Marca como modificada
request.session.save()

Вернуться на верх