File: runworker.py

package info (click to toggle)
python-django-channels 4.3.1-1
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 1,036 kB
  • sloc: python: 3,109; makefile: 155; javascript: 60; sh: 8
file content (45 lines) | stat: -rw-r--r-- 1,593 bytes parent folder | download | duplicates (2)
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
import logging

from django.core.management import BaseCommand, CommandError

from channels import DEFAULT_CHANNEL_LAYER
from channels.layers import get_channel_layer
from channels.routing import get_default_application
from channels.worker import Worker

logger = logging.getLogger("django.channels.worker")


class Command(BaseCommand):
    leave_locale_alone = True
    worker_class = Worker

    def add_arguments(self, parser):
        super(Command, self).add_arguments(parser)
        parser.add_argument(
            "--layer",
            action="store",
            dest="layer",
            default=DEFAULT_CHANNEL_LAYER,
            help="Channel layer alias to use, if not the default.",
        )
        parser.add_argument("channels", nargs="+", help="Channels to listen on.")

    def handle(self, *args, **options):
        # Get the backend to use
        self.verbosity = options.get("verbosity", 1)
        # Get the channel layer they asked for (or see if one isn't configured)
        if "layer" in options:
            self.channel_layer = get_channel_layer(options["layer"])
        else:
            self.channel_layer = get_channel_layer()
        if self.channel_layer is None:
            raise CommandError("You do not have any CHANNEL_LAYERS configured.")
        # Run the worker
        logger.info("Running worker for channels %s", options["channels"])
        worker = self.worker_class(
            application=get_default_application(),
            channels=options["channels"],
            channel_layer=self.channel_layer,
        )
        worker.run()