Viewing file:  connection.py (2.19 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
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]
|