or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

Files

docs

batch-processing.mdcore-observability.mddata-classes.mdevent-handlers.mdfeature-flags.mdindex.mdparameters.mdparser.mdutilities.md

parameters.mddocs/

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

```