Channels: DataBinding's signal receiver not fired while model saved in celery task

[nixoz]

Hello there.

I have found some interesting issue.

1. Created a model - nothing special just basic model
2. Created a DataBinding

class ServiceDataBinding(WebsocketBinding):

    model = Service
    stream = "service.updates"
    fields = ["__all__"]

    @classmethod
    def group_names(cls, *args, **kwargs):
        print("group_names_called")
        return ["main_group"]

    def has_permission(self, user, action, pk):
        return True

1. Adding/Updating/Deleting model instances through Django admin or shell - everything works as expected, binding receivers are fired nicely and works pretty well.

2. Created an async celery task, where I modify the model field and calling model's save() method. Here the task itself

@app.task
def get_services_for_server(server_address, server_key):

    if not server_address:
        raise Exception("Server address should be provided")

    if not server_key:
        raise Exception("Server key should be provided")

    service_data = []

    command = "docker ps -a --format \"{{json .}}\""
    exit_code, stdout, stderr = execute_remote_command(command, server_address, server_key)

    if exit_code is 0:
        for line in stdout.splitlines():
            service_data.append(json.loads(line))

    try:
        if len(service_data) > 0:
            containers = [container["Names"] for container in service_data]
            services = Service.objects.filter(container_name__in=containers, server__address=server_address)

            with transaction.atomic():
                for service in services:
                    container_name = service.container_name
                    container_stats = find_container_stats(service_data, container_name)
                    if service.container_status is not container_stats["Status"]:
                        service.container_status = container_stats["Status"]
                        service.save()

    except Exception as ex:
        logger.error("Error processing data from {}: {}".format(server_address, ex))

1. On task execution the binding's signal receivers not fired for some reason.

2. For debugging purposes, I have created signal receiver in signals.py(registered it with django's @receiver decorator) and imported signals.py in AppConfig ready() method.

3. The independently registered receiver is fired normally on every model instance save() call within the celery task.

Maybe I just missing something or just stupid, but curious to figure out - why this happens this way? I have checked the post_save receivers list and here what's inside:

[((140609553948320, 32316248), <weakref at 0x7fe23683e958; to 'function' at 0x7fe23683dea0 (print_on_receive)>), (((140609444686472, 140609620961208), 32316248), <weakref at 0x7fe23685f198; to 'BindingMetaclass' at 0x7fe23000aa88 (ServiceDataBinding)>)]

So, the receivers are here but fired only the "print_on_receive"...why another is not fired? I have no idea clearly.

Can you guys help me with this?

Darwin (max os 10.12.6)
Django==2.0.1
asgi-redis==1.4.3
asgiref==1.1.2
celery==4.1.0
channels==1.1.8
daphne==1.4.2

Thank you.

[andrewgodwin]

I'm afraid stepping through and debugging the binding receivers in a case like this is not something I have the time for (and we're removing databinding in the next version), so I'm not sure I'll be able to help you here. I presume it works not in a Celery task?

[andrewgodwin]

Also, we just released Channels 2 which doesn't have databinding, so I'm going to close this as an open ticket/

[mattyg]

For anyone still interested in this issue. I was able to work around it by creating a celery task to register the model event bindings that runs at the celery signal celeryd_init.connect.

@celeryd_init.connect
def register_model_bindings(**kwargs):
    bindings = [WalletBinding, TradeBinding, TransferBinding]

    for binding in bindings:
        model = binding.model
        pre_save.connect(binding.pre_save_receiver, sender=model)
        post_save.connect(binding.post_save_receiver, sender=model)
        pre_delete.connect(binding.pre_delete_receiver, sender=model)
        post_delete.connect(binding.post_delete_receiver, sender=model)