0
# Data Loading and Utilities
1

2
Utilities for efficient data loading, file uploads, type conversions, and common GraphQL patterns. These utilities help solve common problems in GraphQL API development like the N+1 query problem and provide helpful tools for schema manipulation.
3

4
## Capabilities
5

6
### DataLoader
7

8
Batch loading utility for efficient data fetching and N+1 query problem prevention.
9

10
```python { .api }
11
class DataLoader:
12
    """Batch loading utility for efficient data fetching."""
13
    
14
    def __init__(
15
        self,
16
        load_fn: Callable[[List[Any]], Awaitable[List[Any]]],
17
        *,
18
        batch: bool = True,
19
        max_batch_size: int = None,
20
        batch_scheduler: Callable = None,
21
        cache: bool = True,
22
        cache_key_fn: Callable[[Any], Any] = None,
23
        cache_map: AbstractCache = None
24
    ):
25
        """
26
        Initialize DataLoader.
27
        
28
        Args:
29
            load_fn: Async function that takes list of keys and returns list of values
30
            batch: Whether to batch requests
31
            max_batch_size: Maximum size of batches
32
            batch_scheduler: Custom batch scheduling function
33
            cache: Whether to cache results
34
            cache_key_fn: Function to generate cache keys
35
            cache_map: Custom cache implementation
36
        """
37
    
38
    async def load(self, key: Any) -> Any:
39
        """
40
        Load single value by key.
41
        
42
        Args:
43
            key: Key to load
44
        
45
        Returns:
46
            Loaded value
47
        """
48
    
49
    async def load_many(self, keys: List[Any]) -> List[Any]:
50
        """
51
        Load multiple values by keys.
52
        
53
        Args:
54
            keys: List of keys to load
55
        
56
        Returns:
57
            List of loaded values in same order as keys
58
        """
59
    
60
    def clear(self, key: Any) -> "DataLoader":
61
        """
62
        Clear cache for specific key.
63
        
64
        Args:
65
            key: Key to clear from cache
66
        
67
        Returns:
68
            DataLoader instance for chaining
69
        """
70
    
71
    def clear_all(self) -> "DataLoader":
72
        """
73
        Clear all cached values.
74
        
75
        Returns:
76
            DataLoader instance for chaining
77
        """
78
    
79
    def prime(self, key: Any, value: Any) -> "DataLoader":
80
        """
81
        Prime cache with key-value pair.
82
        
83
        Args:
84
            key: Key to prime
85
            value: Value to cache
86
        
87
        Returns:
88
            DataLoader instance for chaining
89
        """
90
```
91

92
**Usage Example:**
93

94
```python
95
import asyncio
96
from strawberry.dataloader import DataLoader
97

98
# Define batch loading function
99
async def load_users_by_ids(user_ids: List[str]) -> List[Optional[User]]:
100
    """Load users by IDs from database."""
101
    print(f"Batch loading users: {user_ids}")  # This should only print once per batch
102
    
103
    # Simulate database query
104
    users_data = await database.fetch_users_by_ids(user_ids)
105
    
106
    # Create lookup dictionary
107
    user_lookup = {str(user.id): user for user in users_data}
108
    
109
    # Return users in same order as requested IDs
110
    return [user_lookup.get(user_id) for user_id in user_ids]
111

112
# Create DataLoader
113
user_loader = DataLoader(load_users_by_ids)
114

115
# Usage in resolvers
116
@strawberry.type
117
class Post:
118
    id: strawberry.ID
119
    title: str
120
    author_id: str
121
    
122
    @strawberry.field
123
    async def author(self, info: strawberry.Info) -> Optional[User]:
124
        # DataLoader automatically batches these requests
125
        return await info.context.user_loader.load(self.author_id)
126

127
@strawberry.type
128
class Query:
129
    @strawberry.field
130
    async def posts(self, info: strawberry.Info) -> List[Post]:
131
        posts = await get_all_posts()
132
        # Even though we're loading author for each post,
133
        # DataLoader will batch all author requests into a single query
134
        return posts
135

136
# Context setup
137
class Context:
138
    def __init__(self):
139
        self.user_loader = DataLoader(load_users_by_ids)
140

141
async def get_context():
142
    return Context()
143
```
144

145
### DataLoader Cache Interface
146

147
Abstract cache interface for custom cache implementations.
148

149
```python { .api }
150
class AbstractCache:
151
    """Abstract interface for DataLoader cache implementations."""
152
    
153
    def get(self, key: Any) -> Any:
154
        """Get value from cache."""
155
    
156
    def set(self, key: Any, value: Any) -> None:
157
        """Set value in cache."""
158
    
159
    def delete(self, key: Any) -> None:
160
        """Delete value from cache."""
161
    
162
    def clear(self) -> None:
163
        """Clear all cache values."""
164
```
165

