The SecretStore service provides secure credential management with automatic secret resolution and multi-mode support.
SecretStore
application_sdk.services.secretstore.SecretStore
Unified secret store service for handling secret management across providers.
Class Methods
get_credentials
Resolve credentials based on credential source with automatic secret substitution.
from application_sdk.services.secretstore import SecretStore
credentials = await SecretStore.get_credentials(
credential_guid: str
)
The unique GUID of the credential configuration to resolve
Complete credential data with secrets resolved
If credential resolution fails due to missing configuration, secret store errors, or invalid credential format
Supports Multi-key mode (direct / has secret-path) and Single-key mode (no secret-path, non-direct).
get_secret
Get secret from the Dapr secret store component.
secret_data = SecretStore.get_secret(
secret_key: str,
component_name: str = SECRET_STORE_NAME
)
Key of the secret to fetch from the secret store
component_name
str
default:"SECRET_STORE_NAME"
Name of the Dapr component to fetch from
Processed secret data as a dictionary
If the secret cannot be retrieved from the component
In local development environment, returns empty dict to avoid secret store dependency.
get_deployment_secret
Get a specific key from deployment configuration in the deployment secret store.
value = SecretStore.get_deployment_secret(key: str)
The key to fetch from the deployment secret
The value for the specified key, or None if the key is not found or the component is unavailable
resolve_credentials
Resolve credentials by substituting secret references with actual values.
resolved = SecretStore.resolve_credentials(
credential_config: Dict[str, Any],
secret_data: Dict[str, Any]
)
Base credential configuration with potential references
Secret data containing actual secret values
Credential configuration with all secret references resolved
apply_secret_values
Apply secret values to source data by substituting references.
result = SecretStore.apply_secret_values(
source_data: Dict[str, Any],
secret_data: Dict[str, Any]
)
Original data with potential references to secrets
Secret data containing actual secret values
Deep copy of source data with secret references resolved
save_secret
Store credentials in the state store (development environment only).
guid = await SecretStore.save_secret(config: Dict[str, Any])
The credential configuration to store
The generated credential GUID that can be used to retrieve the credentials
If called in production environment (non-local deployment)
This method is designed for development and testing purposes only. In production environments, secrets should be managed through proper secret management systems.
Enumerations
CredentialSource
from application_sdk.services.secretstore import CredentialSource
CredentialSource.DIRECT # Direct credential source
CredentialSource.AGENT # Agent-based credential source
SecretMode
from application_sdk.services.secretstore import SecretMode
SecretMode.MULTI_KEY # Multi-key secret retrieval
SecretMode.SINGLE_KEY # Single-key secret retrieval
Example Usage
Basic Credential Resolution
from application_sdk.services.secretstore import SecretStore
# Get credentials with automatic secret resolution
credentials = await SecretStore.get_credentials("cred-abc-123")
# Use resolved credentials
print(f"Connecting to {credentials['host']}:{credentials['port']}")
print(f"Database: {credentials['database']}")
print(f"Username: {credentials['username']}")
# Password is automatically resolved from secret store
Database Credentials
from application_sdk.services.secretstore import SecretStore
from application_sdk.clients.sql import SQLClient
async def connect_to_database(credential_guid: str):
"""Connect to database using credentials from secret store."""
# Get credentials
credentials = await SecretStore.get_credentials(credential_guid)
# Create and load SQL client
client = SQLClient()
await client.load(credentials=credentials)
return client
API Credentials
from application_sdk.services.secretstore import SecretStore
from application_sdk.clients.base import BaseClient
class APIClient(BaseClient):
async def load_from_credential_store(self, credential_guid: str):
"""Load API client using credentials from secret store."""
# Resolve credentials
credentials = await SecretStore.get_credentials(credential_guid)
# Set up headers
self.http_headers = {
"Authorization": f"Bearer {credentials['api_key']}",
"X-API-Secret": credentials['api_secret']
}
Fetching Specific Secrets
from application_sdk.services.secretstore import SecretStore
# Get database password
db_secret = SecretStore.get_secret("database-password")
password = db_secret.get("password")
# Get API keys
api_secrets = SecretStore.get_secret("api-credentials")
api_key = api_secrets.get("api_key")
api_secret = api_secrets.get("api_secret")
# Get from specific component
external_secret = SecretStore.get_secret(
secret_key="external-api-key",
component_name="external-secrets"
)
Deployment Configuration
from application_sdk.services.secretstore import SecretStore
# Get deployment-specific configuration
auth_url = SecretStore.get_deployment_secret("ATLAN_AUTH_CLIENT_ID")
if auth_url:
print(f"Auth URL: {auth_url}")
# Get deployment name
deployment_name = SecretStore.get_deployment_secret("deployment_name")
if deployment_name:
print(f"Deployment: {deployment_name}")
Manual Secret Resolution
from application_sdk.services.secretstore import SecretStore
# Credential configuration with references
config = {
"host": "db.example.com",
"port": 5432,
"database": "analytics",
"username": "app_user",
"password": "db_password_key", # Reference to secret
"ssl_cert": "ssl_cert_key" # Reference to secret
}
# Fetch secrets
secrets = SecretStore.get_secret("database-secrets")
# secrets = {"db_password_key": "actual_password", "ssl_cert_key": "-----BEGIN CERTIFICATE-----..."}
# Resolve references
resolved = SecretStore.resolve_credentials(config, secrets)
# resolved["password"] is now "actual_password"
# resolved["ssl_cert"] is now the actual certificate
Development Environment
from application_sdk.services.secretstore import SecretStore
# Save credentials for testing (development only)
test_credentials = {
"host": "localhost",
"port": 5432,
"username": "dev_user",
"password": "dev_password",
"database": "app_dev"
}
# Store credentials
credential_guid = await SecretStore.save_secret(test_credentials)
print(f"Stored with GUID: {credential_guid}")
# Later retrieve
retrieved = await SecretStore.get_credentials(credential_guid)
Handler Integration
from application_sdk.handlers.base import BaseHandler
from application_sdk.services.secretstore import SecretStore
class SecureHandler(BaseHandler):
async def load(self, credentials: dict):
"""Load handler with credentials from secret store."""
# Get credential GUID from input
credential_guid = credentials.get("credential_guid")
if credential_guid:
# Resolve actual credentials
actual_credentials = await SecretStore.get_credentials(
credential_guid
)
else:
# Use provided credentials directly
actual_credentials = credentials
# Load client with resolved credentials
await self.client.load(credentials=actual_credentials)
Activity Integration
from application_sdk.activities import ActivitiesInterface
from application_sdk.services.secretstore import SecretStore
from temporalio import activity
class SecureActivities(ActivitiesInterface):
@activity.defn
async def load_credentials(self, workflow_args):
"""Load and validate credentials."""
credential_guid = workflow_args["credential_guid"]
# Resolve credentials
try:
credentials = await SecretStore.get_credentials(credential_guid)
# Validate required fields
required = ["host", "username", "password"]
missing = [f for f in required if f not in credentials]
if missing:
return {
"success": False,
"error": f"Missing fields: {missing}"
}
# Store in state
state = await self._get_state(workflow_args)
state.workflow_args["credentials"] = credentials
await self._set_state(workflow_args)
return {"success": True}
except Exception as e:
return {
"success": False,
"error": str(e)
}
Nested Secret References
from application_sdk.services.secretstore import SecretStore
# Configuration with nested references
config = {
"host": "api.example.com",
"port": 443,
"extra": {
"auth_token": "auth_token_ref",
"refresh_token": "refresh_token_ref",
"certificate": "cert_ref"
}
}
# Fetch secrets
secrets = SecretStore.get_secret("api-secrets")
# Resolve (handles nested 'extra' dict automatically)
resolved = SecretStore.apply_secret_values(config, secrets)
# resolved["extra"]["auth_token"] contains actual token
# resolved["extra"]["certificate"] contains actual certificate
Best Practices
Credential Management
- Never store raw passwords in code or configuration
- Always use secret references
- Resolve credentials at runtime
- Use credential GUIDs for reusability
- Validate credentials after resolution
Secret Storage
- Use proper secret management in production
- Only use save_secret() in development
- Store secrets in appropriate secret stores
- Use descriptive secret keys
- Rotate secrets regularly
Error Handling
- Always handle credential resolution errors
- Provide fallback behavior
- Log errors without exposing secrets
- Validate resolved credentials
- Implement retry logic for transient failures
Security
- Never log actual secret values
- Validate secret references before resolution
- Use separate secrets for different environments
- Limit secret access by principle of least privilege
- Audit secret access