Skip to content

Commit 33dc6e7

Browse files
committed
Merge branch 'master' into custom-scalar-types
2 parents 98475a8 + 8fc378d commit 33dc6e7

11 files changed

+403
-129
lines changed

README.md

Lines changed: 169 additions & 87 deletions
Large diffs are not rendered by default.

gql/__init__.py

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,13 @@
11
from .client import Client
22
from .gql import gql
3+
from .transport.aiohttp import AIOHTTPTransport
4+
from .transport.requests import RequestsHTTPTransport
5+
from .transport.websockets import WebsocketsTransport
36

4-
__all__ = ["gql", "Client"]
7+
__all__ = [
8+
"gql",
9+
"AIOHTTPTransport",
10+
"Client",
11+
"RequestsHTTPTransport",
12+
"WebsocketsTransport",
13+
]

gql/client.py

Lines changed: 49 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import asyncio
2+
import warnings
23
from typing import Any, AsyncGenerator, Dict, Generator, Optional, Union
34

45
from graphql import (
@@ -22,7 +23,7 @@
2223
class Client:
2324
def __init__(
2425
self,
25-
schema: Optional[GraphQLSchema] = None,
26+
schema: Optional[Union[str, GraphQLSchema]] = None,
2627
introspection=None,
2728
type_def: Optional[str] = None,
2829
transport: Optional[Union[Transport, AsyncTransport]] = None,
@@ -32,23 +33,34 @@ def __init__(
3233
):
3334
assert not (
3435
type_def and introspection
35-
), "Cannot provide introspection type definition at the same time."
36-
if transport and fetch_schema_from_transport:
36+
), "Cannot provide introspection and type definition at the same time."
37+
38+
if type_def:
3739
assert (
3840
not schema
39-
), "Cannot fetch the schema from transport if is already provided."
41+
), "Cannot provide type definition and schema at the same time."
42+
warnings.warn(
43+
"type_def is deprecated; use schema instead",
44+
category=DeprecationWarning,
45+
)
46+
schema = type_def
47+
4048
if introspection:
4149
assert (
4250
not schema
4351
), "Cannot provide introspection and schema at the same time."
4452
schema = build_client_schema(introspection)
45-
elif type_def:
53+
54+
if isinstance(schema, str):
55+
type_def_ast = parse(schema)
56+
schema = build_ast_schema(type_def_ast)
57+
58+
if transport and fetch_schema_from_transport:
4659
assert (
4760
not schema
48-
), "Cannot provide type definition and schema at the same time."
49-
type_def_ast = parse(type_def)
50-
schema = build_ast_schema(type_def_ast)
51-
elif schema and not transport:
61+
), "Cannot fetch the schema from transport if is already provided."
62+
63+
if schema and not transport:
5264
transport = LocalSchemaTransport(schema)
5365

5466
# GraphQL schema
@@ -114,8 +126,8 @@ def execute(self, document: DocumentNode, *args, **kwargs) -> Dict:
114126
loop = asyncio.get_event_loop()
115127

116128
assert not loop.is_running(), (
117-
"Cannot run client.execute if an asyncio loop is running."
118-
" Use execute_async instead."
129+
"Cannot run client.execute(query) if an asyncio loop is running."
130+
" Use 'await client.execute_async(query)' instead."
119131
)
120132

