1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
|
from asgiref.local import Local
from django.conf import settings as django_settings
from django.utils.functional import cached_property
class ConnectionProxy:
"""Proxy for accessing a connection object's attributes."""
def __init__(self, connections, alias):
self.__dict__['_connections'] = connections
self.__dict__['_alias'] = alias
def __getattr__(self, item):
return getattr(self._connections[self._alias], item)
def __setattr__(self, name, value):
return setattr(self._connections[self._alias], name, value)
def __delattr__(self, name):
return delattr(self._connections[self._alias], name)
def __contains__(self, key):
return key in self._connections[self._alias]
def __eq__(self, other):
return self._connections[self._alias] == other
class ConnectionDoesNotExist(Exception):
pass
class BaseConnectionHandler:
settings_name = None
exception_class = ConnectionDoesNotExist
thread_critical = False
def __init__(self, settings=None):
self._settings = settings
self._connections = Local(self.thread_critical)
@cached_property
def settings(self):
self._settings = self.configure_settings(self._settings)
return self._settings
def configure_settings(self, settings):
if settings is None:
settings = getattr(django_settings, self.settings_name)
return settings
def create_connection(self, alias):
raise NotImplementedError('Subclasses must implement create_connection().')
def __getitem__(self, alias):
try:
return getattr(self._connections, alias)
except AttributeError:
if alias not in self.settings:
raise self.exception_class(f"The connection '{alias}' doesn't exist.")
conn = self.create_connection(alias)
setattr(self._connections, alias, conn)
return conn
def __setitem__(self, key, value):
setattr(self._connections, key, value)
def __delitem__(self, key):
delattr(self._connections, key)
def __iter__(self):
return iter(self.settings)
def all(self):
return [self[alias] for alias in self]
|
