Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.10/site-packages/django/db/utils.py: 34%

1import pkgutil 

2from importlib import import_module 

3 

4from django.conf import settings 

5from django.core.exceptions import ImproperlyConfigured 

6 

7# For backwards compatibility with Django < 3.2 

8from django.utils.connection import ConnectionDoesNotExist # NOQA: F401 

9from django.utils.connection import BaseConnectionHandler 

10from django.utils.functional import cached_property 

11from django.utils.module_loading import import_string 

12 

13DEFAULT_DB_ALIAS = "default" 

14DJANGO_VERSION_PICKLE_KEY = "_django_version" 

15 

16 

17class Error(Exception): 

18 pass 

19 

20 

21class InterfaceError(Error): 

22 pass 

23 

24 

25class DatabaseError(Error): 

26 pass 

27 

28 

29class DataError(DatabaseError): 

30 pass 

31 

32 

33class OperationalError(DatabaseError): 

34 pass 

35 

36 

37class IntegrityError(DatabaseError): 

38 pass 

39 

40 

41class InternalError(DatabaseError): 

42 pass 

43 

44 

45class ProgrammingError(DatabaseError): 

46 pass 

47 

48 

49class NotSupportedError(DatabaseError): 

50 pass 

51 

52 

53class DatabaseErrorWrapper: 

54 """ 

55 Context manager and decorator that reraises backend-specific database 

56 exceptions using Django's common wrappers. 

57 """ 

58 

59 def __init__(self, wrapper): 

60 """ 

61 wrapper is a database wrapper. 

62 

63 It must have a Database attribute defining PEP-249 exceptions. 

64 """ 

65 self.wrapper = wrapper 

66 

67 def __enter__(self): 

68 pass 

69 

70 def __exit__(self, exc_type, exc_value, traceback): 

71 if exc_type is None: 

72 return 

73 for dj_exc_type in ( 

74 DataError, 

75 OperationalError, 

76 IntegrityError, 

77 InternalError, 

78 ProgrammingError, 

79 NotSupportedError, 

80 DatabaseError, 

81 InterfaceError, 

82 Error, 

83 ): 

84 db_exc_type = getattr(self.wrapper.Database, dj_exc_type.__name__) 

85 if issubclass(exc_type, db_exc_type): 

86 dj_exc_value = dj_exc_type(*exc_value.args) 

87 # Only set the 'errors_occurred' flag for errors that may make 

88 # the connection unusable. 

89 if dj_exc_type not in (DataError, IntegrityError): 

90 self.wrapper.errors_occurred = True 

91 raise dj_exc_value.with_traceback(traceback) from exc_value 

92 

93 def __call__(self, func): 

94 # Note that we are intentionally not using @wraps here for performance 

95 # reasons. Refs #21109. 

96 def inner(*args, **kwargs): 

97 with self: 

98 return func(*args, **kwargs) 

99 

100 return inner 

101 

102 

103def load_backend(backend_name): 

104 """ 

105 Return a database backend's "base" module given a fully qualified database 

106 backend name, or raise an error if it doesn't exist. 

107 """ 

108 # This backend was renamed in Django 1.9. 

109 if backend_name == "django.db.backends.postgresql_psycopg2": 

110 backend_name = "django.db.backends.postgresql" 

111 

112 try: 

113 return import_module("%s.base" % backend_name) 

114 except ImportError as e_user: 

115 # The database backend wasn't found. Display a helpful error message 

116 # listing all built-in database backends. 

117 import django.db.backends 

118 

119 builtin_backends = [ 

120 name 

121 for _, name, ispkg in pkgutil.iter_modules(django.db.backends.__path__) 

122 if ispkg and name not in {"base", "dummy"} 

123 ] 

124 if backend_name not in ["django.db.backends.%s" % b for b in builtin_backends]: 

125 backend_reprs = map(repr, sorted(builtin_backends)) 

126 raise ImproperlyConfigured( 

127 "%r isn't an available database backend or couldn't be " 

128 "imported. Check the above exception. To use one of the " 

129 "built-in backends, use 'django.db.backends.XXX', where XXX " 

130 "is one of:\n" 

131 " %s" % (backend_name, ", ".join(backend_reprs)) 

132 ) from e_user 

133 else: 

134 # If there's some other error, this must be an error in Django 

135 raise 

136 

137 

138class ConnectionHandler(BaseConnectionHandler): 

139 settings_name = "DATABASES" 

140 # Connections needs to still be an actual thread local, as it's truly 

141 # thread-critical. Database backends should use @async_unsafe to protect 

142 # their code from async contexts, but this will give those contexts 

143 # separate connections in case it's needed as well. There's no cleanup 

144 # after async contexts, though, so we don't allow that if we can help it. 

145 thread_critical = True 

146 

147 def configure_settings(self, databases): 

