0
# Utility Functions
1
2
SDK utility functions for logging, module imports, and internal functionality. These utilities provide debugging helpers, service context management, and internal mechanisms used by the boto3 framework itself.
3
4
## Capabilities
5
6
### Logging Configuration
7
8
Functions for configuring boto3 logging to help with debugging and monitoring AWS API calls.
9
10
```python { .api }
11
def set_stream_logger(name: str = 'boto3', level: int = logging.DEBUG,
12
format_string: str = None) -> None:
13
"""
14
Add a stream handler for the given name and level to the logging module.
15
16
By default, this logs all boto3 messages to stdout. This is useful for
17
debugging AWS API interactions and understanding what requests are being made.
18
19
Parameters:
20
- name: Logger name (default: 'boto3'). Use '' to log everything including botocore
21
- level: Logging level (e.g., logging.INFO, logging.DEBUG)
22
- format_string: Custom log message format string
23
24
Warning:
25
When logging anything from 'botocore', the full wire trace will appear
26
in your logs. If your payloads contain sensitive data, this should not
27
be used in production.
28
"""
29
```
30
31
### Service Context Management
32
33
Classes for managing service-wide information and metadata used internally by boto3's resource system.
34
35
```python { .api }
36
from boto3.utils import ServiceContext, LazyLoadedWaiterModel
37
38
class ServiceContext:
39
"""
40
Container for service-wide, read-only information about an AWS service.
41
42
This class holds metadata and models needed for resource creation and
43
provides context about the service being used.
44
"""
45
46
def __init__(self, service_name: str, service_model,
47
service_waiter_model, resource_json_definitions: Dict[str, Any]):
48
"""
49
Parameters:
50
- service_name: Name of the AWS service
51
- service_model: Botocore service model for the service
52
- service_waiter_model: Waiter model for the service
53
- resource_json_definitions: JSON definitions for service resources
54
"""
55
self.service_name = service_name
56
self.service_model = service_model
57
self.service_waiter_model = service_waiter_model
58
self.resource_json_definitions = resource_json_definitions
59
60
class LazyLoadedWaiterModel:
61
"""
62
A lazily loaded waiter model for AWS services.
63
64
This class defers loading of waiter models until they are actually needed,
65
which improves startup performance and reduces memory usage when waiters
66
are not used.
67
"""
68
69
def __init__(self, bc_session, service_name: str, api_version: str):
70
"""
71
Parameters:
72
- bc_session: Botocore session instance
73
- service_name: AWS service name
74
- api_version: API version for the service
75
"""
76
self._session = bc_session
77
self._service_name = service_name
78
self._api_version = api_version
79
80
def get_waiter(self, waiter_name: str):
81
"""
82
Get a specific waiter by name.
83
84
Parameters:
85
- waiter_name: Name of the waiter to retrieve
86
87
Returns:
88
Botocore waiter instance
89
"""
90
```
91
92
### Module and Function Utilities
93
94
Internal utility functions for dynamic module importing and lazy function loading.
95
96
```python { .api }
97
from boto3.utils import import_module, lazy_call, inject_attribute
98
99
def import_module(name: str):
100
"""
101
Import a module by its string name.
102
103
This function provides a simple way to import modules dynamically
104
at runtime, which is used internally by boto3 for loading service-specific
105
functionality.
106
107
Parameters:
108
- name: Full module name (e.g., 'boto3.s3.transfer')
109
110
Returns:
111
The imported module object
112
113
Note:
114
Does not support relative imports.
115
"""
116
117
def lazy_call(full_name: str, **kwargs) -> Callable:
118
"""
119
Create a lazy-loaded function call.
120
121
Returns a function that will import the specified module and call
122
the specified function only when the returned function is actually called.
123
This is used internally for performance optimization.
124
125
Parameters:
126
- full_name: Full function name including module (e.g., 'module.function')
127
- **kwargs: Additional keyword arguments to pass to the function
128
129
Returns:
130
Function that will perform the lazy import and call when invoked
131
"""
132
133
def inject_attribute(class_attributes: Dict[str, Any], name: str, value: Any) -> None:
134
"""
135
Inject an attribute into a class dictionary.
136
137
Used internally by boto3 to dynamically add methods and attributes
138
to service classes during resource creation.
139
140
Parameters:
141
- class_attributes: Class attribute dictionary (typically cls.__dict__)
142
- name: Attribute name to inject
143
- value: Attribute value to inject
144
145
Raises:
146
RuntimeError: If the attribute already exists in the class
147
"""
148
```
149
150
### Compatibility Functions
151
152
Functions for handling compatibility issues and platform-specific behavior.
153
154
```python { .api }
155
from boto3.compat import filter_python_deprecation_warnings, rename_file, is_append_mode
156
157
def filter_python_deprecation_warnings() -> None:
158
"""
159
Filter Python deprecation warnings from boto3.
160
161
This function configures the warning system to suppress deprecation
162
warnings related to Python versions that boto3 will stop supporting.
163
"""
164
165
def rename_file(current_filename: str, new_filename: str) -> None:
166
"""
167
Cross-platform file rename operation.
168
169
Provides a consistent interface for renaming files across different
170
operating systems, handling platform-specific edge cases.
171
172
Parameters:
173
- current_filename: Current file path
174
- new_filename: New file path
175
"""
176
177
def is_append_mode(fileobj) -> bool:
178
"""
179
Check if a file object is opened in append mode.
180
181
Used internally to determine how to handle file objects passed
182
to various boto3 operations.
183
184
Parameters:
185
- fileobj: File-like object to check
186
187
Returns:
188
True if the file object is in append mode, False otherwise
189
"""
190
```
191
192
### Error Compatibility
193
194
Compatibility constants for handling different types of socket and network errors across Python versions.
195
196
```python { .api }
197
from boto3.compat import SOCKET_ERROR
198
199
SOCKET_ERROR = ConnectionError # Standard socket error type for network issues
200
```
201
202
## Usage Examples
203
204
### Basic Logging Setup
205
206
```python
207
import boto3
208
import logging
209
210
# Enable basic boto3 logging
211
boto3.set_stream_logger('boto3', logging.INFO)
212
213
# Create a client - you'll now see log messages
214
s3_client = boto3.client('s3')
215
s3_client.list_buckets() # This will log the API call
216
```
217
218
### Detailed Debug Logging
219
220
```python
221
import boto3
222
import logging
223
224
# Enable detailed debug logging (includes wire traces)
225
boto3.set_stream_logger('', logging.DEBUG) # Empty string logs everything
226
227
# Custom format for log messages
228
boto3.set_stream_logger(
229
'boto3.resources',
230
logging.INFO,
231
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
232
)
233
234
# Now all AWS operations will show detailed debugging information
235
dynamodb = boto3.resource('dynamodb')
236
table = dynamodb.Table('my-table')
237
table.scan() # This will show extensive debug information
238
```
239
240
### Selective Logging
241
242
```python
243
import boto3
244
import logging
245
246
# Log only specific components
247
boto3.set_stream_logger('boto3.s3', logging.DEBUG) # S3 operations only
248
boto3.set_stream_logger('boto3.dynamodb', logging.INFO) # DynamoDB at info level
249
250
# Regular boto3 usage
251
s3 = boto3.client('s3') # Will show debug logs
252
dynamodb = boto3.client('dynamodb') # Will show info logs
253
ec2 = boto3.client('ec2') # No extra logging
254
```
255
256
### Working with Service Context (Advanced)
257
258
```python
259
import boto3
260
from boto3.utils import ServiceContext
261
262
# This is typically used internally, but can be accessed if needed
263
session = boto3.Session()
264
dynamodb_resource = session.resource('dynamodb')
265
266
# Access the service context (for advanced use cases)
267
service_context = dynamodb_resource.meta.service_model
268
print(f"Service name: {service_context.service_name}")
269
print(f"API version: {service_context.api_version}")
270
```
271
272
### Dynamic Module Loading (Advanced)
273
274
```python
275
from boto3.utils import import_module
276
277
# Dynamically import modules at runtime
278
try:
279
s3_module = import_module('boto3.s3.transfer')
280
transfer_config = s3_module.TransferConfig()
281
print("S3 transfer module loaded successfully")
282
except ImportError as e:
283
print(f"Failed to load module: {e}")
284
```
285
286
### Handling Platform Compatibility
287
288
```python
289
from boto3.compat import rename_file, is_append_mode
290
291
# Cross-platform file operations
292
try:
293
rename_file('old-name.txt', 'new-name.txt')
294
print("File renamed successfully")
295
except OSError as e:
296
print(f"Failed to rename file: {e}")
297
298
# Check file mode before operations
299
with open('data.txt', 'a') as f:
300
if is_append_mode(f):
301
print("File is in append mode")
302
f.write("Additional data\n")
303
```
304
305
### Custom Logging for Production
306
307
```python
308
import boto3
309
import logging
310
import sys
311
312
# Production-friendly logging setup
313
def setup_aws_logging():
314
"""Set up AWS logging for production use."""
315
316
# Create custom formatter
317
formatter = logging.Formatter(
318
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
319
)
320
321
# Create handler for AWS logs
322
aws_handler = logging.StreamHandler(sys.stdout)
323
aws_handler.setFormatter(formatter)
324
aws_handler.setLevel(logging.WARNING) # Only warnings and errors
325
326
# Configure boto3 logger
327
boto3_logger = logging.getLogger('boto3')
328
boto3_logger.addHandler(aws_handler)
329
boto3_logger.setLevel(logging.WARNING)
330
331
# Prevent duplicate logs
332
boto3_logger.propagate = False
333
334
# Use in production
335
setup_aws_logging()
336
337
# Now only warnings and errors will be logged
338
s3_client = boto3.client('s3')
339
try:
340
s3_client.head_bucket(Bucket='nonexistent-bucket')
341
except Exception as e:
342
# This error will be logged
343
print(f"Error: {e}")
344
```
345
346
### Debugging Network Issues
347
348
```python
349
import boto3
350
import logging
351
from boto3.compat import SOCKET_ERROR
352
353
# Enable comprehensive logging for network debugging
354
boto3.set_stream_logger('botocore', logging.DEBUG)
355
356
s3_client = boto3.client('s3')
357
358
try:
359
response = s3_client.list_buckets()
360
except SOCKET_ERROR as e:
361
print(f"Network error occurred: {e}")
362
except Exception as e:
363
print(f"Other error: {e}")
364
```