Hi, I keep getting a 405 when using the example code, and I'm wondering if it's a CORS issue, since my frontend and backend are running on different ports. If so, how do I enable CORS with Ariadne?

Hello guys, how is it going?

I came across using ariadne (which is very well written, by the way) for a project in my company. The project consists of a microservice that is going to provide queries and mostly subscriptions for our front-end layer. Within the architecture of the project, I need to control when the client initiates a subscription and when the subscription has ended to turn on and off, respectively, some of the retrieving information mechanisms (task etc). Within that use case, I can control when the client initiates the operation in the subscription's resolver context, but there was no way of knowing when the client has ended the subscription. Considering that, I added an optional asynchronous callback to the initialization of the server called on_subscription_complete.

I've been a heavy user of the apollo-server which is a node equivalent of ariadne. By no means I want to compare, or blame whos "better", but just in the intention of presenting my idea, apollo-server has a similar feature (in it's version 3) called onOperationComplete that lets the developer know that the operation was ended by the client.

That being said, I made some simple modifications to make this feature possible in ariadne:

1. Created a class called subscription, within the contrib/subscritptions folder, to represent subscriptions within the asgi implementation context. I did it like this to have the opeartion_name in the subscriptions so I'm able to control which subscription was closed.

from typing import AsyncGenerator

class Subscription:
    def __init__(self, operation_name: str, async_generator: AsyncGenerator):
        self.operation_name = operation_name
        self.async_generator = async_generator

2. Added the on_subscription_complete callback to the initialization of the server:

OnConnect = Callable[[WebSocket, Any], None]
OnDisconnect = Callable[[WebSocket], None]
OnSubscriptionComplete = Callable[[Subscription], Awaitable]

class GraphQL:
    def __init__(
        self,
        schema: GraphQLSchema,
        *,
        context_value: Optional[ContextValue] = None,
        root_value: Optional[RootValue] = None,
        on_connect: Optional[OnConnect] = None,
        on_disconnect: Optional[OnDisconnect] = None,
        on_subscription_complete: Optional[OnSubscriptionComplete] = None,
        validation_rules: Optional[ValidationRules] = None,
        debug: bool = False,
        introspection: bool = True,
        logger: Optional[str] = None,
        error_formatter: ErrorFormatter = format_error,
        extensions: Optional[Extensions] = None,
        middleware: Optional[Middlewares] = None,
        keepalive: float = None,
    ):
        self.context_value = context_value
        self.root_value = root_value
        self.on_connect = on_connect
        self.on_disconnect = on_disconnect
        self.on_subscription_complete = on_subscription_complete
        self.validation_rules = validation_rules
        self.debug = debug
        self.introspection = introspection
        self.logger = logger
        self.error_formatter = error_formatter
        self.extensions = extensions
        self.middleware = middleware
        self.keepalive = keepalive
        self.schema = schema

3. Modified the implementation of start_websocket_subscription to support the subscription class

  if not success:
            results = cast(List[dict], results)
            await websocket.send_json(
                {"type": GQL_ERROR, "id": operation_id, "payload": results[0]}
            )
  else:
           results = cast(AsyncGenerator, results)
           subscription = Subscription(data.get("operationName"), results)
           subscriptions[operation_id] = subscription
           asyncio.ensure_future(
               self.observe_async_results(results, operation_id, websocket)
           )

4. Changed the implementation of websocket_server to creating the subscriptions Dict supporting the subscription class

 subscriptions: Dict[str, Subscription] = {}
        await websocket.accept("graphql-ws")
        try:
            while WebSocketState.DISCONNECTED not in (
                websocket.client_state,
                websocket.application_state,
            ):
                message = await websocket.receive_json()
                await self.handle_websocket_message(message, websocket, subscriptions)

5. Modified the implementation of handle_websocket_message to support the new subscriptions Dict

    async def handle_websocket_message(
        self,
        message: dict,
        websocket: WebSocket,
        subscriptions: Dict[str, Subscription],
    ):
        operation_id = cast(str, message.get("id"))
        message_type = cast(str, message.get("type"))

6. Created close_subscription that receives a subscription and calls the on_subscription_complete callback and calls the subscription.async_generator.aclose() that was called in the previous implementation

 async def close_subscription(self, subscription: Subscription):
        if self.on_subscription_complete:
            await self.on_subscription_complete(subscription)
        await subscription.async_generator.aclose()