166
**Custom Cache Example:**
167

168
```python
169
import redis
170
from strawberry.dataloader import AbstractCache
171

172
class RedisCache(AbstractCache):
173
    def __init__(self, redis_client: redis.Redis, ttl: int = 300):
174
        self.redis = redis_client
175
        self.ttl = ttl
176
    
177
    def get(self, key):
178
        cached = self.redis.get(f"dataloader:{key}")
179
        return pickle.loads(cached) if cached else None
180
    
181
    def set(self, key, value):
182
        self.redis.setex(
183
            f"dataloader:{key}",
184
            self.ttl,
185
            pickle.dumps(value)
186
        )
187
    
188
    def delete(self, key):
189
        self.redis.delete(f"dataloader:{key}")
190
    
191
    def clear(self):
192
        for key in self.redis.scan_iter(match="dataloader:*"):
193
            self.redis.delete(key)
194

195
# Use custom cache
196
redis_client = redis.Redis()
197
user_loader = DataLoader(
198
    load_users_by_ids,
199
    cache_map=RedisCache(redis_client, ttl=600)
200
)
201
```
202

203
### DataLoader Internal Types
204

205
Internal types used by DataLoader implementation.
206

207
```python { .api }
208
class LoaderTask:
209
    """Internal task representation for DataLoader."""
210
    pass
211

212
class Batch:
213
    """Internal batch representation for DataLoader."""
214
    pass
215
```
216

217
### Advanced DataLoader Patterns
218

219
```python
220
# DataLoader with custom cache key function
221
def user_cache_key(user_id: str) -> str:
222
    return f"user:{user_id}"
223

224
user_loader = DataLoader(
225
    load_users_by_ids,
226
    cache_key_fn=user_cache_key
227
)
228

229
# DataLoader with limited batch size
230
large_data_loader = DataLoader(
231
    load_large_data,
232
    max_batch_size=50  # Limit batch size for large operations
233
)
234

235
# DataLoader without caching (for frequently changing data)
236
realtime_loader = DataLoader(
237
    load_realtime_data,
238
    cache=False
239
)
240

241
# Priming DataLoader cache
242
async def prime_user_cache(users: List[User]):
243
    for user in users:
244
        user_loader.prime(user.id, user)
245

246
# DataLoader composition
247
class UserService:
248
    def __init__(self):
249
        self.user_loader = DataLoader(self._load_users)
250
        self.user_profile_loader = DataLoader(self._load_profiles)
251
    
252
    async def _load_users(self, user_ids: List[str]) -> List[User]:
253
        return await database.get_users_by_ids(user_ids)
254
    
255
    async def _load_profiles(self, user_ids: List[str]) -> List[UserProfile]:
256
        return await database.get_profiles_by_user_ids(user_ids)
257
    
258
    async def get_user_with_profile(self, user_id: str) -> UserWithProfile:
259
        user, profile = await asyncio.gather(
260
            self.user_loader.load(user_id),
261
            self.user_profile_loader.load(user_id)
262
        )
263
        return UserWithProfile(user=user, profile=profile)
264
```
265

266
## File Uploads
267

268
File upload support for GraphQL mutations with multipart/form-data requests.
269

270
```python { .api }
271
class Upload:
272
    """File upload scalar type for multipart requests."""
273
    
274
    filename: str  # Original filename
275
    content_type: str  # MIME content type
276
    
277
    def read(self, size: int = -1) -> bytes:
278
        """Read file content."""
279
    
280
    async def read_async(self, size: int = -1) -> bytes:
281
        """Read file content asynchronously."""
282
    
283
    def seek(self, offset: int) -> None:
284
        """Seek to position in file."""
285
    
286
    def close(self) -> None:
287
        """Close file handle."""
288
```
289

290
**Usage Example:**
291

292
```python
293
from strawberry.file_uploads import Upload
294

295
@strawberry.type
296
class UploadResult:
297
    success: bool
298
    filename: str
299
    size: int
300
    url: Optional[str]
301

302
@strawberry.type
303
class Mutation:
304
    @strawberry.mutation
305
    async def upload_file(
306
        self,
307
        file: Upload,
308
        description: str = ""
309
    ) -> UploadResult:
310
        # Read file content
311
        content = await file.read_async()
312
        
313
        # Save file to storage
314
        file_path = f"uploads/{file.filename}"
315
        await save_file_to_storage(file_path, content)
316
        
317
        return UploadResult(
318
            success=True,
319
            filename=file.filename,
320
            size=len(content),
321
            url=f"/files/{file.filename}"
322
        )
323
    
324
    @strawberry.mutation
325
    async def upload_multiple_files(
326
        self,
327
        files: List[Upload]
328
    ) -> List[UploadResult]:
329
        results = []
330
        for file in files:
331
            content = await file.read_async()
332
            file_path = f"uploads/{file.filename}"
333
            await save_file_to_storage(file_path, content)
334
            
335
            results.append(UploadResult(
336
                success=True,
337
                filename=file.filename,
338
                size=len(content),
339
                url=f"/files/{file.filename}"
340
            ))
341
        
342
        return results
343
```
344

