/usr/lib/python3/dist-packages/channels/asgi.py is in python3-django-channels 1.1.8.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | from __future__ import unicode_literals
import django
from django.conf import settings
from django.utils.module_loading import import_string
from .routing import Router
from .utils import name_that_thing
class InvalidChannelLayerError(ValueError):
pass
class ChannelLayerManager(object):
"""
Takes a settings dictionary of backends and initialises them on request.
"""
def __init__(self):
self.backends = {}
@property
def configs(self):
# Lazy load settings so we can be imported
return getattr(settings, "CHANNEL_LAYERS", {})
def make_backend(self, name):
"""
Instantiate channel layer.
"""
config = self.configs[name].get("CONFIG", {})
return self._make_backend(name, config)
def make_test_backend(self, name):
"""
Instantiate channel layer using its test config.
"""
try:
config = self.configs[name]["TEST_CONFIG"]
except KeyError:
raise InvalidChannelLayerError("No TEST_CONFIG specified for %s" % name)
return self._make_backend(name, config)
def _make_backend(self, name, config):
# Load the backend class
try:
backend_class = import_string(self.configs[name]['BACKEND'])
except KeyError:
raise InvalidChannelLayerError("No BACKEND specified for %s" % name)
except ImportError:
raise InvalidChannelLayerError(
"Cannot import BACKEND %r specified for %s" % (self.configs[name]['BACKEND'], name)
)
# Get routing
try:
routing = self.configs[name]['ROUTING']
except KeyError:
raise InvalidChannelLayerError("No ROUTING specified for %s" % name)
# Initialise and pass config
asgi_layer = backend_class(**config)
return ChannelLayerWrapper(
channel_layer=asgi_layer,
alias=name,
routing=routing,
)
def __getitem__(self, key):
if key not in self.backends:
self.backends[key] = self.make_backend(key)
return self.backends[key]
def __contains__(self, key):
return key in self.configs
def set(self, key, layer):
"""
Sets an alias to point to a new ChannelLayerWrapper instance, and
returns the old one that it replaced. Useful for swapping out the
backend during tests.
"""
old = self.backends.get(key, None)
self.backends[key] = layer
return old
class ChannelLayerWrapper(object):
"""
Top level channel layer wrapper, which contains both the ASGI channel
layer object as well as alias and routing information specific to Django.
"""
def __init__(self, channel_layer, alias, routing):
self.channel_layer = channel_layer
self.alias = alias
self.routing = routing
self.router = Router(self.routing)
def __getattr__(self, name):
return getattr(self.channel_layer, name)
def __str__(self):
return "%s (%s)" % (self.alias, name_that_thing(self.channel_layer))
def local_only(self):
# TODO: Can probably come up with a nicer check?
return "inmemory" in self.channel_layer.__class__.__module__
def get_channel_layer(alias="default"):
"""
Returns the raw ASGI channel layer for this project.
"""
if django.VERSION[1] > 9:
django.setup(set_prefix=False)
else:
django.setup()
return channel_layers[alias].channel_layer
# Default global instance of the channel layer manager
channel_layers = ChannelLayerManager()
|