0
# Secrets Management
1
2
Integrate with AWS Secrets Manager for secure credential and configuration management within Dagster pipelines.
3
4
## Capabilities
5
6
### Secrets Manager Resource
7
8
```python { .api }
9
class SecretsManagerResource(ResourceWithBoto3Configuration):
10
"""
11
Resource for accessing AWS Secrets Manager.
12
"""
13
14
def get_secret(self, secret_id: str) -> Dict:
15
"""
16
Retrieve secret value from Secrets Manager.
17
18
Parameters:
19
secret_id: Secret ARN or name
20
21
Returns:
22
Dict: Secret value and metadata
23
"""
24
25
def get_secret_string(self, secret_id: str) -> str:
26
"""
27
Get secret as string value.
28
"""
29
30
class SecretsManagerSecretsResource(ConfigurableResource):
31
"""
32
Resource providing secrets as configuration.
33
"""
34
secrets_manager: SecretsManagerResource
35
secret_arns: List[str]
36
37
def secretsmanager_resource(**kwargs) -> SecretsManagerResource: ...
38
def secretsmanager_secrets_resource(**kwargs): ...
39
```
40
41
### Secrets Utilities
42
43
```python { .api }
44
def get_secrets_from_arns(arns: List[str]) -> Dict[str, Any]:
45
"""
46
Retrieve multiple secrets by ARN.
47
48
Parameters:
49
arns: List of secret ARNs
50
51
Returns:
52
Dict: Mapping of ARN to secret value
53
"""
54
55
def get_tagged_secrets(tags: Dict[str, str]) -> Dict[str, Any]:
56
"""
57
Retrieve secrets by tags.
58
59
Parameters:
60
tags: Tag filters
61
62
Returns:
63
Dict: Secrets matching tags
64
"""
65
```
66
67
## Usage Examples
68
69
```python
70
from dagster import op, job, Definitions
71
from dagster_aws.secretsmanager import SecretsManagerResource
72
73
@op(required_resource_keys={"secrets"})
74
def connect_to_database(context):
75
secrets = context.resources.secrets
76
77
# Retrieve database credentials
78
db_secret = secrets.get_secret("prod/database/credentials")
79
credentials = json.loads(db_secret['SecretString'])
80
81
# Use credentials to connect
82
connection = connect_db(
83
host=credentials['host'],
84
username=credentials['username'],
85
password=credentials['password']
86
)
87
88
return connection
89
90
@job(
91
resource_defs={
92
"secrets": SecretsManagerResource(region_name="us-west-2")
93
}
94
)
95
def secure_job():
96
connect_to_database()
97
98
defs = Definitions(jobs=[secure_job])
99
```