Skip to main content
The SecretStore service provides secure secret management with automatic credential resolution and multi-provider support through Dapr secret stores. It handles complex credential resolution logic including direct credentials, agent-managed credentials, and multi-key/single-key secret retrieval modes.

​
Overview

SecretStore supports:
  • Automatic credential resolution: Fetch credentials with automatic secret substitution
  • Multi-mode secret retrieval: Multi-key bundles and single-key secrets
  • Deployment secrets: Access deployment configuration securely
  • Development support: Local credential storage for testing

​
Class Reference

​
SecretStore

​
SecretStore
class
Unified secret store service for handling secret management across providers.All methods are class methods and can be called directly without instantiation.

​
Core Methods

​
get_credentials

Resolve credentials based on credential source with automatic secret substitution. 
await SecretStore.get_credentials(credential_guid: str) -> Dict[str, Any]
This method retrieves credential configuration from the state store and resolves any secret references by fetching actual values from the secret store. It supports both Multi-key mode (direct credentials or credentials with secret-path) and Single-key mode (credentials without secret-path from agent).
​
credential_guid
str
required
The unique GUID of the credential configuration to resolve.
​
return
Dict[str, Any]
Complete credential data with secrets resolved.
Raises:
  • CommonError: If credential resolution fails due to missing configuration, secret store errors, or invalid credential format
Example: 
from application_sdk.services import SecretStore
from application_sdk.common.error_codes import CommonError

# Resolve database credentials
try:
    creds = await SecretStore.get_credentials("db-cred-abc123")
    print(f"Connecting to {creds['host']}:{creds['port']}")
    # Password is automatically resolved from secret store
    db_connection = connect(
        host=creds['host'],
        port=creds['port'],
        username=creds['username'],
        password=creds['password']
    )
except CommonError as e:
    logger.error(f"Failed to resolve credentials: {e}")

​
get_secret

Get secret from the Dapr secret store component. 
SecretStore.get_secret(
    secret_key: str,
    component_name: str = SECRET_STORE_NAME
) -> Dict[str, Any]
Retrieves secret data from the specified Dapr component and processes it into a standardized dictionary format.
​
secret_key
str
required
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.
​
return
Dict[str, Any]
Processed secret data as a dictionary.
Raises:
  • Exception: If the secret cannot be retrieved from the component
Note: In local development environment, returns empty dict to avoid secret store dependency. Example: 
from application_sdk.services import SecretStore

# Get database credentials
db_secret = SecretStore.get_secret("database-credentials")
print(f"Host: {db_secret.get('host')}")

# Get from specific component
api_secret = SecretStore.get_secret(
    "api-keys",
    component_name="external-secrets"
)

​
get_deployment_secret

Get a specific key from deployment configuration in the deployment secret store. 
SecretStore.get_deployment_secret(key: str) -> Any
Validates that the deployment secret store component is registered before attempting to fetch secrets. This method fetches only the specified key from the deployment secret, rather than the entire secret dictionary.
​
key
str
required
The key to fetch from the deployment secret.
​
return
Any
The value for the specified key, or None if the key is not found or the component is unavailable.
Example: 
from application_sdk.services import SecretStore

# Get a specific deployment configuration value
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}")

​
Secret Resolution Methods

​
resolve_credentials

Resolve credentials by substituting secret references with actual values. 
SecretStore.resolve_credentials(
    credential_config: Dict[str, Any],
    secret_data: Dict[str, Any]
) -> Dict[str, Any]
This method processes credential configuration and replaces any reference values with corresponding secrets from the secret data. It handles both top-level fields and nested extra fields.
​
credential_config
Dict[str, Any]
required
Base credential configuration with potential references.
​
secret_data
Dict[str, Any]
required
Secret data containing actual secret values.
​
return
Dict[str, Any]
Credential configuration with all secret references resolved.
Example: 
from application_sdk.services import SecretStore

# Basic secret resolution
config = {"host": "db.example.com", "password": "db_password_key"}
secrets = {"db_password_key": "actual_secret_password"}
resolved = SecretStore.resolve_credentials(config, secrets)
print(resolved)  # {"host": "db.example.com", "password": "actual_secret_password"}

# Resolution with nested 'extra' fields
config = {
    "host": "db.example.com",
    "extra": {"ssl_cert": "cert_key"}
}
secrets = {"cert_key": "-----BEGIN CERTIFICATE-----..."}
resolved = SecretStore.resolve_credentials(config, secrets)

​
apply_secret_values

Apply secret values to source data by substituting references. 
SecretStore.apply_secret_values(
    source_data: Dict[str, Any],
    secret_data: Dict[str, Any]
) -> Dict[str, Any]
This function replaces values in the source data with actual secret values when the source value exists as a key in the secret data. It supports nested structures and preserves the original data structure.
​
source_data
Dict[str, Any]
required
Original data with potential references to secrets.
​
secret_data
Dict[str, Any]
required
Secret data containing actual secret values.
​
return
Dict[str, Any]
Deep copy of source data with secret references resolved.
Example: 
from application_sdk.services import SecretStore

# Simple secret substitution
source = {
    "database_url": "postgresql://user:${db_password}@localhost/app",
    "api_key": "api_key_ref"
}
secrets = {
    "api_key_ref": "sk-1234567890abcdef",
    "db_password": "secure_db_password"
}
resolved = SecretStore.apply_secret_values(source, secrets)

# With nested extra fields
source = {
    "host": "api.example.com",
    "extra": {"token": "auth_token_ref"}
}
secrets = {"auth_token_ref": "bearer_token_123"}
resolved = SecretStore.apply_secret_values(source, secrets)

