Subscriptions hooked to model CUD operations

[thclark]

Feature Request Type

• Core functionality

Description

Using Jayden Windle's graphene-subscriptions, it's straightforward to hook model Create, Update, Delete (CUD) operations to a subscription, which is a foundational feature of using subscriptions with django.

For this to work in strawberry-django there are three pieces of that puzzle:

• ✓ strawberry itself needs to supports subscriptions (which it does, and that's well-documented)
• ✓ django needs to be set up to support subscriptions (django-strawberry documents how to set that up using channels)
• ❓ strawberry-django needs to handle model CUD events, to trigger updates for subscribers to model queries (so that updates are made as the model changes).

I'm not sure if this has been done yet with django-strawberry (perhaps there's something I can't see in the code and isn't documented yet).

Is there anyone who has achieved this who can share how they do it?

Upvote & Fund

• We're using Polar.sh so you can upvote and help fund this issue.
• We receive the funding once the issue is completed & confirmed by you.
• Thank you in advance for helping prioritize & fund our backlog.

[sdobbelaere]

I have some thoughts on the matter yes. This was actually something I've been working on inside our OneSila project.

How I solved it, from memory:
-> A model subscription class that exposes turns any model into a subscription with some light configuration
-> A django post-save signal that is attached to the subscription mechanism that pushed the change to the frontend.

This was a good fit for us, since we have a lot of backend logic and tasks that run async. But if your idea is frontend-heavy, then you could considering a similar approach except that where you complete the CUD mutation you can:
-> Trigger a custom signal
-> Catch it in the strawberry app on any model
-> Push the changes to the frontend

It's been a while since I worked on this functionality, and sadly it never made it in the lib. But as I've expressed then, I'd be happy to work together with anyone wanting to improve the subscriptions.

Long story short - give me a few days to dig into my codebase and make some notes. Happy to share what I have.

[sdobbelaere]

@thclark just had some time to go through the code.

How I solved this inside OneSila is not by hooking the subscriptions into the mutations. Instead leveraging Django's nice signal-system, I choose to hook them into the post-save signals. That's more holistic measure rather than forcing updates through CUD operations. It's just native behaviour. If you handle it outside of the Django-way you'll just limit the usability especially for applications that handle a lot of work in background tasks like huey/celery or have other integrations that push and pull information.

Once a mutation has completed, it will trigger a post-save anyway since the instance will have changed.
That part, comes with a bit of refresh code which is declared on the receiver like so:

@receiver(post_save)
def strawberry_django__refresh_subscription(sender, instance, **kwargs):
    """
    This is to be sent on the every post_save or relevant signal
    """
    try:
        refresh_subscription_receiver(instance)
    except AttributeError:
        # This is a very greedy approach.  There are many post_save signals going around in Django
        # many can fail is they are not models as we have them in the apps
        pass

Every post-save will just work. You could add the behaviour to more signals and create custom ones. But that just seems like overkill. The refresh_subscription_receiver code looks like:

from asgiref.sync import async_to_sync
from strawberry.relay import to_base64
import channels.layers

def get_group(instance):
    return f"{instance.__class__.__name__}"

def get_msg_type(instance):
    group = get_group(instance)
    return f"{group}_{instance.id}"


def get_msg(instance):
    return {'type': get_msg_type(instance)}

def refresh_subscription_receiver(instance):
    group = get_group(instance)
    msg = get_msg(instance)

    channel_layer = channels.layers.get_channel_layer()
    async_to_sync(channel_layer.group_send)(group=group, message=msg)

This would probably benefit from introspecting and updating related-models like M2M and FK relations should there be updates that trickle through.

We can now receive changes to models from the backend once subscribed to an instance.
Great. But how do we even subscribe to model instance?

This part of the puzzle was solved for our use-case by declaring the field in the Subscription class:

class Subscription:
    @subscription
    async def company(self, info: Info, pk: str) -> AsyncGenerator[CompanyType, None]:
        async for i in model_subscriber(info=info, pk=pk, model=Company):
            yield i

Personally I'd like to see more something along the lines of company: AsyncGenerator[CompanyType, None] = model_subscription_field(Company) but I haven't gotten that far yet.

So in that declaration you will find the crucial method model_subscriber which looks like:

from django.db.models import Model

from strawberry import type
from strawberry import subscription
from strawberry.relay.types import GlobalID
from strawberry.types import Info

from typing import AsyncGenerator, Any


async def model_subscriber(info: Info, pk: GlobalID, model: Model) -> AsyncGenerator[Any, None]:
    publisher = ModelInstanceSubscribePublisher(info=info, pk=pk, model=model)
    async for msg in publisher.await_messages():
        yield msg

This field wrapper relies on the actual mechanics found in the Publisher class:

from strawberry_django.auth.utils import get_current_user
from strawberry.relay.utils import from_base64
from asgiref.sync import sync_to_async
from channels.db import database_sync_to_async

