0
# Parameters
1
2
Retrieve and cache parameters from AWS Systems Manager Parameter Store, AWS Secrets Manager, and AWS AppConfig with automatic caching, transformation support, and multiple provider backends. Enables efficient configuration management for serverless applications.
3
4
## Capabilities
5
6
### Parameter Store Functions
7
8
High-level functions for retrieving parameters from AWS Systems Manager Parameter Store.
9
10
```python { .api }
11
def get_parameter(
12
name: str,
13
decrypt: bool = True,
14
max_age: int = 5,
15
transform: str = None,
16
force_fetch: bool = False,
17
**sdk_options,
18
) -> str:
19
"""
20
Retrieve a single parameter from Systems Manager Parameter Store.
21
22
Parameters:
23
- name: Parameter name or path
24
- decrypt: Whether to decrypt SecureString parameters
25
- max_age: Cache TTL in seconds (0 to disable caching)
26
- transform: Transformation to apply (json, base64, auto)
27
- force_fetch: Whether to bypass cache and fetch fresh value
28
- **sdk_options: Additional boto3 get_parameter arguments
29
30
Returns:
31
Parameter value as string (or transformed type if transform specified)
32
33
Raises:
34
GetParameterError: If parameter retrieval fails
35
TransformParameterError: If transformation fails
36
"""
37
38
def set_parameter(
39
name: str,
40
value: str,
41
parameter_type: str = "String",
42
overwrite: bool = True,
43
**sdk_options,
44
) -> Dict[str, Any]:
45
"""
46
Set a parameter in Systems Manager Parameter Store.
47
48
Parameters:
49
- name: Parameter name
50
- value: Parameter value
51
- parameter_type: Parameter type (String, StringList, SecureString)
52
- overwrite: Whether to overwrite existing parameter
53
- **sdk_options: Additional boto3 put_parameter arguments
54
55
Returns:
56
Parameter version and tier information
57
"""
58
59
def get_parameters(
60
path: str,
61
recursive: bool = True,
62
decrypt: bool = True,
63
max_age: int = 5,
64
transform: str = None,
65
force_fetch: bool = False,
66
**sdk_options,
67
) -> Dict[str, Any]:
68
"""
69
Retrieve multiple parameters by path from Parameter Store.
70
71
Parameters:
72
- path: Parameter path prefix
73
- recursive: Whether to retrieve parameters recursively
74
- decrypt: Whether to decrypt SecureString parameters
75
- max_age: Cache TTL in seconds
76
- transform: Transformation to apply to all parameters
77
- force_fetch: Whether to bypass cache
78
- **sdk_options: Additional boto3 get_parameters_by_path arguments
79
80
Returns:
81
Dictionary mapping parameter names to values
82
"""
83
84
def get_parameters_by_name(
85
parameters: List[str],
86
decrypt: bool = True,
87
max_age: int = 5,
88
transform: str = None,
89
force_fetch: bool = False,
90
**sdk_options,
91
) -> Dict[str, Any]:
92
"""
93
Retrieve multiple parameters by name from Parameter Store.
94
95
Parameters:
96
- parameters: List of parameter names
97
- decrypt: Whether to decrypt SecureString parameters
98
- max_age: Cache TTL in seconds
99
- transform: Transformation to apply
100
- force_fetch: Whether to bypass cache
101
- **sdk_options: Additional boto3 get_parameters arguments
102
103
Returns:
104
Dictionary mapping parameter names to values
105
"""
106
```
107
108
### Secrets Manager Functions
109
110
High-level functions for retrieving secrets from AWS Secrets Manager.
111
112
```python { .api }
113
def get_secret(
114
name: str,
115
version_id: str = None,
116
version_stage: str = None,
117
max_age: int = 5,
118
transform: str = None,
119
force_fetch: bool = False,
120
**sdk_options,
121
) -> str:
122
"""
123
Retrieve a secret from AWS Secrets Manager.
124
125
Parameters:
126
- name: Secret name or ARN
127
- version_id: Specific version ID to retrieve
128
- version_stage: Version stage to retrieve (AWSCURRENT, AWSPENDING)
129
- max_age: Cache TTL in seconds
130
- transform: Transformation to apply (json, base64, auto)
131
- force_fetch: Whether to bypass cache
132
- **sdk_options: Additional boto3 get_secret_value arguments
133
134
Returns:
135
Secret value as string (or transformed type)
136
137
Raises:
138
GetParameterError: If secret retrieval fails
139
TransformParameterError: If transformation fails
140
"""
141
142
def get_secrets_by_name(
143
secrets: List[str],
144
max_age: int = 5,
145
transform: str = None,
146
force_fetch: bool = False,
147
**sdk_options,
148
) -> Dict[str, Any]:
149
"""
150
Retrieve multiple secrets by name from Secrets Manager.
151
152
Parameters:
153
- secrets: List of secret names or ARNs
154
- max_age: Cache TTL in seconds
155
- transform: Transformation to apply
156
- force_fetch: Whether to bypass cache
157
- **sdk_options: Additional boto3 arguments
158
159
Returns:
160
Dictionary mapping secret names to values
161
"""
162
163
def set_secret(
164
name: str,
165
secret: str,
166
version_stage: str = None,
167
**sdk_options,
168
) -> Dict[str, Any]:
169
"""
170
Create or update a secret in AWS Secrets Manager.
171
172
Parameters:
173
- name: Secret name or ARN
174
- secret: Secret value (string or JSON)
175
- version_stage: Version stage to set
176
- **sdk_options: Additional boto3 arguments
177
178
Returns:
179
Secret creation/update response
180
"""
181
```
182
183
### AppConfig Functions
184
185
High-level functions for retrieving configuration from AWS AppConfig.
186
187
```python { .api }
188
def get_app_config(
189
name: str,
190
environment: str,
191
application: str,
192
max_age: int = 5,
193
transform: str = None,
194
force_fetch: bool = False,
195
**sdk_options,
196
) -> Union[bytes, str]:
197
"""
198
Retrieve configuration from AWS AppConfig.
199
200
Parameters:
201
- name: Configuration profile name
202
- environment: AppConfig environment
203
- application: AppConfig application name
204
- max_age: Cache TTL in seconds
205
- transform: Transformation to apply (json, base64, auto)
206
- force_fetch: Whether to bypass cache
207
- **sdk_options: Additional AppConfig arguments
208
209
Returns:
210
Configuration data as bytes or string (or transformed type)
211
212
Raises:
213
GetParameterError: If configuration retrieval fails
214
TransformParameterError: If transformation fails
215
"""
216
```
217
218
### Cache Management
219
220
Functions for managing parameter caches across providers.
221
222
```python { .api }
223
def clear_caches() -> None:
224
"""
225
Clear all parameter caches across all providers.
226
227
Use this to force fresh retrieval of all cached parameters.
228
"""
229
```
230
231
### Provider Classes
232
233
Low-level provider classes for advanced parameter management scenarios.
234
235
```python { .api }
236
class BaseProvider:
237
"""Base provider interface for parameter retrieval"""
238
239
def __init__(
240
self,
241
config: Dict[str, Any] = None,
242
):
243
"""
244
Initialize base provider.
245
246
Parameters:
247
- config: Provider-specific configuration
248
"""
249
250
def get(
251
self,
252
name: str,
253
max_age: int = None,
254
transform: str = None,
255
force_fetch: bool = False,
256
**kwargs,
257
) -> Any:
258
"""
259
Retrieve single parameter.
260
261
Parameters:
262
- name: Parameter identifier
263
- max_age: Cache TTL in seconds
264
- transform: Transformation to apply
265
- force_fetch: Whether to bypass cache
266
- **kwargs: Provider-specific arguments
267
268
Returns:
269
Parameter value
270
"""
271
272
def get_multiple(
273
self,
274
path: str = None,
275
names: List[str] = None,
276
max_age: int = None,
277
transform: str = None,
278
force_fetch: bool = False,
279
**kwargs,
280
) -> Dict[str, Any]:
281
"""
282
Retrieve multiple parameters.
283
284
Parameters:
285
- path: Parameter path prefix (for hierarchical providers)
286
- names: List of parameter names
287
- max_age: Cache TTL in seconds
288
- transform: Transformation to apply
289
- force_fetch: Whether to bypass cache
290
- **kwargs: Provider-specific arguments
291
292
Returns:
293
Dictionary of parameter name to value mappings
294
"""
295
296
def set(
297
self,
298
name: str,
299
value: Any,
300
**kwargs,
301
) -> Any:
302
"""
303
Set parameter value.
304
305
Parameters:
306
- name: Parameter identifier
307
- value: Parameter value
308
- **kwargs: Provider-specific arguments
309
310
Returns:
311
Set operation response
312
"""
313
314
class SSMProvider(BaseProvider):
315
"""Systems Manager Parameter Store provider"""
316
317
def __init__(
318
self,
319
config: Dict[str, Any] = None,
320
):
321
"""
322
Initialize SSM provider.
323
324
Parameters:
325
- config: SSM-specific configuration including boto3 session options
326
"""
327
328
def get(
329
self,
330
name: str,
331
decrypt: bool = True,
332
max_age: int = 5,
333
transform: str = None,
334
force_fetch: bool = False,
335
**sdk_options,
336
) -> Any:
337
"""Get single parameter from SSM Parameter Store"""
338
339
def get_multiple(
340
self,
341
path: str = None,
342
names: List[str] = None,
343
recursive: bool = True,
344
decrypt: bool = True,
345
max_age: int = 5,
346
transform: str = None,
347
force_fetch: bool = False,
348
**sdk_options,
349
) -> Dict[str, Any]:
350
"""Get multiple parameters from SSM Parameter Store"""
351
352
def set(
353
self,
354
name: str,
355
value: str,
356
parameter_type: str = "String",
357
overwrite: bool = True,
358
**sdk_options,
359
) -> Dict[str, Any]:
360
"""Set parameter in SSM Parameter Store"""
361
362
class SecretsProvider(BaseProvider):
363
"""AWS Secrets Manager provider"""
364
365
def __init__(
366
self,
367
config: Dict[str, Any] = None,
368
):
369
"""
370
Initialize Secrets Manager provider.
371
372
Parameters:
373
- config: Secrets Manager-specific configuration
374
"""
375
376
def get(
377
self,
378
name: str,
379
version_id: str = None,
380
version_stage: str = None,
381
max_age: int = 5,
382
transform: str = None,
383
force_fetch: bool = False,
384
**sdk_options,
385
) -> Any:
386
"""Get secret from Secrets Manager"""
387
388
def get_multiple(
389
self,
390
names: List[str],
391
max_age: int = 5,
392
transform: str = None,
393
force_fetch: bool = False,
394
**sdk_options,
395
) -> Dict[str, Any]:
396
"""Get multiple secrets from Secrets Manager"""
397
398
def set(
399
self,
400
name: str,
401
secret: str,
402
version_stage: str = None,
403
**sdk_options,
404
) -> Dict[str, Any]:
405
"""Create or update secret in Secrets Manager"""
406
407
class AppConfigProvider(BaseProvider):
408
"""AWS AppConfig provider"""
409
410
def __init__(
411
self,
412
environment: str,
413
application: str,
414
config: Dict[str, Any] = None,
415
):
416
"""
417
Initialize AppConfig provider.
418
419
Parameters:
420
- environment: AppConfig environment name
421
- application: AppConfig application name
422
- config: AppConfig-specific configuration
423
"""
424
425
def get(
426
self,
427
name: str,
428
max_age: int = 5,
429
transform: str = None,
430
force_fetch: bool = False,
431
**sdk_options,
432
) -> Union[bytes, str]:
433
"""Get configuration from AppConfig"""
434
435
class DynamoDBProvider(BaseProvider):
436
"""DynamoDB provider for parameter storage"""
437
438
def __init__(
439
self,
440
table_name: str,
441
key_attr: str = "id",
442
value_attr: str = "value",
443
config: Dict[str, Any] = None,
444
):
445
"""
446
Initialize DynamoDB provider.
447
448
Parameters:
449
- table_name: DynamoDB table name
450
- key_attr: Attribute name for parameter key
451
- value_attr: Attribute name for parameter value
452
- config: DynamoDB-specific configuration
453
"""
454
455
def get(
456
self,
457
name: str,
458
max_age: int = 5,
459
transform: str = None,
460
force_fetch: bool = False,
461
**sdk_options,
462
) -> Any:
463
"""Get parameter from DynamoDB table"""
464
465
def get_multiple(
466
self,
467
names: List[str],
468
max_age: int = 5,
469
transform: str = None,
470
force_fetch: bool = False,
471
**sdk_options,
472
) -> Dict[str, Any]:
473
"""Get multiple parameters from DynamoDB table"""
474
475
def set(
476
self,
477
name: str,
478
value: Any,
479
**sdk_options,
480
) -> Dict[str, Any]:
481
"""Set parameter in DynamoDB table"""
482
```
483
484
## Usage Examples
485
486
### Basic Parameter Retrieval
487
488
```python
489
from aws_lambda_powertools.utilities.parameters import (
490
get_parameter,
491
get_secret,
492
get_app_config
493
)
494
from aws_lambda_powertools.utilities.typing import LambdaContext
495
import json
496
497
def lambda_handler(event: dict, context: LambdaContext) -> dict:
498
# Get database URL from Parameter Store
499
db_url = get_parameter("/myapp/database/url", decrypt=True)
500
501
# Get API key from Secrets Manager with JSON transformation
502
api_credentials = get_secret("myapp/api-credentials", transform="json")
503
api_key = api_credentials["api_key"]
504
505
# Get feature flags from AppConfig
506
feature_config = get_app_config(
507
name="feature-flags",
508
environment="production",
509
application="myapp",
510
transform="json"
511
)
512
513
# Use retrieved configuration
514
if feature_config.get("new_feature_enabled", False):
515
result = new_feature_logic(db_url, api_key)
516
else:
517
result = legacy_logic(db_url, api_key)
518
519
return {
520
"statusCode": 200,
521
"body": json.dumps(result)
522
}
523
524
def new_feature_logic(db_url: str, api_key: str) -> dict:
525
# New feature implementation
526
return {"message": "New feature executed", "version": "2.0"}
527
528
def legacy_logic(db_url: str, api_key: str) -> dict:
529
# Legacy implementation
530
return {"message": "Legacy feature executed", "version": "1.0"}
531
```
532
533
### Bulk Parameter Retrieval
534
535
```python
536
from aws_lambda_powertools.utilities.parameters import (
537
get_parameters,
538
get_parameters_by_name,
539
get_secrets_by_name
540
)
541
from aws_lambda_powertools.utilities.typing import LambdaContext
542
543
def lambda_handler(event: dict, context: LambdaContext) -> dict:
544
# Get all parameters under a path
545
app_config = get_parameters(
546
path="/myapp/config/",
547
recursive=True,
548
decrypt=True,
549
transform="auto" # Auto-detect JSON/base64
550
)
551
552
# Get specific parameters by name
553
database_params = get_parameters_by_name(
554
parameters=[
555
"/myapp/database/host",
556
"/myapp/database/port",
557
"/myapp/database/name"
558
],
559
decrypt=True
560
)
561
562
# Get multiple secrets
563
secrets = get_secrets_by_name(
564
secrets=[
565
"myapp/database-credentials",
566
"myapp/external-api-keys"
567
],
568
transform="json"
569
)
570
571
# Build database connection string
572
db_host = database_params["/myapp/database/host"]
573
db_port = database_params["/myapp/database/port"]
574
db_name = database_params["/myapp/database/name"]
575
db_creds = secrets["myapp/database-credentials"]
576
577
connection_string = f"postgresql://{db_creds['username']}:{db_creds['password']}@{db_host}:{db_port}/{db_name}"
578
579
return {
580
"statusCode": 200,
581
"body": "Configuration loaded successfully"
582
}
583
```
584
585
### Advanced Provider Usage
586
587
```python
588
from aws_lambda_powertools.utilities.parameters import (
589
SSMProvider,
590
SecretsProvider,
591
AppConfigProvider,
592
clear_caches
593
)
594
from aws_lambda_powertools.utilities.typing import LambdaContext
595
import boto3
596
597
# Initialize custom providers with specific configurations
598
ssm_provider = SSMProvider(config={
599
"boto3_session": boto3.Session(region_name="us-west-2")
600
})
601
602
secrets_provider = SecretsProvider(config={
603
"boto3_session": boto3.Session(region_name="us-west-2")
604
})
605
606
appconfig_provider = AppConfigProvider(
607
environment="production",
608
application="myapp"
609
)
610
611
def lambda_handler(event: dict, context: LambdaContext) -> dict:
612
# Use providers directly for more control
613
614
# Get encrypted parameter with custom caching
615
sensitive_config = ssm_provider.get(
616
name="/myapp/sensitive-config",
617
decrypt=True,
618
max_age=300, # Cache for 5 minutes
619
transform="json"
620
)
621
622
# Get secret with specific version
623
api_key = secrets_provider.get(
624
name="myapp/api-key",
625
version_stage="AWSCURRENT",
626
max_age=3600 # Cache for 1 hour
627
)
628
629
# Get AppConfig with custom polling
630
feature_flags = appconfig_provider.get(
631
name="feature-flags",
632
max_age=60, # Poll every minute
633
transform="json"
634
)
635
636
# Process based on feature flags
637
results = []
638
639
if feature_flags.get("enable_advanced_processing", False):
640
results.append(advanced_processing(sensitive_config, api_key))
641
642
if feature_flags.get("enable_reporting", False):
643
results.append(generate_report(sensitive_config))
644
645
# Clear caches if needed (e.g., for testing)
646
if event.get("clear_cache", False):
647
clear_caches()
648
649
return {
650
"statusCode": 200,
651
"results": results
652
}
653
654
def advanced_processing(config: dict, api_key: str) -> dict:
655
"""Advanced processing using configuration"""
656
return {
657
"type": "advanced",
658
"processed_items": config.get("batch_size", 100),
659
"api_version": "v2"
660
}
661
662
def generate_report(config: dict) -> dict:
663
"""Generate report based on configuration"""
664
return {
665
"type": "report",
666
"format": config.get("report_format", "json"),
667
"timestamp": "2024-01-01T00:00:00Z"
668
}
669
```
670
671
### Configuration Management Pattern
672
673
```python
674
from aws_lambda_powertools.utilities.parameters import (
675
get_parameters,
676
get_secret,
677
get_app_config
678
)
679
from aws_lambda_powertools.utilities.typing import LambdaContext
680
from typing import Dict, Any
681
import os
682
683
class ConfigManager:
684
"""Centralized configuration management"""
685
686
def __init__(self):
687
self._config_cache = {}
688
self.app_name = os.environ.get("APP_NAME", "myapp")
689
self.environment = os.environ.get("ENVIRONMENT", "dev")
690
691
def get_database_config(self) -> Dict[str, Any]:
692
"""Get database configuration"""
693
if "database" not in self._config_cache:
694
# Get database parameters
695
db_params = get_parameters(
696
path=f"/{self.app_name}/{self.environment}/database/",
697
recursive=True,
698
decrypt=True
699
)
700
701
# Get database credentials
702
db_secret = get_secret(
703
name=f"{self.app_name}/{self.environment}/database-credentials",
704
transform="json"
705
)
706
707
self._config_cache["database"] = {
708
**db_params,
709
"credentials": db_secret
710
}
711
712
return self._config_cache["database"]
713
714
def get_api_config(self) -> Dict[str, Any]:
715
"""Get external API configuration"""
716
if "api" not in self._config_cache:
717
# Get API settings from Parameter Store
718
api_params = get_parameters(
719
path=f"/{self.app_name}/{self.environment}/api/",
720
recursive=True
721
)
722
723
# Get API keys from Secrets Manager
724
api_secrets = get_secret(
725
name=f"{self.app_name}/{self.environment}/api-keys",
726
transform="json"
727
)
728
729
self._config_cache["api"] = {
730
**api_params,
731
"keys": api_secrets
732
}
733
734
return self._config_cache["api"]
735
736
def get_feature_flags(self) -> Dict[str, Any]:
737
"""Get feature flags from AppConfig"""
738
if "features" not in self._config_cache:
739
features = get_app_config(
740
name="feature-flags",
741
environment=self.environment,
742
application=self.app_name,
743
transform="json",
744
max_age=30 # Refresh every 30 seconds
745
)
746
747
self._config_cache["features"] = features
748
749
return self._config_cache["features"]
750
751
def refresh_cache(self) -> None:
752
"""Force refresh of all cached configuration"""
753
self._config_cache.clear()
754
755
# Global configuration manager instance
756
config_manager = ConfigManager()
757
758
def lambda_handler(event: dict, context: LambdaContext) -> dict:
759
# Get configuration through manager
760
db_config = config_manager.get_database_config()
761
api_config = config_manager.get_api_config()
762
features = config_manager.get_feature_flags()
763
764
# Use configuration
765
database_url = build_database_url(db_config)
766
767
results = []
768
769
if features.get("enable_data_processing", False):
770
results.append(process_data(database_url))
771
772
if features.get("enable_external_api", False):
773
external_data = call_external_api(api_config)
774
results.append(external_data)
775
776
# Refresh config cache if requested
777
if event.get("refresh_config", False):
778
config_manager.refresh_cache()
779
780
return {
781
"statusCode": 200,
782
"processed": len(results),
783
"results": results
784
}
785
786
def build_database_url(config: Dict[str, Any]) -> str:
787
"""Build database connection URL from configuration"""
788
host = config.get(f"/{config_manager.app_name}/{config_manager.environment}/database/host")
789
port = config.get(f"/{config_manager.app_name}/{config_manager.environment}/database/port")
790
database = config.get(f"/{config_manager.app_name}/{config_manager.environment}/database/name")
791
792
credentials = config["credentials"]
793
username = credentials["username"]
794
password = credentials["password"]
795
796
return f"postgresql://{username}:{password}@{host}:{port}/{database}"
797
798
def call_external_api(config: Dict[str, Any]) -> Dict[str, Any]:
799
"""Call external API using configuration"""
800
import requests
801
802
base_url = config.get(f"/{config_manager.app_name}/{config_manager.environment}/api/base_url")
803
timeout = int(config.get(f"/{config_manager.app_name}/{config_manager.environment}/api/timeout", "30"))
804
805
api_key = config["keys"]["primary_api_key"]
806
807
response = requests.get(
808
f"{base_url}/data",
809
headers={"Authorization": f"Bearer {api_key}"},
810
timeout=timeout
811
)
812
813
return response.json()
814
```
815
816
### Error Handling and Retries
817
818
```python
819
from aws_lambda_powertools.utilities.parameters import (
820
get_parameter,
821
get_secret,
822
GetParameterError,
823
TransformParameterError
824
)
825
from aws_lambda_powertools.utilities.typing import LambdaContext
826
import time
827
import json
828
829
def lambda_handler(event: dict, context: LambdaContext) -> dict:
830
try:
831
# Get configuration with retry logic
832
config = get_configuration_with_retry()
833
834
# Process event using configuration
835
result = process_event(event, config)
836
837
return {
838
"statusCode": 200,
839
"body": json.dumps(result)
840
}
841
842
except GetParameterError as e:
843
print(f"Failed to retrieve parameter: {str(e)}")
844
return {
845
"statusCode": 500,
846
"body": json.dumps({"error": "Configuration unavailable"})
847
}
848
849
except TransformParameterError as e:
850
print(f"Failed to transform parameter: {str(e)}")
851
return {
852
"statusCode": 500,
853
"body": json.dumps({"error": "Configuration format invalid"})
854
}
855
856
def get_configuration_with_retry(max_retries: int = 3, retry_delay: float = 1.0) -> dict:
857
"""Get configuration with exponential backoff retry"""
858
859
for attempt in range(max_retries):
860
try:
861
# Attempt to get all required configuration
862
config = {}
863
864
# Get database configuration
865
config["database_url"] = get_parameter(
866
"/myapp/database/url",
867
decrypt=True,
868
force_fetch=True # Always get fresh value on retry
869
)
870
871
# Get API configuration
872
api_config = get_secret(
873
"myapp/api-config",
874
transform="json",
875
force_fetch=True
876
)
877
config["api"] = api_config
878
879
# Get feature flags
880
feature_flags = get_parameter(
881
"/myapp/feature-flags",
882
transform="json",
883
force_fetch=True
884
)
885
config["features"] = feature_flags
886
887
return config
888
889
except (GetParameterError, TransformParameterError) as e:
890
if attempt == max_retries - 1:
891
# Last attempt failed, re-raise
892
raise
893
894
# Wait before retrying with exponential backoff
895
wait_time = retry_delay * (2 ** attempt)
896
print(f"Configuration retrieval failed (attempt {attempt + 1}/{max_retries}): {str(e)}")
897
print(f"Retrying in {wait_time} seconds...")
898
time.sleep(wait_time)
899
900
def process_event(event: dict, config: dict) -> dict:
901
"""Process event using retrieved configuration"""
902
903
# Use database connection
904
database_url = config["database_url"]
905
906
# Use API configuration
907
api_endpoint = config["api"]["endpoint"]
908
api_key = config["api"]["key"]
909
910
# Check feature flags
911
features = config["features"]
912
913
result = {
914
"processed_at": time.time(),
915
"event_id": event.get("id", "unknown"),
916
"features_enabled": []
917
}
918
919
if features.get("enable_processing", False):
920
result["features_enabled"].append("processing")
921
# Process event data
922
result["processed_records"] = len(event.get("records", []))
923
924
if features.get("enable_notifications", False):
925
result["features_enabled"].append("notifications")
926
# Send notifications
927
result["notifications_sent"] = 1
928
929
return result
930
```
931
932
## Types
933
934
```python { .api }
935
from typing import Dict, Any, List, Union, Optional, Callable
936
import boto3
937
938
# Parameter transformation types
939
TransformType = Literal["json", "base64", "auto"]
940
941
# Parameter provider configuration
942
ProviderConfig = Dict[str, Any]
943
944
# Parameter retrieval options
945
ParameterOptions = Dict[str, Any]
946
947
# Exception types
948
class GetParameterError(Exception):
949
"""Raised when parameter retrieval fails"""
950
def __init__(self, message: str, parameter_name: str = None): ...
951
952
class TransformParameterError(Exception):
953
"""Raised when parameter transformation fails"""
954
def __init__(self, message: str, parameter_name: str = None, transform: str = None): ...
955
956
# Provider response types
957
ParameterValue = Union[str, int, float, bool, Dict[str, Any], List[Any], bytes]
958
ParameterDict = Dict[str, ParameterValue]
959
960
# Boto3 session type for provider configuration
961
Boto3Session = boto3.Session
962
963
# Cache configuration
964
class CacheConfig:
965
def __init__(
966
self,
967
max_age: int = 5,
968
max_size: int = 1000,
969
enabled: bool = True,
970
):
971
"""
972
Cache configuration for parameter providers.
973
974
Parameters:
975
- max_age: Default cache TTL in seconds
976
- max_size: Maximum number of cached items
977
- enabled: Whether caching is enabled
978
"""
979
980
# SDK options for AWS service calls
981
SSMOptions = Dict[str, Any]
982
SecretsManagerOptions = Dict[str, Any]
983
AppConfigOptions = Dict[str, Any]
984
DynamoDBOptions = Dict[str, Any]
985
```