​
Development Methods

​
save_secret

Store credentials in the state store (development environment only). 
await SecretStore.save_secret(config: Dict[str, Any]) -> str
This method is designed for development and testing purposes only. In production environments, secrets should be managed through proper secret management systems.
​
config
Dict[str, Any]
required
The credential configuration to store.
​
return
str
The generated credential GUID that can be used to retrieve the credentials.
Raises:
  • ValueError: If called in production environment (non-local deployment)
  • Exception: If there’s an error storing the credentials
Example: 
from application_sdk.services import SecretStore

# Development environment only
config = {
    "host": "localhost",
    "port": 5432,
    "username": "dev_user",
    "password": "dev_password",
    "database": "app_dev"
}
guid = await SecretStore.save_secret(config)
print(f"Stored credentials with GUID: {guid}")

# Later retrieve these credentials
retrieved = await SecretStore.get_credentials(guid)

​
Enums

​
CredentialSource

Enumeration of credential source types.
​
CredentialSource.DIRECT
enum
Direct credentials stored in the state store.
​
CredentialSource.AGENT
enum
Agent-managed credentials with secret references.

​
SecretMode

Enumeration of secret retrieval modes.
​
SecretMode.MULTI_KEY
enum
Multi-key mode: Fetch entire secret bundle at once.
​
SecretMode.SINGLE_KEY
enum
Single-key mode: Fetch secrets individually per field.

​
Usage Patterns

​
Database Connection with Credentials

Retrieve and use database credentials: 
from application_sdk.services import SecretStore
import psycopg2

@activity.defn
async def connect_to_database(credential_guid: str):
    # Get resolved credentials
    creds = await SecretStore.get_credentials(credential_guid)
    
    # Create database connection
    connection = psycopg2.connect(
        host=creds['host'],
        port=creds['port'],
        database=creds['database'],
        user=creds['username'],
        password=creds['password'],
        **creds.get('extra', {})
    )
    
    return connection

​
API Client with Secret Key

Use secret store for API authentication: 
from application_sdk.services import SecretStore
import httpx

@activity.defn
async def call_external_api(endpoint: str):
    # Get API credentials
    api_secret = SecretStore.get_secret("external-api-key")
    api_key = api_secret.get('api_key')
    
    # Make authenticated request
    async with httpx.AsyncClient() as client:
        response = await client.get(
            endpoint,
            headers={"Authorization": f"Bearer {api_key}"}
        )
        return response.json()

​
Multi-Environment Configuration

Access deployment-specific configuration: 
from application_sdk.services import SecretStore

@activity.defn
async def initialize_service():
    # Get deployment configuration
    atlan_base_url = SecretStore.get_deployment_secret("ATLAN_BASE_URL")
    tenant_id = SecretStore.get_deployment_secret("APP_TENANT_ID")
    
    # Initialize service with deployment-specific settings
    service = ExternalService(
        base_url=atlan_base_url,
        tenant_id=tenant_id
    )
    
    return service

​
Credential Configuration in Workflow

Store and retrieve credentials across workflow activities: 
from application_sdk.services import SecretStore
import uuid

@workflow.defn
class DataIngestionWorkflow:
    @workflow.run
    async def run(self, source_config: dict):
        # Store credential configuration (development only)
        if is_development():
            cred_guid = await SecretStore.save_secret(source_config)
        else:
            # In production, credential_guid is provided
            cred_guid = source_config['credential_guid']
        
        # Pass credential GUID to activities
        data = await workflow.execute_activity(
            extract_data,
            cred_guid,
            start_to_close_timeout=timedelta(minutes=30)
        )
        
        return data

@activity.defn
async def extract_data(credential_guid: str) -> list:
    # Resolve credentials in activity
    creds = await SecretStore.get_credentials(credential_guid)
    
    # Use credentials to connect and extract data
    data = fetch_from_source(creds)
    return data

​
Credential Resolution Logic

The get_credentials method follows this logic:
1

Fetch Configuration

Retrieve credential configuration from StateStore using the credential GUID.
2

Determine Mode

  • Multi-key mode: If credentialSource == 'direct' OR secret-path is defined
  • Single-key mode: Otherwise (agent-managed without secret-path)
3

Fetch Secrets

  • Multi-key: Fetch entire secret bundle using secret_path (if agent) or credential_guid (if direct)
  • Single-key: Fetch each field individually from secret store
4

Merge & Resolve

  • Direct credentials: Merge secret data directly into configuration
  • Agent credentials: Resolve references by substituting secret keys with values

​
Best Practices

Production Secrets: Never use save_secret() in production. It’s designed for development/testing only and will raise ValueError in production environments.
Error Handling: Always wrap credential resolution in try-except blocks to handle CommonError exceptions gracefully.
Local Development: In local environment (DEPLOYMENT_NAME == LOCAL_ENVIRONMENT), get_secret() returns empty dict to avoid secret store dependency.
Component Registration: Methods that depend on Dapr components (like get_deployment_secret) check for component registration and return None if unavailable.

​
Secret Store Configuration

Configure secret stores through constants: 
from application_sdk.constants import (
    SECRET_STORE_NAME,              # Default: "secretstore"
    DEPLOYMENT_SECRET_STORE_NAME,   # Default: "deployment-secretstore"
    DEPLOYMENT_SECRET_PATH,         # Default: "deployment-config"
)
  • StateStore - SecretStore uses StateStore for credential configuration
  • EventStore - Shows authentication pattern with AtlanAuthClient

Build docs developers (and LLMs) love

Get started for freeTalk to us