148 databases = super().configure_settings(databases) 

149 if databases == {}: 

150 databases[DEFAULT_DB_ALIAS] = {"ENGINE": "django.db.backends.dummy"} 

151 elif DEFAULT_DB_ALIAS not in databases: 

152 raise ImproperlyConfigured( 

153 f"You must define a '{DEFAULT_DB_ALIAS}' database." 

154 ) 

155 elif databases[DEFAULT_DB_ALIAS] == {}: 

156 databases[DEFAULT_DB_ALIAS]["ENGINE"] = "django.db.backends.dummy" 

157 

158 # Configure default settings. 

159 for conn in databases.values(): 

160 conn.setdefault("ATOMIC_REQUESTS", False) 

161 conn.setdefault("AUTOCOMMIT", True) 

162 conn.setdefault("ENGINE", "django.db.backends.dummy") 

163 if conn["ENGINE"] == "django.db.backends." or not conn["ENGINE"]: 

164 conn["ENGINE"] = "django.db.backends.dummy" 

165 conn.setdefault("CONN_MAX_AGE", 0) 

166 conn.setdefault("CONN_HEALTH_CHECKS", False) 

167 conn.setdefault("OPTIONS", {}) 

168 conn.setdefault("TIME_ZONE", None) 

169 for setting in ["NAME", "USER", "PASSWORD", "HOST", "PORT"]: 

170 conn.setdefault(setting, "") 

171 

172 test_settings = conn.setdefault("TEST", {}) 

173 default_test_settings = [ 

174 ("CHARSET", None), 

175 ("COLLATION", None), 

176 ("MIGRATE", True), 

177 ("MIRROR", None), 

178 ("NAME", None), 

179 ] 

180 for key, value in default_test_settings: 

181 test_settings.setdefault(key, value) 

182 return databases 

183 

184 @property 

185 def databases(self): 

186 # Maintained for backward compatibility as some 3rd party packages have 

187 # made use of this private API in the past. It is no longer used within 

188 # Django itself. 

189 return self.settings 

190 

191 def create_connection(self, alias): 

192 db = self.settings[alias] 

193 backend = load_backend(db["ENGINE"]) 

194 return backend.DatabaseWrapper(db, alias) 

195 

196 

197class ConnectionRouter: 

198 def __init__(self, routers=None): 

199 """ 

200 If routers is not specified, default to settings.DATABASE_ROUTERS. 

201 """ 

202 self._routers = routers 

203 

204 @cached_property 

205 def routers(self): 

206 if self._routers is None: 

207 self._routers = settings.DATABASE_ROUTERS 

208 routers = [] 

209 for r in self._routers: 

210 if isinstance(r, str): 

211 router = import_string(r)() 

212 else: 

213 router = r 

214 routers.append(router) 

215 return routers 

216 

217 def _router_func(action): 

218 def _route_db(self, model, **hints): 

219 chosen_db = None 

220 for router in self.routers: 

221 try: 

222 method = getattr(router, action) 

223 except AttributeError: 

224 # If the router doesn't have a method, skip to the next one. 

225 pass 

226 else: 

227 chosen_db = method(model, **hints) 

228 if chosen_db: 

229 return chosen_db 

230 instance = hints.get("instance") 

231 if instance is not None and instance._state.db: 

232 return instance._state.db 

233 return DEFAULT_DB_ALIAS 

234 

235 return _route_db 

236 

237 db_for_read = _router_func("db_for_read") 

238 db_for_write = _router_func("db_for_write") 

239 

240 def allow_relation(self, obj1, obj2, **hints): 

241 for router in self.routers: 

242 try: 

243 method = router.allow_relation 

244 except AttributeError: 

245 # If the router doesn't have a method, skip to the next one. 

246 pass 

247 else: 

248 allow = method(obj1, obj2, **hints) 

249 if allow is not None: 

250 return allow 

251 return obj1._state.db == obj2._state.db 

252 

253 def allow_migrate(self, db, app_label, **hints): 

254 for router in self.routers: 

255 try: 

256 method = router.allow_migrate 

257 except AttributeError: 

258 # If the router doesn't have a method, skip to the next one. 

259 continue 

260 

261 allow = method(db, app_label, **hints) 

262 

263 if allow is not None: 

264 return allow 

265 return True 

266 

267 def allow_migrate_model(self, db, model): 

268 return self.allow_migrate( 

269 db, 

270 model._meta.app_label, 

271 model_name=model._meta.model_name, 

272 model=model, 

273 ) 

274 

275 def get_migratable_models(self, app_config, db, include_auto_created=False): 

276 """Return app models allowed to be migrated on provided db.""" 

277 models = app_config.get_models(include_auto_created=include_auto_created) 

278 return [model for model in models if self.allow_migrate_model(db, model)]