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
            email
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
```