implemented graphql server python over rsocket

Author: jell-o-fishi

examples/graphql/client_graphql.py

@@ -1,6 +1,7 @@
 import asyncio
 import logging
 import sys
+from pathlib import Path
 
 from gql import gql, Client
 
@@ -16,16 +17,11 @@ async def main(server_port: int):
 
     async with RSocketClient(single_transport_provider(TransportTCP(*connection)),
                              metadata_encoding=WellKnownMimeTypes.MESSAGE_RSOCKET_COMPOSITE_METADATA) as client:
+        with (Path(__file__).parent / 'rsocket.graphqls').open() as fd:
+            schema = fd.read()
+
         graphql = Client(
-            schema="""
-            type Query {
-    greeting: Greeting
-}
-
-type Greeting {
-    message: String
-}
-            """,
+            schema=schema,
             transport=RSocketTransport(client),
         )
 

examples/graphql/rsocket.graphqls

@@ -0,0 +1,11 @@
+type Query {
+    greeting: Greeting
+}
+
+type Subscription {
+    greetings: Greeting
+}
+
+type Greeting {
+    message: String
+}

examples/graphql/server_graphql.py

@@ -1,76 +1,101 @@
+import asyncio
 import json
 import logging
 import sys
+from pathlib import Path
 from typing import AsyncGenerator, Tuple
 
-from graphql import parse, subscribe
-from graphql_server import get_graphql_params
-from quart import Quart
+from graphql import build_schema, subscribe, parse
 
-from examples.graphql.schema import AsyncSchema
 from rsocket.frame_helpers import str_to_bytes
-from rsocket.graphql.helpers import execute_query_in_payload
+from rsocket.graphql.helpers import execute_query_in_payload, get_graphql_params
 from rsocket.helpers import create_future
 from rsocket.payload import Payload
 from rsocket.routing.request_router import RequestRouter
 from rsocket.routing.routing_request_handler import RoutingRequestHandler
+from rsocket.rsocket_server import RSocketServer
 from rsocket.streams.stream_from_async_generator import StreamFromAsyncGenerator
-from rsocket.transports.quart_websocket import websocket_handler
+from rsocket.transports.tcp import TransportTCP
 
-app = Quart(__name__)
 
-router = RequestRouter()
+def greeting(*args):
+    return {
+        'message': "Hello world"
+    }
 
 
-@router.response('graphql')
-async def graphql_query(payload: Payload):
-    execution_result = await execute_query_in_payload(payload, AsyncSchema)
+def greetings(*args):
+    async def results():
+        for i in range(10):
+            yield {'greetings': {'message': f"Hello world {i}"}}
+            await asyncio.sleep(1)
 
-    response_data = str_to_bytes(json.dumps({
-        'data': execution_result.data
-    }))
+    return results()
 
-    return create_future(Payload(response_data))
 
+class GraphqlRequestHandler:
 
-@router.stream('graphql')
-async def graphql_subscription(payload: Payload):
-    async def generator() -> AsyncGenerator[Tuple[Payload, bool], None]:
-        data = json.loads(payload.data.decode('utf-8'))
-        params = get_graphql_params(data, {})
-        schema = AsyncSchema
-        document = parse(params.query)
+    def __init__(self):
+        with (Path(__file__).parent / 'rsocket.graphqls').open() as fd:
+            schema = build_schema(fd.read())
+
+        schema.query_type.fields['greeting'].resolve = greeting
+        schema.subscription_type.fields['greetings'].subscribe = greetings
+
+        router = RequestRouter()
+
+        @router.response('graphql')
+        async def graphql_query(payload: Payload):
+            execution_result = await execute_query_in_payload(payload, schema)
 
-        async for execution_result in await subscribe(
-                schema,
-                document,
-                variable_values=params.variables,
-                operation_name=params.operation_name
-        ):
-            item = execution_result.data
             response_data = str_to_bytes(json.dumps({
-                'data': item[0]
+                'data': execution_result.data
             }))
