File: subscription_resolve.py

package info (click to toggle)
python-apischema 0.18.3-1
  • links: PTS, VCS
  • area: main
  • in suites: sid, trixie
  • size: 1,608 kB
  • sloc: python: 15,266; sh: 7; makefile: 7
file content (55 lines) | stat: -rw-r--r-- 1,054 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
import asyncio
from dataclasses import dataclass
from typing import AsyncIterable

import graphql
from graphql import print_schema

from apischema.graphql import Subscription, graphql_schema


def hello() -> str:
    return "world"


async def events() -> AsyncIterable[str]:
    yield "bonjour"
    yield "au revoir"


@dataclass
class Message:
    body: str


# Message can also be used directly as a function
schema = graphql_schema(
    query=[hello],
    subscription=[Subscription(events, alias="messageReceived", resolver=Message)],
)
schema_str = """\
type Query {
  hello: String!
}

type Subscription {
  messageReceived: Message!
}

type Message {
  body: String!
}"""
assert print_schema(schema) == schema_str


async def test():
    subscription = await graphql.subscribe(
        schema, graphql.parse("subscription {messageReceived {body}}")
    )
    assert [event.data async for event in subscription] == [
        {"messageReceived": {"body": "bonjour"}},
        {"messageReceived": {"body": "au revoir"}},
    ]


asyncio.run(test())