如何创建实现订阅的自定义 GraphQLView？

问题描述

我确实找到了一个类似的帖子 here，但它不起作用。

背景

我有一个显示在前端的项目列表。每次在后端更新这些项目时。前端也应该更新。

为了使其工作，我使用 fetch API 将 GraphQL 订阅查询从前端发送到 django 后端视图，我想在那里执行它并在订阅完成时将结果发送回以更新前端。

问题

当我在后端更新模型时，订阅似乎没有触发。然而，当我调试时，我可以看到正在触发的事件（post_save），但订阅似乎没有得到解决。

前端：

fetch('http://localhost:8000/custom_graphql',{
        method: 'POST',headers: {
            "Content-Type": "application/json","X-CSRFToken": csrftoken
        },body: JSON.stringify({
            query: "\
                subscription {\
                    vendorUpdated {\
                        id\
                        name\
                    }\
                }\
            "
        })  
    })
    .then(res => res.json())
    .then(data => {       
        console.log(data);
    })  

由于 graphene-django 不支持订阅，因此我将 graphene-subscription 与它一起使用。

后端：

urls.py

urlpatterns = [    
    ...
    path('custom_graphql',views.CustomGraphQLView.as_view(),name='custom_graphql'),]

views.py

def clean_query(query):
    return query.replace("\n","")    

def get_graphql_result_from(request):
    body_data = json.loads(request.body.decode("utf-8"))
    query = clean_query(body_data['query'])                
    result = schema.execute(query,context_value=request)                      
    result = result.data  
    return result

class CustomGraphQLView(GraphQLCustomCoreBackend,View):

    def __init__(self,executor=None):
        super(GraphQLCustomCoreBackend,self).__init__(executor)
        self.execute_params['allow_subscriptions'] = True        

    def post(self,request):
        result = get_graphql_result_from(request)
        return JsonResponse(result,safe=False)  

schema.py

class Query(BaseQuery):
    pass     
   
class Mutation(BaseMutation):   
    pass 

class Subscription(BaseSubscriptions):
    pass    

schema = graphene.Schema(
    query=Query,mutation=Mutation,subscription=Subscription
)  

subscription.py

class BaseSubscriptions(graphene.ObjectType):
    
    vendor_updated = graphene.Field(VendorType)  
    
    def resolve_vendor_updated(root,info):  
        return root.filter(
            lambda event:
                (event.operation == CREATED or event.operation == UPDATED) and
                isinstance(event.instance,Vendor)
        ).map(lambda event: event.instance) 

settings.py

GRAPHENE = {
    "SCHEMA": "myapp.mypackage.schema.schema"    
}

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels.layers.InMemoryChannelLayer"
    }
}

ASGI_APPLICATION = 'myproject.routing.application'

routing.py

application = ProtocolTypeRouter({
    "websocket": URLRouter([
        path('custom_graphql',GraphqlSubscriptionConsumer)
    ]),})

我认为问题在于 GraphqlSubscriptionConsumer 可能没有被执行。

我已经在我的 repo

上发布了整个项目的代码

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）