7. Used the close_subscription in the websocket_server and handle_websocket_message to close the subscriptions accordantly

    async def websocket_server(self, websocket: WebSocket) -> None:
        subscriptions: Dict[str, Subscription] = {}
        await websocket.accept("graphql-ws")
        try:
            while WebSocketState.DISCONNECTED not in (
                websocket.client_state,
                websocket.application_state,
            ):
                message = await websocket.receive_json()
                await self.handle_websocket_message(message, websocket, subscriptions)
        except WebSocketDisconnect:
            pass
        finally:
            for subscription in subscriptions.values():
                await self.close_subscription(subscription=subscription)

     async def handle_websocket_message(
        self,
        message: dict,
        websocket: WebSocket,
        subscriptions: Dict[str, Subscription],
    ):
        operation_id = cast(str, message.get("id"))
        message_type = cast(str, message.get("type"))

        if message_type == GQL_CONNECTION_INIT:
            await self.handle_websocket_connection_init_message(
                message,
                websocket,
            )
        elif message_type == GQL_CONNECTION_TERMINATE:
            await self.handle_websocket_connection_terminate_message(
                websocket,
            )
        elif message_type == GQL_START:
            await self.start_websocket_subscription(
                message.get("payload"), operation_id, websocket, subscriptions
            )
        elif message_type == GQL_STOP:
            if operation_id in subscriptions:
                await self.close_subscription(subscription=subscriptions[operation_id])
                del subscriptions[operation_id]

And finally, here's me using it

async def on_disconnect(websocket):
    print("client disconnect")


async def on_subscription_complete(subscription):
    print(subscription)


def build_server() -> CORSMiddleware:
    schema = build_schema()
    graphql_server = CORSMiddleware(
        GraphQL(
            schema,
            on_connect=on_connect,
            on_disconnect=on_disconnect,
            on_subscription_complete=on_subscription_complete,
            context_value=context_value,
            debug=True,
        ),
        allow_origins=["*"],
        allow_methods=("GET", "POST", "OPTIONS"),
    )

    return graphql_server

Well, I hope this is a feature that helps improve ariadne as a whole and I hope I'm able to contribute in some way. If you guys have questions, please let me know. If you guys think this implementation could be done any better, please, also let me know.

Thanks in advance.

GraphQL Playground has been retired, and will no longer receive updates: https://github.com/graphql/graphql-playground/issues/1143

Down the line, I think it would make sense for Ariadne to switch to GraphiQL because of this.

Adding the option to trace default resolvers if someone want. This is actually very useful in tracing base field types for which resolvers are not explicitly set because those can be use in apollo studio to see which fields are being accessed and how much thereby helping api lifecycle management

Hi, it is very useful library, thanks. Do you plan to implement support extending a schema, documentation? In my case are default scheme and extensions. I need to extend scheme by some extensions according some rule.

Currently Ariadne implements subscription-transport-ws protocol which Apollo has deprecated and replaced with graphql-ws.

We should either update ariadne.asgi.GraphQL to graphql-ws or refactor it to make WebSocket support handled via strategy pattern, eg:

app = GraphQL(schema, websocket=GraphQLWS(options))

GraphQL enums are represented in Python via the GraphQLEnumType. If those types are created from the SDL, they define no Python representation for GraphQL enum items. Developers can use Ariadne-provided EnumType to specify how enum items should be represented in Python themselves, but we also include "fallback" step that fills enum values with their str representations.

However neither of those approaches takes care of enum items defined as default values for field arguments. Eg:

type Query {
    users(role: UserRole = ADMIN): [User]
}

input FooBarInput {
    foo: FooEnum! = FOO
}

We will need to add additional step to make_executable_schema that gathers all enums from schema together with their Python representations, and then introspects schema's fields values for places where field or input has default value type of enum. If this is the case, we should set correct default_value on GraphQLArgument in the schema.

Hey there! I ported the custom server example from django to flask.

It's quite short and it would be nice to add it to the docs.

from ariadne import gql, ResolverMap, make_executable_schema
from ariadne.constants import PLAYGROUND_HTML
from graphql import format_error, graphql_sync
from flask import request, jsonify
from flask.views import MethodView

type_defs = gql("""
    type Query {
        hello: String!
    }
""")


query = ResolverMap("Query")


@query.field("hello")
def resolve_hello(_, info):
    request = info.context
    print(request.headers)
    user_agent = request.headers.get("User-Agent", "Guest")
    return "Hello, %s!" % user_agent


schema = make_executable_schema(type_defs, query)


