是否可以从应用程序创建和连接 Django 数据库?
Is it possible to create and connect Django databases from the application?
是否可以从应用程序创建和连接 Django 数据库?
一个例子。
在用户注册时的系统中,django create a new database 会生成具有确切设置的迁移和创建新数据库。
DATABASE_ROUTERS = ['base.routers.routers.TenantRouter']
MIDDLEWARE = [
...
'base.middleware.middleware.Multidb',
...
]
def Multidb(get_response):
def middleware(request):
try:
usuario = request.user
empresa = usuario.empresa
except:
empresa = request.GET['empresa']
@thread_local(using_db=empresa)
def execute_request(request):
return get_response(request)
response = execute_request(request)
return response
return middleware
class TenantRouter(object):
def db_for_read(self, model, **hints):
return get_thread_local('using_db', 'default')
def db_for_write(self, model, **hints):
return get_thread_local('using_db', 'default')
def allow_relation(self, obj1, obj2, **hints):
return True
def allow_migrate(self, db, app_label, model_name=None, **hints):
return True
class thread_local(object):
""" a decorator that wraps a function in a thread local definition block
useful for passing variables down the stack w/o actually passing them
examples: what database to read from, whether to cache queries, etc
adapted from django.test.utils.override_settings
Usage:
@thread_local(SITE_NAME_SHORT='foobar')
def override(request):
...
"""
def __init__(self, **kwargs):
self.options = kwargs
def __enter__(self):
for attr, value in self.options.items():
setattr(threadlocal, attr, value)
def __exit__(self, exc_type, exc_value, traceback):
for attr in self.options.keys():
setattr(threadlocal, attr, None)
def __call__(self, test_func):
@wraps(test_func)
def inner(*args, **kwargs):
# the thread_local class is also a context manager
# which means it will call __enter__ and __exit__
with self:
return test_func(*args, **kwargs)
return inner
def get_thread_local(attr, default=None):
""" use this method from lower in the stack to get the value """
return getattr(threadlocal, attr, default)
是否可以从应用程序创建和连接 Django 数据库?
一个例子。
在用户注册时的系统中,django create a new database 会生成具有确切设置的迁移和创建新数据库。
DATABASE_ROUTERS = ['base.routers.routers.TenantRouter']
MIDDLEWARE = [
...
'base.middleware.middleware.Multidb',
...
]
def Multidb(get_response):
def middleware(request):
try:
usuario = request.user
empresa = usuario.empresa
except:
empresa = request.GET['empresa']
@thread_local(using_db=empresa)
def execute_request(request):
return get_response(request)
response = execute_request(request)
return response
return middleware
class TenantRouter(object):
def db_for_read(self, model, **hints):
return get_thread_local('using_db', 'default')
def db_for_write(self, model, **hints):
return get_thread_local('using_db', 'default')
def allow_relation(self, obj1, obj2, **hints):
return True
def allow_migrate(self, db, app_label, model_name=None, **hints):
return True
class thread_local(object):
""" a decorator that wraps a function in a thread local definition block
useful for passing variables down the stack w/o actually passing them
examples: what database to read from, whether to cache queries, etc
adapted from django.test.utils.override_settings
Usage:
@thread_local(SITE_NAME_SHORT='foobar')
def override(request):
...
"""
def __init__(self, **kwargs):
self.options = kwargs
def __enter__(self):
for attr, value in self.options.items():
setattr(threadlocal, attr, value)
def __exit__(self, exc_type, exc_value, traceback):
for attr in self.options.keys():
setattr(threadlocal, attr, None)
def __call__(self, test_func):
@wraps(test_func)
def inner(*args, **kwargs):
# the thread_local class is also a context manager
# which means it will call __enter__ and __exit__
with self:
return test_func(*args, **kwargs)
return inner
def get_thread_local(attr, default=None):
""" use this method from lower in the stack to get the value """
return getattr(threadlocal, attr, default)