Generate graphql subscription methods

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## 0.9.7 (2025-06-05)
+
+### Changes
+
+- Generate GraphQL subscription methods based on WebStack GraphQL schema.
+
 ## 0.9.6 (2025-06-04)
 
 ### Changes

devbin/mujin_webstackclientpy_generategraphclient.py

@@ -84,41 +84,66 @@ def _DiscoverMethods(queryOrMutationType):
             'description': field.description,
             'returnType': _DiscoverType(field.type),
         })
-    return methods    
+    return methods
 
-def _PrintMethod(queryOrMutation, operationName, parameters, description, returnType):
-    if queryOrMutation == 'query' and operationName.startswith("List"):
+def _PrintMethod(queryOrMutationOrSubscription, operationName, parameters, description, returnType):
+    if queryOrMutationOrSubscription == 'query' and operationName.startswith("List"):
         print('    @UseLazyGraphQuery')
-    builtinParameterNames = ('fields', 'timeout')
-    print('    def %s(self, %s):' % (operationName, ', '.join([
+
+    builtinParameterNamesRequired = ()
+    builtinParameterNamesOptional = ()
+    if queryOrMutationOrSubscription in ('query', 'mutation'):
+        builtinParameterNamesOptional = ('fields', 'timeout')
+    elif queryOrMutationOrSubscription == 'subscription':
+        builtinParameterNamesRequired = ('callbackFunction',)
+        builtinParameterNamesOptional = ('fields',)
+    builtinParameterNames = builtinParameterNamesRequired + builtinParameterNamesOptional
+    operationParameters = [
         '%s=None' % parameter['parameterName'] if parameter['parameterNullable'] else parameter['parameterName']
         for parameter in parameters
         if parameter['parameterName'] not in builtinParameterNames
-    ] + ['fields=None', 'timeout=None'])))
+    ]
+    fullParameterList = list(builtinParameterNamesRequired) + operationParameters + ['%s=None' % name for name in builtinParameterNamesOptional]
+    print('    def %s(self, %s):' % (operationName, ', '.join(fullParameterList)))
+
     if description:
         print('        """%s' % description)
-        print('')
-        print('        Args:')
-        for parameter in parameters:
-            if parameter['parameterName'] in builtinParameterNames:
-                continue
-            isOptionalString = ", optional" if parameter['parameterNullable'] else ""
-            print('            %s (%s%s): %s' % (parameter['parameterName'], _FormatTypeForDocstring(parameter['parameterType']), isOptionalString, _IndentNewlines(parameter['parameterDescription'])))
-        print('            fields (list or dict, optional): Specifies a subset of fields to return.')
-        print('            timeout (float, optional): Number of seconds to wait for response.')
-        print('')
-        print('        Returns:')
-        print('            %s: %s' % (_FormatTypeForDocstring(returnType['typeName']), _IndentNewlines(returnType['description'])))
+    else:
         print('        """')
+    print('')
+    print('        Args:')
+    if queryOrMutationOrSubscription == 'subscription':
+        print('            callbackFunction (Callable[[Optional[ControllerGraphClientException], Optional[dict]], None]):')
+        print('                A function with signature that will be called when the subscription is triggered:')
+        print('                    def CallbackFunction(error: Optional[ControllerGraphClientException], response: Optional[dict]) -> None')
+        print('                - error: Contains an error message (or `None` if no error occurred).')
+        print('                - response: Contains the returned payload (or `None` if an error occurred).')
+    for parameter in parameters:
+        if parameter['parameterName'] in builtinParameterNames:
+            continue
+        isOptionalString = ", optional" if parameter['parameterNullable'] else ""
+        print('            %s (%s%s): %s' % (parameter['parameterName'], _FormatTypeForDocstring(parameter['parameterType']), isOptionalString, _IndentNewlines(parameter['parameterDescription'])))
+    print('            fields (list or dict, optional): Specifies a subset of fields to return.')
+    if queryOrMutationOrSubscription in ('query', 'mutation'):
+        print('            timeout (float, optional): Number of seconds to wait for response.')
+    print('')
+    print('        Returns:')
+    print('            %s: %s' % (_FormatTypeForDocstring(returnType['typeName']), _IndentNewlines(returnType['description'])))
+    print('        """')
+
     print('        parameterNameTypeValues = [')
     for parameter in parameters:
         if parameter['parameterName'] in builtinParameterNames:
             continue
         print('            (\'%s\', \'%s\', %s),' % (parameter['parameterName'], parameter['parameterType'], parameter['parameterName']))
     print('        ]')
