0
# Schema and Execution
1

2
Schema creation and query execution system with support for queries, mutations, subscriptions, and context management. The Schema class is the central component that orchestrates GraphQL operation execution.
3

4
## Capabilities
5

6
### Schema Class
7

8
Main schema class for defining and executing GraphQL operations.
9

10
```python { .api }
11
class Schema:
12
    """Main GraphQL schema class for executing queries, mutations, and subscriptions."""
13
    
14
    def __init__(
15
        self,
16
        query: Type = None,
17
        mutation: Type = None,
18
        subscription: Type = None,
19
        *,
20
        types: List[Type] = None,
21
        extensions: List[SchemaExtension] = None,
22
        execution_context_class: Type = None,
23
        introspection: bool = True,
24
        auto_camel_case: bool = True
25
    ):
26
        """
27
        Initialize a GraphQL schema.
28
        
29
        Args:
30
            query: Root query type
31
            mutation: Root mutation type (optional)
32
            subscription: Root subscription type (optional)
33
            types: Additional types to include in schema
34
            extensions: Schema extensions for custom functionality
35
            execution_context_class: Custom execution context class
36
            introspection: Whether to enable introspection
37
            auto_camel_case: Whether to auto-convert field names to camelCase
38
        """
39
    
40
    # Schema properties
41
    query_type: Type  # Root query type
42
    mutation_type: Optional[Type]  # Root mutation type
43
    subscription_type: Optional[Type]  # Root subscription type
44
    types: List[Type]  # All types in the schema
45
```
46

47
**Usage Example:**
48

49
```python
50
@strawberry.type
51
class Query:
52
    @strawberry.field
53
    def hello(self) -> str:
54
        return "Hello, World!"
55

56
@strawberry.type
57
class Mutation:
58
    @strawberry.mutation
59
    def create_user(self, name: str) -> User:
60
        return User(id="1", name=name)
61

62
@strawberry.type
63
class Subscription:
64
    @strawberry.subscription
65
    async def user_updates(self) -> AsyncIterator[User]:
66
        # Subscription logic here
67
        yield User(id="1", name="Alice")
68

69
# Create schema
70
schema = strawberry.Schema(
71
    query=Query,
72
    mutation=Mutation,
73
    subscription=Subscription,
74
    extensions=[
75
        QueryDepthLimiter(max_depth=10),
76
        ValidationCache()
77
    ]
78
)
79
```
80

81
### Synchronous Execution
82

83
Execute GraphQL queries synchronously.
84

85
```python { .api }
86
def execute_sync(
87
    self,
88
    query: str,
89
    variable_values: Dict[str, Any] = None,
90
    context_value: Any = None,
91
    root_value: Any = None,
92
    operation_name: str = None
93
) -> ExecutionResult:
94
    """
95
    Execute a GraphQL query synchronously.
96
    
97
    Args:
98
        query: GraphQL query string
99
        variable_values: Variables for the query
100
        context_value: Context passed to resolvers
101
        root_value: Root value for top-level resolvers
102
        operation_name: Name of operation to execute (for multi-operation queries)
103
    
104
    Returns:
105
        ExecutionResult with data and errors
106
    """
107
```
108

109
**Usage Example:**
110

111
```python
112
# Simple query execution
113
result = schema.execute_sync("""
114
    query {
115
        hello
116
    }
117
""")
118

119
print(result.data)  # {'hello': 'Hello, World!'}
120

121
# Query with variables
122
result = schema.execute_sync(
123
    """
124
    query GetUser($id: ID!) {
125
        user(id: $id) {
126
            name
127
            email
128
        }
129
    }
130
    """,
131
    variable_values={"id": "123"}
132
)
133

134
# Query with context
135
class Context:
136
    def __init__(self, user=None):
137
        self.user = user
138

139
result = schema.execute_sync(
140
    """
141
    query {
142
        currentUser {
143
            name
144
        }
145
    }
146
    """,
147
    context_value=Context(user=current_user)
148
)
149
```
150

151
### Asynchronous Execution
152

153
Execute GraphQL queries asynchronously for better performance with async resolvers.
154