345
**Frontend Usage (JavaScript):**
346

347
```javascript
348
// Single file upload
349
const mutation = `
350
  mutation UploadFile($file: Upload!, $description: String) {
351
    uploadFile(file: $file, description: $description) {
352
      success
353
      filename
354
      size
355
      url
356
    }
357
  }
358
`;
359

360
const variables = {
361
  file: file, // File object from input element
362
  description: "My uploaded file"
363
};
364

365
// Multiple file upload
366
const multiMutation = `
367
  mutation UploadMultipleFiles($files: [Upload!]!) {
368
    uploadMultipleFiles(files: $files) {
369
      success
370
      filename
371
      size
372
    }
373
  }
374
`;
375
```
376

377
## Type Creation Utilities
378

379
Utilities for programmatically creating and manipulating GraphQL types.
380

381
```python { .api }
382
def create_type(
383
    name: str,
384
    fields: Dict[str, Any],
385
    *,
386
    description: str = None,
387
    interfaces: List[Type] = None,
388
    directives: List = None
389
) -> Type:
390
    """
391
    Programmatically create GraphQL types.
392
    
393
    Args:
394
        name: Type name
395
        fields: Dictionary of field name to field definition
396
        description: Type description
397
        interfaces: Interfaces this type implements
398
        directives: GraphQL directives to apply
399
    
400
    Returns:
401
        New GraphQL type
402
    """
403

404
def merge_types(
405
    name: str,
406
    types: List[Type],
407
    *,
408
    description: str = None
409
) -> Type:
410
    """
411
    Merge multiple GraphQL types into a single type.
412
    
413
    Args:
414
        name: New type name
415
        types: List of types to merge
416
        description: Merged type description
417
    
418
    Returns:
419
        Merged GraphQL type
420
    """
421
```
422

423
**Usage Examples:**
424

425
```python
426
from strawberry.tools import create_type, merge_types
427

428
# Create type programmatically
429
UserType = create_type(
430
    "User",
431
    {
432
        "id": strawberry.ID,
433
        "name": str,
434
        "email": str,
435
        "age": int
436
    },
437
    description="User account information"
438
)
439

440
# Create type with methods
441
def get_full_name(self) -> str:
442
    return f"{self.first_name} {self.last_name}"
443

444
PersonType = create_type(
445
    "Person",
446
    {
447
        "first_name": str,
448
        "last_name": str,
449
        "full_name": strawberry.field(resolver=get_full_name)
450
    }
451
)
452

453
# Merge multiple types
454
@strawberry.type
455
class BaseUser:
456
    id: strawberry.ID
457
    name: str
458

459
@strawberry.type
460
class UserPreferences:
461
    theme: str
462
    language: str
463

464
# Merge into single type
465
ExtendedUser = merge_types(
466
    "ExtendedUser",
467
    [BaseUser, UserPreferences],
468
    description="User with preferences"
469
)
470
```
471

472
## Schema Utilities
473

474
### Schema Printing
475

476
Print GraphQL schema in SDL (Schema Definition Language) format.
477

478
```python { .api }
479
def print_schema(schema: Schema) -> str:
480
    """
481
    Print GraphQL schema as SDL string.
482
    
483
    Args:
484
        schema: GraphQL schema to print
485
    
486
    Returns:
487
        Schema Definition Language string
488
    """
489
```
490

491
**Usage Example:**
492

493
```python
494
from strawberry.printer import print_schema
495

496
schema = strawberry.Schema(query=Query, mutation=Mutation)
497
sdl = print_schema(schema)
498

499
print(sdl)
500
# Output:
501
# type Query {
502
#   users: [User!]!
503
#   user(id: ID!): User
504
# }
505
# 
506
# type User {
507
#   id: ID!
508
#   name: String!
509
#   email: String!
510
# }
511
```
512

513
## Subscription Utilities
514

515
Constants and utilities for GraphQL subscriptions.
516

517
```python { .api }
518
# WebSocket protocol constants
519
GRAPHQL_TRANSPORT_WS_PROTOCOL: str  # Modern GraphQL-WS transport protocol
520
GRAPHQL_WS_PROTOCOL: str  # Legacy GraphQL-WS protocol
521
```
522

523
**Usage Example:**
524