-    print('        return self._CallSimpleGraphAPI(\'%s\', operationName=\'%s\', parameterNameTypeValues=parameterNameTypeValues, returnType=\'%s\', fields=fields, timeout=timeout)' % (queryOrMutation, operationName, returnType['baseTypeName']))
 
-def _PrintClient(serverVersion, queryMethods, mutationMethods):
+    if queryOrMutationOrSubscription in ('query', 'mutation'):
+        print('        return self._CallSimpleGraphAPI(\'%s\', operationName=\'%s\', parameterNameTypeValues=parameterNameTypeValues, returnType=\'%s\', fields=fields, timeout=timeout)' % (queryOrMutationOrSubscription, operationName, returnType['baseTypeName']))
+    elif queryOrMutationOrSubscription == 'subscription':
+        print('        return self._CallSubscribeGraphAPI(operationName=\'%s\', parameterNameTypeValues=parameterNameTypeValues, returnType=\'%s\', callbackFunction=callbackFunction, fields=fields)' % (operationName, returnType['baseTypeName']))
+
+def _PrintClient(serverVersion, queryMethods, mutationMethods, subscriptionMethods):
     print('# -*- coding: utf-8 -*-')
     print('#')
     print('# DO NOT EDIT, THIS FILE WAS AUTO-GENERATED')
@@ -128,6 +153,7 @@ def _PrintClient(serverVersion, queryMethods, mutationMethods):
     print('')
     print('from .webstackgraphclientutils import GraphClientBase')
     print('from .webstackgraphclientutils import UseLazyGraphQuery')
+    print('from .controllerwebclientraw import Subscription')
     print('')
     print('class GraphQueries:')
     print('')
@@ -141,7 +167,22 @@ def _PrintClient(serverVersion, queryMethods, mutationMethods):
         _PrintMethod('mutation', **mutationMethod)
         print('')
     print('')
-    print('class GraphClient(GraphClientBase, GraphQueries, GraphMutations):')
+    print('class GraphSubscriptions:')
+    print('')
+    print('    def Unsubscribe(self, subscription: Subscription):')
+    print('        """')
+    print('        Cancel an actively running subscription instance.')
+    print('')
+    print('        Args:')
+    print('            subscription (Subscription): The subscription instance to cancel.')
+    print('        """')
+    print('        self._webclient.UnsubscribeGraphAPI(subscription)')
+    print('')
+    for subscriptionMethod in subscriptionMethods:
+        _PrintMethod('subscription', **subscriptionMethod)
+        print('')
+    print('')
+    print('class GraphClient(GraphClientBase, GraphQueries, GraphMutations, GraphSubscriptions):')
     print('    pass')
     print('')
     print('#')
@@ -157,8 +198,9 @@ def _Main():
     serverVersion, schema = _FetchServerVersionAndSchema(options.url, options.username, options.password)
     queryMethods = _DiscoverMethods(schema.query_type)
     mutationMethods = _DiscoverMethods(schema.mutation_type)
+    subscriptionMethods = _DiscoverMethods(schema.subscription_type)
 
-    _PrintClient(serverVersion, queryMethods, mutationMethods)
+    _PrintClient(serverVersion, queryMethods, mutationMethods, subscriptionMethods)
 
 
 if __name__ == "__main__":

python/mujinwebstackclient/version.py

@@ -1,4 +1,4 @@
-__version__ = '0.9.6'
+__version__ = '0.9.7'
 
 # Do not forget to update CHANGELOG.md