155
```python { .api }
156
async def execute(
157
    self,
158
    query: str,
159
    variable_values: Dict[str, Any] = None,
160
    context_value: Any = None,
161
    root_value: Any = None,
162
    operation_name: str = None
163
) -> ExecutionResult:
164
    """
165
    Execute a GraphQL query asynchronously.
166
    
167
    Args:
168
        query: GraphQL query string
169
        variable_values: Variables for the query
170
        context_value: Context passed to resolvers
171
        root_value: Root value for top-level resolvers
172
        operation_name: Name of operation to execute
173
    
174
    Returns:
175
        ExecutionResult with data and errors
176
    """
177
```
178

179
**Usage Example:**
180

181
```python
182
import asyncio
183

184
async def main():
185
    # Async query execution
186
    result = await schema.execute("""
187
        query {
188
            users {
189
                id
190
                name
191
                posts {
192
                    title
193
                }
194
            }
195
        }
196
    """)
197
    
198
    if result.errors:
199
        print("Errors:", result.errors)
200
    else:
201
        print("Data:", result.data)
202

203
# Run async execution
204
asyncio.run(main())
205
```
206

207
### Subscription Execution
208

209
Execute GraphQL subscriptions for real-time data streaming.
210

211
```python { .api }
212
async def subscribe(
213
    self,
214
    query: str,
215
    variable_values: Dict[str, Any] = None,
216
    context_value: Any = None,
217
    root_value: Any = None,
218
    operation_name: str = None
219
) -> AsyncIterator[ExecutionResult]:
220
    """
221
    Execute a GraphQL subscription.
222
    
223
    Args:
224
        query: GraphQL subscription string
225
        variable_values: Variables for the subscription
226
        context_value: Context passed to resolvers
227
        root_value: Root value for top-level resolvers
228
        operation_name: Name of operation to execute
229
    
230
    Yields:
231
        ExecutionResult for each subscription event
232
    """
233
```
234

235
**Usage Example:**
236

237
```python
238
async def handle_subscription():
239
    subscription_query = """
240
        subscription {
241
            messageUpdates(roomId: "123") {
242
                id
243
                content
244
                author {
245
                    name
246
                }
247
                timestamp
248
            }
249
        }
250
    """
251
    
252
    async for result in schema.subscribe(subscription_query):
253
        if result.errors:
254
            print("Subscription error:", result.errors)
255
        else:
256
            message = result.data["messageUpdates"]
257
            print(f"New message: {message['content']} from {message['author']['name']}")
258
```
259

260
## Execution Results
261

262
### ExecutionResult
263

264
Result object returned by query and mutation execution.
265

266
```python { .api }
267
class ExecutionResult:
268
    """Result of GraphQL query or mutation execution."""
269
    
270
    data: Optional[Dict[str, Any]]  # Query result data
271
    errors: Optional[List[GraphQLError]]  # Execution errors
272
    extensions: Optional[Dict[str, Any]]  # Extension data
273
```
274

275
**Usage Example:**
276

277
```python
278
result = schema.execute_sync("""
279
    query {
280
        user(id: "invalid") {
281
            name
282
        }
283
    }
284
""")
285

286
if result.errors:
287
    for error in result.errors:
288
        print(f"Error: {error.message}")
289
        print(f"Path: {error.path}")
290
        print(f"Locations: {error.locations}")
291
else:
292
    print(f"Data: {result.data}")
293

294
# Access extensions (from schema extensions)
295
if result.extensions:
296
    print(f"Query complexity: {result.extensions.get('complexity')}")
297
    print(f"Execution time: {result.extensions.get('execution_time')}")
298
```
299

300
### SubscriptionExecutionResult
301

302
Result object for subscription execution with async iteration support.
303

304
```python { .api }
305
class SubscriptionExecutionResult:
306
    """Result of GraphQL subscription execution."""
307
    
308
    def __aiter__(self) -> AsyncIterator[ExecutionResult]:
309
        """Async iterator for subscription results."""
310
    
311
    async def __anext__(self) -> ExecutionResult:
312
        """Get next subscription result."""
313
```
314