525
```python
526
from strawberry.subscriptions import (
527
    GRAPHQL_TRANSPORT_WS_PROTOCOL,
528
    GRAPHQL_WS_PROTOCOL
529
)
530

531
# Use in ASGI app
532
app = strawberry.asgi.GraphQL(
533
    schema,
534
    subscription_protocols=[
535
        GRAPHQL_TRANSPORT_WS_PROTOCOL,
536
        GRAPHQL_WS_PROTOCOL  # For backwards compatibility
537
    ]
538
)
539
```
540

541
## Code Generation Utilities
542

543
### Schema Code Generation
544

545
Generate Python code from GraphQL queries.
546

547
```python { .api }
548
class CodegenFile:
549
    """Generated code file."""
550
    
551
    path: str
552
    content: str
553

554
class CodegenResult:
555
    """Result of code generation."""
556
    
557
    files: List[CodegenFile]
558

559
class QueryCodegen:
560
    """Generate code from GraphQL queries."""
561
    
562
    def __init__(self, schema: Schema): ...
563
    
564
    def generate(
565
        self,
566
        query: str,
567
        *,
568
        target_language: str = "python",
569
        plugins: List[QueryCodegenPlugin] = None
570
    ) -> CodegenResult: ...
571

572
class QueryCodegenPlugin:
573
    """Base class for codegen plugins."""
574
    pass
575

576
class ConsolePlugin(QueryCodegenPlugin):
577
    """Plugin for console output."""
578
    pass
579
```
580

581
### SDL Code Generation
582

583
Generate Strawberry code from GraphQL SDL.
584

585
```python { .api }
586
def codegen(
587
    schema_sdl: str,
588
    *,
589
    output_dir: str = None,
590
    plugins: List[str] = None
591
) -> None:
592
    """
593
    Generate Strawberry code from GraphQL SDL.
594
    
595
    Args:
596
        schema_sdl: GraphQL Schema Definition Language string
597
        output_dir: Directory to write generated files
598
        plugins: List of codegen plugins to use
599
    """
600
```
601

602
**Usage Example:**
603

604
```python
605
from strawberry.codegen import codegen
606

607
schema_sdl = """
608
type User {
609
  id: ID!
610
  name: String!
611
  email: String!
612
  posts: [Post!]!
613
}
614

615
type Post {
616
  id: ID!
617
  title: String!
618
  content: String!
619
  author: User!
620
}
621

622
type Query {
623
  users: [User!]!
624
  posts: [Post!]!
625
}
626
"""
627

628
# Generate Strawberry code from SDL
629
codegen(
630
    schema_sdl,
631
    output_dir="./generated",
632
    plugins=["strawberry"]
633
)
634
```
635

636
## CLI Utilities
637

638
Command-line interface for Strawberry operations.
639

640
```python { .api }
641
def run() -> None:
642
    """Main CLI application entry point."""
643

644
# Available commands:
645
# strawberry server - Development server
646
# strawberry export-schema - Export schema SDL
647
# strawberry codegen - Generate code from SDL
648
# strawberry schema-codegen - Generate types from queries
649
```
650

651
**CLI Usage Examples:**
652

653
```bash
654
# Start development server
655
strawberry server myapp.schema:schema --host 0.0.0.0 --port 8000
656

657
# Export schema to file
658
strawberry export-schema myapp.schema:schema --output schema.graphql
659

660
# Generate code from SDL
661
strawberry codegen --schema schema.graphql --output generated/
662

663
# Generate types from queries
664
strawberry schema-codegen --schema myapp.schema:schema --queries queries/ --output types.py
665
```
666

667
## Performance Utilities
668

669
### Caching Patterns
670

671
```python
672
from functools import lru_cache
673
import asyncio
674

675
class FieldCache:
676
    """Simple field-level caching utility."""
677
    
678
    def __init__(self, ttl: int = 300):
679
        self.ttl = ttl
680
        self.cache = {}
681
    
682
    def cached_field(self, ttl: int = None):
683
        def decorator(func):
684
            @functools.wraps(func)
685
            async def wrapper(*args, **kwargs):
686
                cache_key = f"{func.__name__}:{hash(str(args + tuple(kwargs.items())))}"
687
                
688
                if cache_key in self.cache:
689
                    value, timestamp = self.cache[cache_key]
690
                    if time.time() - timestamp < (ttl or self.ttl):
691
                        return value
692
                
693
                result = await func(*args, **kwargs)
694
                self.cache[cache_key] = (result, time.time())
695
                return result
696
            
697
            return wrapper
698
        return decorator
699

700
# Usage
701
field_cache = FieldCache(ttl=600)
702

703
@strawberry.type
704
class User:
705
    id: strawberry.ID
706
    name: str
707
    
708
    @field_cache.cached_field(ttl=300)
709
    @strawberry.field
710
    async def expensive_computation(self) -> str:
711
        # Expensive operation that benefits from caching
712
        await asyncio.sleep(1)  # Simulate expensive operation
713
        return f"computed_value_for_{self.id}"
714
```