例如,我有一个项目(mysite),其中包含两个应用程序(myapp、myapp2)。这些应用程序使用不同的数据库:
settings.py:
在每个应用程序中添加一个模型:
myapp/models.py:
它表明Django试图在
settings.py:
DATABASES = {
"default": {
"ENGINE": "django.db.backends.postgresql",
"HOST": "127.0.0.1",
"USER": "postgres",
"PASSWORD": "root",
"PORT": "5432",
"NAME": "test",
},
"test": {
"ENGINE": "django.db.backends.mysql",
"HOST": "127.0.0.1",
"USER": "root",
"PASSWORD": "root",
"PORT": "3306",
"NAME": "test",
},
}
接下来我编写了一个数据库路由器:
from django.conf import settings
class AppRouter:
def db_for_read(self, model, **hints):
return settings.APP_DB_MAPPER.get(model._meta.app_label)
def db_for_write(self, model, **hints):
return settings.APP_DB_MAPPER.get(model._meta.app_label)
def allow_relation(self, obj1, obj2, **hints):
db1 = settings.APP_DB_MAPPER.get(obj1._meta.app_label)
db2 = settings.APP_DB_MAPPER.get(obj2._meta.app_label)
if db1 and db2:
if db1 == db2:
return True
else:
return False
return None
def allow_migrate(self, db, app_label, model_name=None, **hints):
if app_label in settings.APP_DB_MAPPER:
return db == settings.APP_DB_MAPPER[app_label]
return None
然后应用它。
settings.py:
DATABASE_ROUTERS = ["mysite.app_router.AppRouter"]
APP_DB_MAPPER = {"myapp": "default", "myapp2": "test"}
在每个应用程序中添加一个模型:
myapp/models.py:
from django.db import models
from django.contrib.auth.models import User
class TestDefault(models.Model):
name = models.CharField(max_length=100)
user = models.OneToOneField(User, on_delete=models.SET_NULL, null=True)
myapp2/models.py:
from django.db import models
from django.contrib.auth.models import User
class TestApp(models.Model):
name = models.CharField(max_length=100)
user = models.OneToOneField(User, on_delete=models.SET_NULL, null=True)
迁移完成后,会创建以下表:
mysql("test" 数据库):
+----------------------------+
| Tables_in_test |
+----------------------------+
| auth_group |
| auth_group_permissions |
| auth_permission |
| auth_user |
| auth_user_groups |
| auth_user_user_permissions |
| django_admin_log |
| django_content_type |
| django_migrations |
| django_session |
| myapp2_testapp |
+----------------------------+
postgres("default" db):
+----------+----------------------------+--------+----------+
| Schema | Name | Type | Owner |
|----------+----------------------------+--------+----------|
| public | auth_group | table | postgres |
| public | auth_group_permissions | table | postgres |
| public | auth_permission | table | postgres |
| public | auth_user | table | postgres |
| public | auth_user_groups | table | postgres |
| public | auth_user_user_permissions | table | postgres |
| public | django_admin_log | table | postgres |
| public | django_content_type | table | postgres |
| public | django_migrations | table | postgres |
| public | django_session | table | postgres |
| public | myapp_testdefault | table | postgres |
+----------+----------------------------+--------+----------+
然后我试图在shell中测试关系:
from mysite.myapp.models import TestDefault
from mysite.myapp2.models import TestApp
from django.contrib.auth.models import User
user = User.objects.create(username="1") # in default db
TestDefault.objects.create(name="123",user=user)
user = User.objects.using("test").create(username="1") # in test db
TestApp.objects.create(name="123",user=user)
# then try to delete user in test db
user.delete() # error!
追踪回溯:
---------------------------------------------------------------------------
ProgrammingError Traceback (most recent call last)
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/django/db/backends/utils.py in _execute(self, sql, params, *ignored_wrapper_args)
83 else:
---> 84 return self.cursor.execute(sql, params)
85
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/django/db/backends/mysql/base.py in execute(self, query, args)
70 # args is None means no string interpolation
---> 71 return self.cursor.execute(query, args)
72 except Database.OperationalError as e:
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/MySQLdb/cursors.py in execute(self, query, args)
208 assert isinstance(query, (bytes, bytearray))
--> 209 res = self._query(query)
210 return res
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/MySQLdb/cursors.py in _query(self, q)
314 self._result = None
--> 315 db.query(q)
316 self._do_get_result(db)
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/MySQLdb/connections.py in query(self, query)
225 query = bytes(query)
--> 226 _mysql.connection.query(self, query)
227
ProgrammingError: (1146, "Table 'test.myapp_testdefault' doesn't exist")
The above exception was the direct cause of the following exception:
ProgrammingError Traceback (most recent call last)
<ipython-input-22-da112720c459> in <module>
----> 1 user.delete()
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/django/db/models/base.py in delete(self, using, keep_parents)
916
917 collector = Collector(using=using)
--> 918 collector.collect([self], keep_parents=keep_parents)
919 return collector.delete()
920
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/django/db/models/deletion.py in collect(self, objs, source, nullable, collect_related, source_attr, reverse_dependency, keep_parents)
221 if self.can_fast_delete(sub_objs, from_field=field):
222 self.fast_deletes.append(sub_objs)
--> 223 elif sub_objs:
224 field.remote_field.on_delete(self, field, sub_objs, self.using)
225 for field in model._meta.private_fields:
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/django/db/models/query.py in __bool__(self)
276
277 def __bool__(self):
--> 278 self._fetch_all()
279 return bool(self._result_cache)
280
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/django/db/models/query.py in _fetch_all(self)
1240 def _fetch_all(self):
1241 if self._result_cache is None:
-> 1242 self._result_cache = list(self._iterable_class(self))
1243 if self._prefetch_related_lookups and not self._prefetch_done:
1244 self._prefetch_related_objects()
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/django/db/models/query.py in __iter__(self)
53 # Execute the query. This will also fill compiler.select, klass_info,
54 # and annotations.
---> 55 results = compiler.execute_sql(chunked_fetch=self.chunked_fetch, chunk_size=self.chunk_size)
56 select, klass_info, annotation_col_map = (compiler.select, compiler.klass_info,
57 compiler.annotation_col_map)
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/django/db/models/sql/compiler.py in execute_sql(self, result_type, chunked_fetch, chunk_size)
1098 cursor = self.connection.cursor()
1099 try:
-> 1100 cursor.execute(sql, params)
1101 except Exception:
1102 # Might fail for server-side cursors (e.g. connection closed)
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/django/db/backends/utils.py in execute(self, sql, params)
97 start = time()
98 try:
---> 99 return super().execute(sql, params)
100 finally:
101 stop = time()
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/django/db/backends/utils.py in execute(self, sql, params)
65
66 def execute(self, sql, params=None):
---> 67 return self._execute_with_wrappers(sql, params, many=False, executor=self._execute)
68
69 def executemany(self, sql, param_list):
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/django/db/backends/utils.py in _execute_with_wrappers(self, sql, params, many, executor)
74 for wrapper in reversed(self.db.execute_wrappers):
75 executor = functools.partial(wrapper, executor)
---> 76 return executor(sql, params, many, context)
77
78 def _execute(self, sql, params, *ignored_wrapper_args):
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/django/db/backends/utils.py in _execute(self, sql, params, *ignored_wrapper_args)
82 return self.cursor.execute(sql)
83 else:
---> 84 return self.cursor.execute(sql, params)
85
86 def _executemany(self, sql, param_list, *ignored_wrapper_args):
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/django/db/utils.py in __exit__(self, exc_type, exc_value, traceback)
87 if dj_exc_type not in (DataError, IntegrityError):
88 self.wrapper.errors_occurred = True
---> 89 raise dj_exc_value.with_traceback(traceback) from exc_value
90
91 def __call__(self, func):
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/django/db/backends/utils.py in _execute(self, sql, params, *ignored_wrapper_args)
82 return self.cursor.execute(sql)
83 else:
---> 84 return self.cursor.execute(sql, params)
85
86 def _executemany(self, sql, param_list, *ignored_wrapper_args):
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/django/db/backends/mysql/base.py in execute(self, query, args)
69 try:
70 # args is None means no string interpolation
---> 71 return self.cursor.execute(query, args)
72 except Database.OperationalError as e:
73 # Map some error codes to IntegrityError, since they seem to be
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/MySQLdb/cursors.py in execute(self, query, args)
207
208 assert isinstance(query, (bytes, bytearray))
--> 209 res = self._query(query)
210 return res
211
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/MySQLdb/cursors.py in _query(self, q)
313 db = self._get_db()
314 self._result = None
--> 315 db.query(q)
316 self._do_get_result(db)
317 self._post_get_result()
~/projects/learn-some/learn-some-django/test/.venv/lib/python3.7/site-packages/MySQLdb/connections.py in query(self, query)
224 if isinstance(query, bytearray):
225 query = bytes(query)
--> 226 _mysql.connection.query(self, query)
227
228 def _bytes_literal(self, bs):
ProgrammingError: (1146, "Table 'test.myapp_testdefault' doesn't exist")
它表明Django试图在
test
数据库中查找testdefault
表中的记录。这是不可能的。为什么它会这样做?在这种情况下删除用户的正确方法是什么?
Model.delete
可以解决这个问题。但是文档中提到,Django应该删除正确的记录,这可能是Django中的一个bug。几年过去了,我已经不再从事这方面的工作。我有时间时会深入研究一下。 - Guo Jiaqi