Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/django/db/utils.py: 67%
169 statements
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
1import pkgutil
2from importlib import import_module
4from django.conf import settings
5from django.core.exceptions import ImproperlyConfigured
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
13DEFAULT_DB_ALIAS = "default"
14DJANGO_VERSION_PICKLE_KEY = "_django_version"
17class Error(Exception):
18 pass
21class InterfaceError(Error):
22 pass
25class DatabaseError(Error):
26 pass
29class DataError(DatabaseError):
30 pass
33class OperationalError(DatabaseError):
34 pass
37class IntegrityError(DatabaseError):
38 pass
41class InternalError(DatabaseError):
42 pass
45class ProgrammingError(DatabaseError):
46 pass
49class NotSupportedError(DatabaseError):
50 pass
53class DatabaseErrorWrapper:
54 """
55 Context manager and decorator that reraises backend-specific database
56 exceptions using Django's common wrappers.
57 """
59 def __init__(self, wrapper):
60 """
61 wrapper is a database wrapper.
63 It must have a Database attribute defining PEP-249 exceptions.
64 """
65 self.wrapper = wrapper
67 def __enter__(self):
68 pass
70 def __exit__(self, exc_type, exc_value, traceback):
71 if exc_type is None: 71 ↛ 73line 71 didn't jump to line 73, because the condition on line 71 was never false
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
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)
100 return inner
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": 109 ↛ 110line 109 didn't jump to line 110, because the condition on line 109 was never true
110 backend_name = "django.db.backends.postgresql"
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
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
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
147 def configure_settings(self, databases):
148 databases = super().configure_settings(databases)
149 if databases == {}: 149 ↛ 150line 149 didn't jump to line 150, because the condition on line 149 was never true
150 databases[DEFAULT_DB_ALIAS] = {"ENGINE": "django.db.backends.dummy"}
151 elif DEFAULT_DB_ALIAS not in databases: 151 ↛ 152line 151 didn't jump to line 152, because the condition on line 151 was never true
152 raise ImproperlyConfigured(
153 f"You must define a '{DEFAULT_DB_ALIAS}' database."
154 )
155 elif databases[DEFAULT_DB_ALIAS] == {}: 155 ↛ 156line 155 didn't jump to line 156, because the condition on line 155 was never true
156 databases[DEFAULT_DB_ALIAS]["ENGINE"] = "django.db.backends.dummy"
157 return databases
159 @property
160 def databases(self):
161 return self.settings
163 def ensure_defaults(self, alias):
164 """
165 Put the defaults into the settings dictionary for a given connection
166 where no settings is provided.
167 """
168 try:
169 conn = self.databases[alias]
170 except KeyError:
171 raise self.exception_class(f"The connection '{alias}' doesn't exist.")
173 conn.setdefault("ATOMIC_REQUESTS", False)
174 conn.setdefault("AUTOCOMMIT", True)
175 conn.setdefault("ENGINE", "django.db.backends.dummy")
176 if conn["ENGINE"] == "django.db.backends." or not conn["ENGINE"]: 176 ↛ 177line 176 didn't jump to line 177, because the condition on line 176 was never true
177 conn["ENGINE"] = "django.db.backends.dummy"
178 conn.setdefault("CONN_MAX_AGE", 0)
179 conn.setdefault("OPTIONS", {})
180 conn.setdefault("TIME_ZONE", None)
181 for setting in ["NAME", "USER", "PASSWORD", "HOST", "PORT"]:
182 conn.setdefault(setting, "")
184 def prepare_test_settings(self, alias):
185 """
186 Make sure the test settings are available in the 'TEST' sub-dictionary.
187 """
188 try:
189 conn = self.databases[alias]
190 except KeyError:
191 raise self.exception_class(f"The connection '{alias}' doesn't exist.")
193 test_settings = conn.setdefault("TEST", {})
194 default_test_settings = [
195 ("CHARSET", None),
196 ("COLLATION", None),
197 ("MIGRATE", True),
198 ("MIRROR", None),
199 ("NAME", None),
200 ]
201 for key, value in default_test_settings:
202 test_settings.setdefault(key, value)
204 def create_connection(self, alias):
205 self.ensure_defaults(alias)
206 self.prepare_test_settings(alias)
207 db = self.databases[alias]
208 backend = load_backend(db["ENGINE"])
209 return backend.DatabaseWrapper(db, alias)
211 def close_all(self):
212 for alias in self:
213 try:
214 connection = getattr(self._connections, alias)
215 except AttributeError:
216 continue
217 connection.close()
220class ConnectionRouter:
221 def __init__(self, routers=None):
222 """
223 If routers is not specified, default to settings.DATABASE_ROUTERS.
224 """
225 self._routers = routers
227 @cached_property
228 def routers(self):
229 if self._routers is None: 229 ↛ 231line 229 didn't jump to line 231, because the condition on line 229 was never false
230 self._routers = settings.DATABASE_ROUTERS
231 routers = []
232 for r in self._routers: 232 ↛ 233line 232 didn't jump to line 233, because the loop on line 232 never started
233 if isinstance(r, str):
234 router = import_string(r)()
235 else:
236 router = r
237 routers.append(router)
238 return routers
240 def _router_func(action):
241 def _route_db(self, model, **hints):
242 chosen_db = None
243 for router in self.routers: 243 ↛ 244line 243 didn't jump to line 244, because the loop on line 243 never started
244 try:
245 method = getattr(router, action)
246 except AttributeError:
247 # If the router doesn't have a method, skip to the next one.
248 pass
249 else:
250 chosen_db = method(model, **hints)
251 if chosen_db:
252 return chosen_db
253 instance = hints.get("instance")
254 if instance is not None and instance._state.db:
255 return instance._state.db
256 return DEFAULT_DB_ALIAS
258 return _route_db
260 db_for_read = _router_func("db_for_read")
261 db_for_write = _router_func("db_for_write")
263 def allow_relation(self, obj1, obj2, **hints):
264 for router in self.routers: 264 ↛ 265line 264 didn't jump to line 265, because the loop on line 264 never started
265 try:
266 method = router.allow_relation
267 except AttributeError:
268 # If the router doesn't have a method, skip to the next one.
269 pass
270 else:
271 allow = method(obj1, obj2, **hints)
272 if allow is not None:
273 return allow
274 return obj1._state.db == obj2._state.db
276 def allow_migrate(self, db, app_label, **hints):
277 for router in self.routers: 277 ↛ 278line 277 didn't jump to line 278, because the loop on line 277 never started
278 try:
279 method = router.allow_migrate
280 except AttributeError:
281 # If the router doesn't have a method, skip to the next one.
282 continue
284 allow = method(db, app_label, **hints)
286 if allow is not None:
287 return allow
288 return True
290 def allow_migrate_model(self, db, model):
291 return self.allow_migrate(
292 db,
293 model._meta.app_label,
294 model_name=model._meta.model_name,
295 model=model,
296 )
298 def get_migratable_models(self, app_config, db, include_auto_created=False):
299 """Return app models allowed to be migrated on provided db."""
300 models = app_config.get_models(include_auto_created=include_auto_created)
301 return [model for model in models if self.allow_migrate_model(db, model)]