121133
data: Dict[Any, Any] = loop.run_until_complete(
@@ -132,11 +144,11 @@ async def subscribe_async(
132144
) -> AsyncGenerator[Dict, None]:
133145
async with self as session:
134146

135-
self._generator: AsyncGenerator[Dict, None] = session.subscribe(
147+
generator: AsyncGenerator[Dict, None] = session.subscribe(
136148
document, *args, **kwargs
137149
)
138150

139-
async for result in self._generator:
151+
async for result in generator:
140152
yield result
141153

142154
def subscribe(
@@ -152,18 +164,35 @@ def subscribe(
152164
loop = asyncio.get_event_loop()
153165

154166
assert not loop.is_running(), (
155-
"Cannot run client.subscribe if an asyncio loop is running."
156-
" Use subscribe_async instead."
167+
"Cannot run client.subscribe(query) if an asyncio loop is running."
168+
" Use 'await client.subscribe_async(query)' instead."
157169
)
158170

159171
try:
160172
while True:
161-
result = loop.run_until_complete(async_generator.__anext__())
173+
# Note: we need to create a task here in order to be able to close
174+
# the async generator properly on python 3.8
175+
# See https://bugs.python.org/issue38559
176+
generator_task = asyncio.ensure_future(async_generator.__anext__())
177+
result = loop.run_until_complete(generator_task)
162178
yield result
163179

164180
except StopAsyncIteration:
165181
pass
166182

183+
except (KeyboardInterrupt, Exception):
184+
185+
# Graceful shutdown by cancelling the task and waiting clean shutdown
186+
generator_task.cancel()
187+
188+
try:
189+
loop.run_until_complete(generator_task)
190+
except (StopAsyncIteration, asyncio.CancelledError):
191+
pass
192+
193+
# Then reraise the exception
194+
raise
195+
167196
async def __aenter__(self):
168197

169198
assert isinstance(
@@ -183,9 +212,10 @@ async def __aexit__(self, exc_type, exc, tb):
183212

184213
def __enter__(self):
185214

186-
assert not isinstance(
187-
self.transport, AsyncTransport
188-
), "Only a sync transport can be use. Use 'async with Client(...)' instead"
215+
assert not isinstance(self.transport, AsyncTransport), (
216+
"Only a sync transport can be used."
217+
" Use 'async with Client(...) as session:' instead"
218+
)
189219

190220
self.transport.connect()
191221

gql/transport/websockets.py

Lines changed: 25 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -175,6 +175,23 @@ async def _receive(self) -> str:
175175

176176
return answer
177177

178+
async def _wait_ack(self) -> None:
179+
"""Wait for the connection_ack message. Keep alive messages are ignored
180+
"""
181+
182+
while True:
183+
init_answer = await self._receive()
184+
185+
answer_type, answer_id, execution_result = self._parse_answer(init_answer)
186+
187+
if answer_type == "connection_ack":
188+
return
189+
190+
if answer_type != "ka":
191+
raise TransportProtocolError(
192+
"Websocket server did not return a connection ack"
193+
)
194+
178195
async def _send_init_message_and_wait_ack(self) -> None:
179196
"""Send init message to the provided websocket and wait for the connection ACK.
180197
@@ -188,14 +205,7 @@ async def _send_init_message_and_wait_ack(self) -> None:
188205
await self._send(init_message)
189206

190207
# Wait for the connection_ack message or raise a TimeoutError
191-
init_answer = await asyncio.wait_for(self._receive(), self.ack_timeout)
192-
193-
answer_type, answer_id, execution_result = self._parse_answer(init_answer)
194-
195-
if answer_type != "connection_ack":
196-
raise TransportProtocolError(
197-
"Websocket server did not return a connection ack"
198-
)
208+
await asyncio.wait_for(self._wait_ack(), self.ack_timeout)
199209

200210
async def _send_stop_message(self, query_id: int) -> None:
201211
"""Send stop message to the provided websocket connection and query_id.
@@ -233,16 +243,14 @@ async def _send_query(
233243
query_id = self.next_query_id
234244
self.next_query_id += 1
235245

246+
payload: Dict[str, Any] = {"query": print_ast(document)}
247+
if variable_values:
248+
payload["variables"] = variable_values
249+
if operation_name:
250+
payload["operationName"] = operation_name
251+
236252
query_str = json.dumps(
237-
{
238-
"id": str(query_id),
239-
"type": "start",
240-
"payload": {
241-
"variables": variable_values or {},
242-
"operationName": operation_name or "",
243-
"query": print_ast(document),
244-
},
245-
}
253+
{"id": str(query_id), "type": "start", "payload": payload}
246254
)
247255

248256
await self._send(query_str)

setup.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@
3232

3333
setup(
3434
name="gql",
35-
version="3.0.0a0",
35+
version="3.0.0a1",
3636
description="GraphQL client for Python",
3737
long_description=open("README.md").read(),
3838
long_description_content_type="text/markdown",

tests/starwars/test_validation.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@ def local_schema():
1313
@pytest.fixture
1414
def typedef_schema():
1515
return Client(
16-
type_def="""
16+
schema="""
1717
schema {
1818
query: Query
1919
}

tests/test_async_client_validation.py

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,7 @@ async def server_starwars(ws, path):
8282
{"schema": StarWarsSchema},
8383
{"introspection": StarWarsIntrospection},
8484
{"type_def": StarWarsTypeDef},
85+
{"schema": StarWarsTypeDef},
8586
],
8687
)
8788
async def test_async_client_validation(
@@ -124,6 +125,7 @@ async def test_async_client_validation(
124125
{"schema": StarWarsSchema},
125126
{"introspection": StarWarsIntrospection},
126127
{"type_def": StarWarsTypeDef},
128+
{"schema": StarWarsTypeDef},
127129
],
128130
)
129131
async def test_async_client_validation_invalid_query(

tests/test_transport.py

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -119,3 +119,11 @@ def test_named_query(client):
119119
with use_cassette("queries"):
120120
result = client.execute(query, operation_name="Planet2")
121121
assert result == expected
122+
123+
124+
def test_import_transports_directly_from_gql():
125+
from gql import AIOHTTPTransport, RequestsHTTPTransport, WebsocketsTransport
126+
127+
assert AIOHTTPTransport
128+
assert RequestsHTTPTransport
129+
assert WebsocketsTransport

tests/test_websocket_exceptions.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -241,7 +241,7 @@ async def test_websocket_transport_protocol_errors(event_loop, client_and_server
241241

242242
async def server_without_ack(ws, path):
243243
# Sending something else than an ack
244-
await WebSocketServer.send_keepalive(ws)
244+
await WebSocketServer.send_complete(ws, 1)
245245
await ws.wait_closed()
246246

247247

tests/test_websocket_query.py

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -472,3 +472,43 @@ async def test_websocket_add_extra_parameters_to_connect(event_loop, server):
472472

473473
async with Client(transport=sample_transport) as session:
474474
await session.execute(query)
475+
476+
477+
async def server_sending_keep_alive_before_connection_ack(ws, path):
478+
await WebSocketServer.send_keepalive(ws)
479+
await WebSocketServer.send_keepalive(ws)
480+
await WebSocketServer.send_keepalive(ws)
481+
await WebSocketServer.send_keepalive(ws)
482+
await WebSocketServer.send_connection_ack(ws)
483+
result = await ws.recv()
484+
print(f"Server received: {result}")
485+
await ws.send(query1_server_answer.format(query_id=1))
486+
await WebSocketServer.send_complete(ws, 1)
487+
await ws.wait_closed()
488+
489+
490+
@pytest.mark.asyncio
491+
@pytest.mark.parametrize(
492+
"server", [server_sending_keep_alive_before_connection_ack], indirect=True
493+
)
494+
@pytest.mark.parametrize("query_str", [query1_str])
495+
async def test_websocket_non_regression_bug_108(
496+
event_loop, client_and_server, query_str
497+
):
498+
499+
# This test will check that we now ignore keepalive message
500+
# arriving before the connection_ack
501+
# See bug #108
502+
503+
session, server = client_and_server
504+
505+
query = gql(query_str)
506+
507+
result = await session.execute(query)
508+
509+
print("Client received:", result)
510+
511+
continents = result["continents"]
512+
africa = continents[0]
513+
514+
assert africa["code"] == "AF"

0 commit comments

Comments
 (0)