class ModelInstanceSubscribePublisher:
    """
    This Publisher is capable of subscribing and publishing updated instances
    to graphql subscriptions.

    It expects to be called with the Info context from strawberry django,
    the instance pk and the model.

    It can be used like so:
    ```
    publisher = ModelInstanceSubscribePublisher(info=info, pk=pk, model=model)
    async for msg in publisher.await_messages():
        yield msg
    ```

    """

    def __init__(self, info: Info, pk: GlobalID, model: Model):
        self.info = info
        self.pk = pk
        self.model = model

        self.ws = info.context["ws"]
        self.channel_layer = self.ws.channel_layer

    async def verify_logged_in(self):
        user = get_current_user(self.info)

        if user.is_anonymous:
            raise Exception("No access")

    async def verify_return_type(self):
        return_type = self.info.return_type.__name__
        requested_type = self.decode_global_id_type()

        if return_type != requested_type:
            msg = f"Requested GlobalID of type {return_type} instead of {requested_type}"
            raise TypeError(msg)

    def get_queryset(self):
        return self.model.objects.all()

    def get_instance(self, pk):
        return self.get_queryset().get(pk=pk)

    def decode_global_id_type(self):
        global_id_type, _ = from_base64(self.pk)
        return global_id_type

    def decode_pk(self):
        _, pk = from_base64(self.pk)
        return pk

    @database_sync_to_async
    def set_instance(self):
        pk = self.decode_pk()
        instance = self.get_instance(pk)
        self.instance = instance

    @sync_to_async
    def refresh_instance(self):
        self.instance.refresh_from_db()
        return self.instance

    @property
    def group(self):
        return get_group(self.instance)

    @property
    def msg_type(self):
        return get_msg_type(self.instance)

    @property
    def msg(self):
        return get_msg(self.instance)

    async def subscribe(self):
        await self.channel_layer.group_add(self.group, self.ws.channel_name)

    async def send_message(self):
        await self.channel_layer.group_send(group=self.group, message=self.msg)

    async def send_initial_message(self):
        await self.send_message()

    async def await_messages(self):
        await self.verify_logged_in()
        await self.verify_return_type()
        await self.set_instance()
        await self.subscribe()
        await self.send_initial_message()

        async with self.ws.listen_to_channel(type=self.msg_type, groups=[self.group]) as messages:
            async for msg in messages:
                yield await self.refresh_instance()

From this code, I did remove a few things that are specific to our application user design.
@bellini666 will probably be able to point out which protections and securities we're missing here that would be called in normal queries. My personal code-style is also significantly different from the library, I prefer a class based approach with many small methods. The library is built more around larger methods.

At the time, the library wasn't quite ready to do these things out of the box. I would have happily completed it, but couldn't find traction to help out especially on the last part.

How to setup the subscriptions / channels have been added to the library some time ago, including documentation how to setup channels with a fresh router and a few more things: https://strawberry-graphql.github.io/strawberry-graphql-django/guide/subscriptions/

If you want to see how I fitted it into the large scheme of things:

• https://github.com/OneSila/OneSilaHeadless/tree/master/OneSila/core/schema/core/subscriptions
• https://github.com/OneSila/OneSilaHeadless/tree/master/OneSila/contacts/schema
• https://github.com/OneSila/OneSilaHeadless/blob/master/OneSila/OneSila/asgi.py

There you are. Admittedly it was quite the effort taking it this far. I hope with the 3 of us we can take it further so that subscriptions "just work".

[thclark]

@sdobbelaere this is amazing! Thank you, really nice work. Here's how I recommend proceeding:

• I have channels and ASGI set up already, I'll try to hook up the rest of your code on a simple model.
• If (big if!) I can get it to work, let's compare solutions.
  • I won't be able to share the complete codebase publicly but I'll pull out the relevant bits and paste them here.
• We identify commonalities in the API as key bits to go in the library
  • (I'm guessing on first sight that the ModelInstanceSubscribePublisher class will be common and the rest will be app-specific)
• We add the relevant parts to strawberry-django, the create a kitchen-sink example server to check it works and see what the DX is like.
• Document the above.

The first two steps there from my side will not be at all fast, as it's a non-commercial side project for which I need this, but I'll try and get to it in the coming weeks.

An Aside (practical usage)

Vaguely related, if anyone's interested in a real-world application of strawberry: I volunteer with the International Electrotechnical Committee and we're working on a digital standard for Wind Turbine specifications. My side project mentioned above is a tool that'll allow people to collaboratively create, update and share Wind Turbine specifications.

[sdobbelaere]

@thclark Glad you like it. It took good bit of R&D / Deep diving into the lib to get there. Overall it was designed to be usable with minimal effort - so you should be able to get it to work quite easily if I did my work right.

It's designed for everything to be common except for the actual field declaration:

class Subscription:
    @subscription
    async def company(self, info: Info, pk: str) -> AsyncGenerator[CompanyType, None]:
        async for i in model_subscriber(info=info, pk=pk, model=Company):
            yield I

Depending how the week starts after the weekend, I might do a PR with some docs to have a public starting point.

Unfortunately I don't have the availability off the bat to deep dive for all the potential missing use cases. Currently this could be considered "experimental".

As for your side-project, entirely alien world to me. So I'll be keeping an eye out for it. In my case, I'm working on a Saas Smart Enterprise solution which my colleague and I are building entirely publicly. It's early days still, but if interested check it out: https://github.com/OneSila you can also see how I implemented strawberry for the first time.