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())
|
