[Python] 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()

Continue reading...