-            yield Payload(response_data), item[1]
 
-    return StreamFromAsyncGenerator(generator)
+            return create_future(Payload(response_data))
+
+        @router.stream('graphql')
+        async def graphql_subscription(payload: Payload):
+            async def generator() -> AsyncGenerator[Tuple[Payload, bool], None]:
+                data = json.loads(payload.data.decode('utf-8'))
+                params = get_graphql_params(data, {})
+                document = parse(params.query)
+
+                async for execution_result in await subscribe(
+                        schema,
+                        document,
+                        operation_name=params.operation_name
+                ):
+                    item = execution_result.data
+                    response_data = str_to_bytes(json.dumps({
+                        'data': item
+                    }))
+                    yield Payload(response_data), False
 
+                yield Payload(), True
 
-@router.response('ping')
-async def ping():
-    return create_future(Payload(b'pong'))
+            return StreamFromAsyncGenerator(generator)
+
+        self.router = router
 
 
 def handler_factory():
-    return RoutingRequestHandler(router)
+    return RoutingRequestHandler(GraphqlRequestHandler().router)
+
+
+async def run_server(server_port):
+    logging.info('Starting server at localhost:%s', server_port)
+
+    def session(*connection):
+        RSocketServer(TransportTCP(*connection), handler_factory=handler_factory)
 
+    server = await asyncio.start_server(session, 'localhost', server_port)
 
-@app.websocket("/")
-async def ws():
-    await websocket_handler(handler_factory=handler_factory)
+    async with server:
+        await server.serve_forever()
 
 
-if __name__ == "__main__":
-    port = sys.argv[1] if len(sys.argv) > 1 else 7000
+if __name__ == '__main__':
+    port = sys.argv[1] if len(sys.argv) > 1 else 9191
     logging.basicConfig(level=logging.DEBUG)
-    app.run(port=port)
+    asyncio.run(run_server(port))

requirements.txt

@@ -21,3 +21,4 @@ cbitstruct==1.0.9
 cloudevents==1.10.0
 pydantic==1.10.13
 Werkzeug==3.0.0
+graphql-core==3.2.3

rsocket/graphql/helpers.py

@@ -1,22 +1,42 @@
 import json
+from collections import namedtuple
+from typing import Dict, Optional, Union
 
 from graphql import execute, parse, ExecutionResult, GraphQLSchema
-from graphql_server import get_graphql_params
 
 from rsocket.payload import Payload
 
+GraphQLParams = namedtuple("GraphQLParams", "query variables operation_name")
+
 
 async def execute_query_in_payload(payload: Payload,
                                    schema: GraphQLSchema) -> ExecutionResult:
     data = json.loads(payload.data.decode('utf-8'))
     params = get_graphql_params(data, {})
     document = parse(params.query)
 
-    execution_result = await execute(
+    execution_result = execute(
         schema,
         document,
         variable_values=params.variables,
         operation_name=params.operation_name
     )
 
     return execution_result
+
+
+def get_graphql_params(data: Dict, query_data: Dict) -> GraphQLParams:
+    query = data.get("query") or query_data.get("query")
+    variables = data.get("variables") or query_data.get("variables")
+    operation_name = data.get("operationName") or query_data.get("operationName")
+
+    return GraphQLParams(query, load_json_variables(variables), operation_name)
+
+
+def load_json_variables(variables: Optional[Union[str, Dict]]) -> Optional[Dict]:
+    if variables and isinstance(variables, str):
+        try:
+            return json.loads(variables)
+        except Exception:
+            raise Exception("Variables are invalid JSON.")
+        return variables  # type: ignore

setup.cfg

@@ -57,6 +57,8 @@ optimized = cbitstruct>=1.0.9
 cloudevents =
     cloudevents>=1.9.0
     pydantic>=1.10.0
+graphql =
+    graphql-core>=3.2.0
 
 [options.entry_points]
 cli.console_scripts =