class GraphQlView(MethodView):

    def get(self):
        return PLAYGROUND_HTML, 200

    def post(self):
        data = request.get_json()
        if data is None or not isinstance(data, dict):
            return 'Bad Request', 400

        variables = data.get('variables')
        if variables and not isinstance(variables, dict):
            return 'Bad Request', 400

        # Note: Passing the request to the context is option. In Flask, the current
        #   request is allways accessible as flask.request.
        result = graphql_sync(
            schema,
            data.get('query'),
            context_value=request,
            variable_values=variables,
            operation_name=data.get('operationName')
        )

        response = {"data": result.data}
        if result.errors:
            response["errors"] = [format_error(e) for e in result.errors]

        return jsonify(response)

To avoid the error on e.g. isinstance(QueryType(), SchemaBindable)

TypeError: Instance and class checks can only be used with @runtime_checkable protocols

type FooBar {
  # some more flat fields
}

type Bar {
  n: Int!
  sub: FooBar!
}

type Foo {
  foo: [Bar!]!
}

We have a structure like this. The foo property on Foo can contain thousands of items. Sadly we need all of them at once, so we can't paginate over it. There's only one resolver to resolve foo itself which is quite fast (getting the nested object from a Mongo DB). Our current workaround is to implement a custom JSON scalar which just returns the data, by-passing all child resolvers. But this way we're loosing everything that GraphQL gives us. Is there anything that can be used to speed this up without using a JSON scalar or is this the best solution already? I know that GraphQL comes with some overhead but this can go up to 20 seconds vs 1 second with JSON scalar.

We're using Ariadne in sync-mode.

Please see initial work for making the playground editable / ability to disable

1. I didn't think about supporting python3.6 (could add variable dataclass backport for 3.6)
2. I don't really know how to use pytest / snapshots and would love some pointers, but I'll try and get tests in
3. Is there a formatting / linting guide?

I initially wanted to try and implement the handling of different content-type headers, but I'll move that into a seperate PR.

Unfortunately, multipart's File class seems to have a serious regression compared to cgi's FieldStorage class: Where FieldStorage.type contained the declared MIME type of the uploaded file (or None if not given), File does not seem to have this information. This makes is basically impossible to download uploaded files while keeping the file type intact.

Adding Middlewares to ASGI GraphQL as documented in https://ariadnegraphql.org/docs/middleware#custom-middleware-example doesn't work.

This is the example from docs at the time of writing this issue:

from ariadne.asgi import GraphQL
from ariadne.asgi.handlers import GraphQLHTTPHandler
from graphql import MiddlewareManager


app = GrapqhQL(
    schema,
    http_handler=GraphQLHTTPHandler(
        middleware=MiddlewareManager(lowercase_middleware),
    ),
)

Firts, mypy will throw following error:

error: Argument "middleware" to "GraphQLHTTPHandler" has incompatible type "MiddlewareManager"; expected
"Union[Callable[[Any, Union[Any, Callable[[Any], Any], None]], Optional[List[Union[Tuple[Any, ...], List[Any], MiddlewareManager, None]]]], Optional[List[Union[Tuple[Any, ...], List[Any], MiddlewareManager, None]]], None]"

Second, running this code will throw following error:

self = <ariadne.asgi.handlers.http.GraphQLHTTPHandler object at 0x127605390>, request = <starlette.requests.Request object at 0x1279b5600>
context = {'request': <starlette.requests.Request object at 0x1279b5600>}

    async def get_middleware_for_request(
        self, request: Any, context: Optional[ContextValue]
    ) -> Optional[MiddlewareManager]:
        middleware = self.middleware
        if callable(middleware):
            middleware = middleware(request, context)
            if isawaitable(middleware):
                middleware = await middleware  # type: ignore
        if middleware:
            middleware = cast(list, middleware)
>           return MiddlewareManager(*middleware)
E           TypeError: graphql.execution.middleware.MiddlewareManager() argument after * must be an iterable, not MiddlewareManager

By looking at the expected type I've tried:

from ariadne.asgi import GraphQL
from ariadne.asgi.handlers import GraphQLHTTPHandler
from graphql import MiddlewareManager


app = GrapqhQL(
    schema,
    http_handler=GraphQLHTTPHandler(
        middleware=[
            MiddlewareManager(lowercase_middleware),
        ],
    ),
)

It does satisfy mypy, but when the code is run the middlewares aren't ever getting called.

By looking at the test suite:

https://github.com/mirumee/ariadne/blob/9a6cd62c9172b5e43cc62cccdec9f28294d7238d/tests/asgi/test_configuration.py#L589..L621

For example in this part:

def test_middlewares_are_passed_to_query_executor(schema):
    http_handler = GraphQLHTTPHandler(middleware=[middleware])
    app = GraphQL(schema, http_handler=http_handler)
    client = TestClient(app)
    response = client.post("/", json={"query": '{ hello(name: "BOB") }'})
    assert response.json() == {"data": {"hello": "**Hello, BOB!**"}}

I've noted, that you use a different syntax. This one is working, but it contradicts the doc and doesn't work with static type checking.

From mypy you will get following error:

error: List item 0 has incompatible type "Callable[[Callable[..., Any], Any, GraphQLResolveInfo, KwArg(Any)], Any]"; expected
"Union[Tuple[Any, ...], List[Any], MiddlewareManager, None]"  [list-item]

Workaround that I've found is to do:

from ariadne.asgi import GraphQL
from ariadne.asgi.handlers import GraphQLHTTPHandler


app = GrapqhQL(
    schema,
    http_handler=GraphQLHTTPHandler(
        middleware=[
            lowercase_middleware,  # type: ignore[list-item]
        ],
    ),
)

I'd suggest adding type annotations to the test suite and run mypy on it.

Similar to #547 but this bug is specific to type "interface". It seems that if you try to define a schema in which an interface type contains an enum argument with a default value, an error will be thrown when trying to execute an introspection query. This is with:

ariadne 0.16.1 graphql-core 3.2.3

Example:

import enum
import ariadne

type_defs = """
    type Query {
        users: [User!]!
    }

    enum Role {
        ADMIN
        USER
    }

    interface User {
        name(role: Role! = USER): String!
    }
"""

QueryGraphQLType = ariadne.QueryType()

class Role(enum.Enum):
    ADMIN = "ADMIN"
    USER = "USER"

RoleGraphQLType = ariadne.EnumType("Role", Role)

UserGraphQLType = ariadne.InterfaceType("User")

schema = ariadne.make_executable_schema(
    type_defs,
    QueryGraphQLType,
    RoleGraphQLType,
    UserGraphQLType,
)

query = "{__schema{types{name,fields{name,args{name,defaultValue}}}}}"

result = ariadne.graphql_sync(schema, {"query": query}, debug=True)
import pprint
pprint.pprint(result)

Error:

Enum 'Role' cannot represent value: 'USER'

GraphQL request:1:44
1 | {__schema{types{name,fields{name,args{name,defaultValue}}}}}
  |                                            ^
Traceback (most recent call last):
  File "/venv/lib/python3.8/site-packages/graphql/execution/execute.py", line 521, in execute_field
    result = resolve_fn(source, info, **args)
  File "/venv/lib/python3.8/site-packages/graphql/type/introspection.py", line 485, in default_value
    value_ast = ast_from_value(item[1].default_value, item[1].type)
  File "/venv/lib/python3.8/site-packages/graphql/utilities/ast_from_value.py", line 63, in ast_from_value
    ast_value = ast_from_value(value, type_.of_type)
  File "/venv/lib/python3.8/site-packages/graphql/utilities/ast_from_value.py", line 108, in ast_from_value
    serialized = type_.serialize(value)  # type: ignore
  File "/venv/lib/python3.8/site-packages/graphql/type/definition.py", line 1233, in serialize
    raise GraphQLError(
graphql.error.graphql_error.GraphQLError: Enum 'Role' cannot represent value: 'USER'

We're trying to generate an OpenAPI schema using starlette.schemas.SchemaGenerator, however it does not include any definitions for ariadne.

As per starlette's docs:

Schema generation works by inspecting the routes on the application through app.routes, and using the docstrings or other attributes on the endpoints in order to determine a complete API schema.

Is it possible to add a docstring into ariadne's graphql endpoint, or allow it to be configured manually so that it can be picked up by ariadne's schema generator?

EDIT: One possible workaround is to wrap ariadne's route with another route, but it would be great if this was available out of the box. EDIT 2: Seems like SchemaGenerator.get_endpoints doesn't find ariadne's endpoints. We're using it as follows:

class GraphQLAdapter(Starlette):
    SCHEMA_PATH = ...

    def __init__(self):
        ...
        schema = make_executable_schema(...)
        graphql = GraphQL(schema)

        super().__init__(
            routes=[
                Mount('', graphql),
            ],
        )