0
# ASGI Integration
1
2
ASGI application for serving GraphQL APIs with WebSocket support for subscriptions and integration with ASGI servers like Uvicorn, Gunicorn, and Hypercorn.
3
4
## Capabilities
5
6
### GraphQL ASGI Application
7
8
Main ASGI application class for serving GraphQL APIs with HTTP and WebSocket support.
9
10
```python { .api }
11
class GraphQL:
12
"""ASGI application for GraphQL APIs."""
13
14
def __init__(
15
self,
16
schema: GraphQLSchema,
17
*,
18
context_value: Optional[Any] = None,
19
root_value: Optional[Any] = None,
20
debug: bool = False,
21
explorer: Optional[Explorer] = None,
22
extensions: Optional[list[Extension]] = None,
23
middleware: Optional[Middlewares] = None,
24
error_formatter: Optional[Callable] = None,
25
validation_rules: Optional[list] = None,
26
require_query: bool = False,
27
websocket_handler: Optional[WebSocketHandler] = None,
28
on_connect: Optional[OnConnect] = None,
29
on_disconnect: Optional[OnDisconnect] = None,
30
on_operation: Optional[OnOperation] = None,
31
on_complete: Optional[OnComplete] = None,
32
keepalive: Optional[float] = None,
33
**kwargs
34
):
35
"""
36
Initialize GraphQL ASGI application.
37
38
Parameters:
39
- schema: Executable GraphQL schema
40
- context_value: Context value or factory function
41
- root_value: Root value for GraphQL execution
42
- debug: Enable debug mode
43
- explorer: GraphQL explorer for development
44
- extensions: List of GraphQL extensions
45
- middleware: ASGI middleware
46
- error_formatter: Custom error formatting function
47
- validation_rules: Custom validation rules
48
- require_query: Require 'query' in request data
49
- websocket_handler: Custom WebSocket handler
50
- on_connect: WebSocket connection callback
51
- on_disconnect: WebSocket disconnection callback
52
- on_operation: GraphQL operation callback
53
- on_complete: Operation completion callback
54
- keepalive: WebSocket keepalive interval
55
"""
56
```
57
58
### WebSocket Types and Callbacks
59
60
Type definitions for WebSocket operations and lifecycle callbacks.
61
62
```python { .api }
63
# Callback types
64
OnConnect = Callable[[dict], Union[dict, bool, Awaitable[Union[dict, bool]]]]
65
OnDisconnect = Callable[[dict], Union[None, Awaitable[None]]]
66
OnOperation = Callable[[dict, dict], Union[dict, Awaitable[dict]]]
67
OnComplete = Callable[[dict, dict], Union[None, Awaitable[None]]]
68
69
# Data types
70
Operation = dict[str, Any]
71
Extensions = dict[str, Any]
72
MiddlewareList = list[Callable]
73
Middlewares = Union[MiddlewareList, Callable]
74
75
# Exception types
76
class WebSocketConnectionError(Exception):
77
"""Exception raised for WebSocket connection errors."""
78
```
79
80
## Usage Examples
81
82
### Basic ASGI Application
83
84
```python
85
from ariadne import make_executable_schema, gql, QueryType
86
from ariadne.asgi import GraphQL
87
88
# Create schema
89
type_defs = gql("""
90
type Query {
91
hello: String!
92
}
93
""")
94
95
query = QueryType()
96
97
@query.field("hello")
98
def resolve_hello(*_):
99
return "Hello, World!"
100
101
schema = make_executable_schema(type_defs, query)
102
103
# Create ASGI app
104
app = GraphQL(schema, debug=True)
105
106
# Run with uvicorn
107
# uvicorn app:app --reload
108
```
109
110
### ASGI App with Context Factory
111
112
```python
113
from ariadne.asgi import GraphQL
114
import asyncio
115
116
async def get_context_value(request):
117
"""Context factory function."""
118
return {
119
"request": request,
120
"user": await get_current_user(request),
121
"db": database_connection
122
}
123
124
app = GraphQL(
125
schema,
126
context_value=get_context_value,
127
debug=False
128
)
129
```
130
131
### WebSocket Subscriptions
132
133
```python
134
from ariadne.asgi import GraphQL
135
136
async def on_connect(websocket, connection_params):
137
"""Called when WebSocket connection is established."""
138
print(f"WebSocket connected: {connection_params}")
139
# Return False to reject connection
140
# Return dict to set connection context
141
return {"user_id": connection_params.get("user_id")}
142
143
async def on_disconnect(websocket, connection_context):
144
"""Called when WebSocket connection is closed."""
145
print(f"WebSocket disconnected: {connection_context}")
146
147
async def on_operation(websocket, connection_context, operation):
148
"""Called for each GraphQL operation."""
149
print(f"GraphQL operation: {operation['type']}")
150
return operation
151
152
app = GraphQL(
153
schema,
154
on_connect=on_connect,
155
on_disconnect=on_disconnect,
156
on_operation=on_operation,
157
keepalive=10.0 # Send keepalive every 10 seconds
158
)
159
```
160
161
### Custom Error Handling
162
163
```python
164
from ariadne.asgi import GraphQL
165
166
def custom_error_formatter(error, debug=False):
167
"""Custom error formatter."""
168
formatted = {
169
"message": str(error),
170
"code": getattr(error, "code", "INTERNAL_ERROR")
171
}
172
173
if debug:
174
formatted["locations"] = getattr(error, "locations", None)
175
formatted["path"] = getattr(error, "path", None)
176
177
return formatted
178
179
app = GraphQL(
180
schema,
181
error_formatter=custom_error_formatter,
182
debug=True
183
)
184
```
185
186
### ASGI Middleware Integration
187
188
```python
189
from ariadne.asgi import GraphQL
190
from starlette.middleware.cors import CORSMiddleware
191
from starlette.applications import Starlette
192
193
# Create GraphQL app
194
graphql_app = GraphQL(schema)
195
196
# Wrap with Starlette for middleware
197
app = Starlette()
198
app.add_middleware(
199
CORSMiddleware,
200
allow_origins=["*"],
201
allow_methods=["GET", "POST"],
202
allow_headers=["*"],
203
)
204
205
app.mount("/graphql", graphql_app)
206
```