0
# GQL
1
2
A comprehensive GraphQL client library for Python that enables developers to execute GraphQL queries, mutations, and subscriptions using multiple transport protocols including HTTP, WebSockets (with support for Apollo, GraphQL-WS, Phoenix channels, and AWS AppSync), and file-based operations. The library offers both synchronous and asynchronous usage patterns with support for concurrent requests, local query validation using GraphQL schemas, file uploads, custom scalars and enums, request batching, and a command-line interface for executing queries and downloading schemas.
3
4
## Package Information
5
6
- **Package Name**: gql
7
- **Language**: Python
8
- **Installation**: `pip install gql`
9
10
## Core Imports
11
12
```python
13
from gql import gql, Client, GraphQLRequest, FileVar
14
```
15
16
For transport classes:
17
```python
18
from gql.transport.requests import RequestsHTTPTransport
19
from gql.transport.aiohttp import AIOHTTPTransport
20
from gql.transport.httpx import HTTPXTransport, HTTPXAsyncTransport
21
from gql.transport.websockets import WebsocketsTransport
22
```
23
24
For DSL (Domain Specific Language) functionality:
25
```python
26
from gql.dsl import DSLSchema, DSLQuery, dsl_gql
27
```
28
29
## Basic Usage
30
31
### Simple HTTP Client
32
33
```python
34
from gql import gql, Client
35
from gql.transport.requests import RequestsHTTPTransport
36
37
# Create transport and client
38
transport = RequestsHTTPTransport(url="https://api.example.com/graphql")
39
client = Client(transport=transport, fetch_schema_from_transport=True)
40
41
# Execute a query
42
query = gql('''
43
query GetUser($id: ID!) {
44
user(id: $id) {
45
name
46
47
}
48
}
49
''')
50
51
result = client.execute(query, variable_values={"id": "123"})
52
print(result["user"]["name"])
53
```
54
55
### Asynchronous Usage
56
57
```python
58
import asyncio
59
from gql import gql, Client
60
from gql.transport.aiohttp import AIOHTTPTransport
61
62
async def main():
63
transport = AIOHTTPTransport(url="https://api.example.com/graphql")
64
65
async with Client(transport=transport) as session:
66
query = gql('{ hello }')
67
result = await session.execute(query)
68
print(result)
69
70
asyncio.run(main())
71
```
72
73
### WebSocket Subscriptions
74
75
```python
76
import asyncio
77
from gql import gql, Client
78
from gql.transport.websockets import WebsocketsTransport
79
80
async def main():
81
transport = WebsocketsTransport(url="wss://api.example.com/graphql")
82
83
async with Client(transport=transport) as session:
84
subscription = gql('''
85
subscription {
86
messageAdded {
87
id
88
content
89
user
90
}
91
}
92
''')
93
94
async for result in session.subscribe(subscription):
95
print(f"New message: {result}")
96
97
asyncio.run(main())
98
```
99
100
## Architecture
101
102
The gql library follows a modular architecture with clear separation of concerns:
103
104
- **Client**: Central orchestrator managing transport connections, schema handling, and execution coordination
105
- **Transports**: Protocol-specific implementations for HTTP, WebSocket, and local schema operations
106
- **GraphQLRequest**: Immutable container for GraphQL operations with variables and metadata
107
- **DSL**: Domain-specific language for programmatic query construction without string templates
108
- **Sessions**: Context managers for batched operations and connection lifecycle management
109
- **Utilities**: Schema introspection, variable serialization, result parsing, and custom scalar/enum handling
110
111
This design enables flexible transport switching, comprehensive async/sync support, and seamless integration with the broader GraphQL ecosystem including graphene, graphql-core, and graphql-js.
112
113
## Capabilities
114
115
### Client and Sessions
116
117
Core client functionality for executing GraphQL operations, managing connections, and handling schemas. Includes both synchronous and asynchronous clients with session management for connection pooling and batched operations.
118
119
```python { .api }
120
class Client:
121
def __init__(
122
self,
123
*,
124
schema: Optional[Union[str, GraphQLSchema]] = None,
125
introspection: Optional[IntrospectionQuery] = None,
126
transport: Optional[Union[Transport, AsyncTransport]] = None,
127
fetch_schema_from_transport: bool = False,
128
introspection_args: Optional[Dict] = None,
129
execute_timeout: Optional[Union[int, float]] = 10,
130
serialize_variables: bool = False,
131
parse_results: bool = False,
132
batch_interval: float = 0,
133
batch_max: int = 10
134
): ...
135
136
def execute(self, request: GraphQLRequest, **kwargs) -> ExecutionResult: ...
137
async def execute_async(self, request: GraphQLRequest, **kwargs) -> ExecutionResult: ...
138
def session(self, **kwargs) -> SyncClientSession: ...
139
async def connect_async(self, **kwargs) -> AsyncClientSession: ...
140
141
def gql(request_string: str) -> GraphQLRequest: ...
142
143
class GraphQLRequest:
144
def __init__(
145
self,
146
request: Union[DocumentNode, "GraphQLRequest", str],
147
*,
148
variable_values: Optional[Dict[str, Any]] = None,
149
operation_name: Optional[str] = None
150
): ...
151
```
152
153
[Client and Sessions](./client-sessions.md)
154
155
### Transport Protocols
156
157
Comprehensive transport implementations for HTTP, WebSocket, and local schema operations. Includes synchronous and asynchronous variants with protocol-specific optimizations and authentication support.
158
159
```python { .api }
160
# HTTP Transports
161
class RequestsHTTPTransport(Transport): ...
162
class HTTPXTransport(Transport): ...
163
class HTTPXAsyncTransport(AsyncTransport): ...
164
class AIOHTTPTransport(AsyncTransport): ...
165
166
# WebSocket Transports
167
class WebsocketsTransport(AsyncTransport): ...
168
class AIOHTTPWebsocketsTransport(AsyncTransport): ...
169
class PhoenixChannelWebsocketsTransport(AsyncTransport): ...
170
class AppSyncWebsocketsTransport(AsyncTransport): ...
171
172
# Local Schema Transport
173
class LocalSchemaTransport(AsyncTransport): ...
174
175
# File Upload Support
176
class FileVar:
177
def __init__(
178
self,
179
f: Any,
180
*,
181
filename: Optional[str] = None,
182
content_type: Optional[str] = None,
183
streaming: bool = False,
184
streaming_block_size: int = 64 * 1024
185
): ...
186
```
187
188
[Transport Protocols](./transports.md)
189
190
### DSL Query Builder
191
192
Domain-specific language for programmatic GraphQL query construction without string templates. Provides type-safe query building with schema introspection, variable handling, and fragment support.
193
194
```python { .api }
195
class DSLSchema:
196
def __init__(self, schema: GraphQLSchema): ...
197
198
class DSLQuery:
199
def __init__(self, *fields, **fields_with_alias): ...
200
201
class DSLMutation:
202
def __init__(self, *fields, **fields_with_alias): ...
203
204
class DSLSubscription:
205
def __init__(self, *fields, **fields_with_alias): ...
206
207
class DSLFragment:
208
def __init__(self, name: str): ...
209
def on(self, type_condition: DSLType) -> DSLFragment: ...
210
def select(self, *fields, **fields_with_alias) -> DSLFragment: ...
211
212
def dsl_gql(*operations, **operations_with_name) -> GraphQLRequest: ...
213
```
214
215
[DSL Query Builder](./dsl.md)
216
217
### Schema Utilities
218
219
Utilities for schema introspection, custom scalar/enum handling, variable serialization, and result parsing. Includes schema building from introspection and result processing with custom type deserialization.
220
221
```python { .api }
222
def build_client_schema(introspection: IntrospectionQuery) -> GraphQLSchema: ...
223
def get_introspection_query_ast(**options) -> DocumentNode: ...
224
def parse_result(schema, document, result, operation_name=None) -> Optional[Dict[str, Any]]: ...
225
def serialize_variable_values(schema, document, variable_values, operation_name=None) -> Dict[str, Any]: ...
226
def update_schema_scalar(schema, name, scalar) -> None: ...
227
def update_schema_enum(schema, name, values, use_enum_values=False) -> None: ...
228
```
229
230
[Schema Utilities](./utilities.md)
231
232
### Command Line Interface
233
234
Full-featured CLI for executing GraphQL queries, downloading schemas, and interacting with GraphQL servers. Supports multiple transport protocols, authentication methods, and output formatting.
235
236
```python { .api }
237
def gql_cli() -> None: ...
238
def get_parser(with_examples: bool = False) -> ArgumentParser: ...
239
def get_transport(args: Namespace) -> Optional[AsyncTransport]: ...
240
def autodetect_transport(url: URL) -> str: ...
241
```
242
243
Available as `gql-cli` command with comprehensive options for server interaction, authentication, and schema management.
244
245
[Command Line Interface](./cli.md)
246
247
## Types
248
249
Core types used across the gql library:
250
251
```python { .api }
252
from typing import Union, Optional, Dict, List, Any, AsyncGenerator
253
from graphql import GraphQLSchema, ExecutionResult, DocumentNode, IntrospectionQuery
254
255
# Core request type
256
class GraphQLRequest:
257
document: DocumentNode
258
variable_values: Optional[Dict[str, Any]]
259
operation_name: Optional[str]
260
261
@property
262
def payload(self) -> Dict[str, Any]: ...
263
264
# Transport base classes
265
class Transport:
266
def execute(self, request: GraphQLRequest, *args, **kwargs) -> ExecutionResult: ...
267
def execute_batch(self, reqs: List[GraphQLRequest], *args, **kwargs) -> List[ExecutionResult]: ...
268
269
class AsyncTransport:
270
async def execute(self, request: GraphQLRequest) -> ExecutionResult: ...
271
async def execute_batch(self, reqs: List[GraphQLRequest], *args, **kwargs) -> List[ExecutionResult]: ...
272
def subscribe(self, request: GraphQLRequest) -> AsyncGenerator[ExecutionResult, None]: ...
273
274
# File upload type
275
class FileVar:
276
f: Any
277
filename: Optional[str]
278
content_type: Optional[str]
279
streaming: bool
280
streaming_block_size: int
281
```