315
## Execution Context
316

317
### ExecutionContext
318

319
Context object that manages the execution of a GraphQL operation.
320

321
```python { .api }
322
class ExecutionContext:
323
    """Context for GraphQL operation execution."""
324
    
325
    schema: Schema  # GraphQL schema
326
    document: DocumentNode  # Parsed GraphQL document
327
    root_value: Any  # Root value for resolvers
328
    context_value: Any  # Context passed to resolvers
329
    variable_values: Dict[str, Any]  # Operation variables
330
    operation_name: Optional[str]  # Name of operation being executed
331
    field_resolver: Callable  # Field resolver function
332
    type_resolver: Callable  # Type resolver function
333
    subscribe_field_resolver: Callable  # Subscription field resolver
334
```
335

336
## Advanced Schema Configuration
337

338
### Custom Types Registration
339

340
```python { .api }
341
# Register additional types that might not be directly referenced
342
@strawberry.type
343
class AdminUser:
344
    id: strawberry.ID
345
    name: str
346
    admin_level: int
347

348
schema = strawberry.Schema(
349
    query=Query,
350
    types=[AdminUser]  # Ensure AdminUser is included in schema
351
)
352
```
353

354
### Schema Introspection Control
355

356
```python { .api }
357
# Disable introspection for production
358
production_schema = strawberry.Schema(
359
    query=Query,
360
    introspection=False  # Disable introspection queries
361
)
362

363
# Or use extension
364
from strawberry.extensions import DisableIntrospection
365

366
schema = strawberry.Schema(
367
    query=Query,
368
    extensions=[DisableIntrospection()]
369
)
370
```
371

372
### Custom Field Name Conversion
373

374
```python { .api }
375
# Disable automatic camelCase conversion
376
schema = strawberry.Schema(
377
    query=Query,
378
    auto_camel_case=False  # Keep original Python field names
379
)
380

381
@strawberry.type
382
class User:
383
    user_id: strawberry.ID  # Will remain as 'user_id' in GraphQL
384
    first_name: str  # Will remain as 'first_name' in GraphQL
385
```
386

387
### Error Handling
388

389
```python { .api }
390
class CustomError(Exception):
391
    """Custom application error."""
392
    def __init__(self, message: str, code: str = None):
393
        self.message = message
394
        self.code = code
395
        super().__init__(message)
396

397
@strawberry.type
398
class Query:
399
    @strawberry.field
400
    def risky_operation(self) -> str:
401
        try:
402
            return perform_risky_operation()
403
        except CustomError as e:
404
            # Errors are automatically added to the result
405
            raise e
406
        except Exception as e:
407
            # Generic error handling
408
            raise GraphQLError(f"Operation failed: {str(e)}")
409

410
# Execute with error handling
411
result = schema.execute_sync("""
412
    query {
413
        riskyOperation
414
    }
415
""")
416

417
if result.errors:
418
    for error in result.errors:
419
        print(f"Error: {error.message}")
420
        if hasattr(error.original_error, 'code'):
421
            print(f"Code: {error.original_error.code}")
422
```
423

424
### Schema Printing
425

426
```python { .api }
427
from strawberry.printer import print_schema
428

429
# Print schema SDL (Schema Definition Language)
430
sdl = print_schema(schema)
431
print(sdl)
432
```
433

434
**Example Output:**
435

436
```graphql
437
type Query {
438
  hello: String!
439
  user(id: ID!): User
440
}
441

442
type User {
443
  id: ID!
444
  name: String!
445
  email: String!
446
}
447

448
type Mutation {
449
  createUser(name: String!): User!
450
}
451
```
452

453
### Multi-Operation Queries
454

455
```python { .api }
456
# Execute specific operation from multi-operation query
457
multi_operation_query = """
458
    query GetUsers {
459
        users { id name }
460
    }
461
    
462
    query GetPosts {
463
        posts { id title }
464
    }
465
"""
466

467
# Execute specific operation
468
result = schema.execute_sync(
469
    multi_operation_query,
470
    operation_name="GetUsers"
471
)
472
```