or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

configuration.mddestinations.mdhelpers.mdincremental.mdindex.mdpipeline.mdschema.mdsource-filesystem.mdsource-rest-api.mdsource-sql-database.mdsources-resources.md
tile.json

configuration.mddocs/

Configuration and Secrets

Hierarchical configuration system with multiple providers (TOML files, environment variables, secrets management) and dependency injection.

Capabilities

Configuration Accessors

# Top-level accessors
import dlt

config: Any  # Dictionary-like access to configuration values
secrets: Any  # Dictionary-like access to secret values

Configuration Decorator

def configspec(
    cls: type = None,
    init: bool = True,
    **kwargs
) -> type:
    """
    Marks a class as a configuration specification.

    Args:
        cls: Class to decorate
        init: Generate __init__ method
        **kwargs: Configuration options

    Returns:
        Decorated configuration class

    Example:
        @dlt.configspec
        class MyConfig:
            api_key: str = dlt.secrets.value
            timeout: int = 30
    """

Dependency Injection

def with_config(
    func: Callable = None,
    sections: tuple = None,
    auto_pipeline_section: bool = False,
    **kwargs
) -> Callable:
    """
    Decorator that injects configuration into function parameters.

    Args:
        func: Function to decorate
        sections: Configuration sections to search
        auto_pipeline_section: Use pipeline-specific section
        **kwargs: Additional options

    Returns:
        Decorated function with config injection

    Example:
        @dlt.source
        @dlt.with_config(sections=("sources", "myapi"))
        def my_source(api_key: str = dlt.secrets.value):
            return my_resource(api_key)
    """

Configuration Functions

def resolve_configuration(
    config: Any,
    sections: tuple = None,
    **kwargs
) -> Any:
    """
    Resolves configuration from providers.

    Args:
        config: Configuration object or class
        sections: Sections to search
        **kwargs: Additional options

    Returns:
        Resolved configuration instance
    """
def inject_section(
    section: str
) -> Callable:
    """
    Injects specific configuration section.

    Args:
        section: Section name

    Returns:
        Decorator function

    Example:
        @dlt.inject_section("sources.myapi")
        def my_function(config: MyConfig):
            ...
    """
def last_config() -> Any:
    """
    Gets the last resolved configuration.

    Returns:
        Last configuration instance or None
    """
def get_fun_spec(func: Callable) -> Any:
    """
    Gets function specification for configuration injection.

    Args:
        func: Function to inspect

    Returns:
        Function specification
    """
def create_resolved_partial(
    func: Callable,
    **kwargs
) -> Callable:
    """
    Creates partial function with resolved configuration.

    Args:
        func: Function to wrap
        **kwargs: Override values

    Returns:
        Partial function with config resolved
    """

Type Utilities

def is_valid_hint(hint: Any) -> bool:
    """
    Checks if type hint is valid for configuration.

    Args:
        hint: Type hint to check

    Returns:
        True if valid configuration hint
    """
def is_secret_hint(hint: Any) -> bool:
    """
    Checks if type hint marks a secret value.

    Args:
        hint: Type hint to check

    Returns:
        True if secret hint
    """
def resolve_type(hint: Any) -> type:
    """
    Resolves type from hint.

    Args:
        hint: Type hint

    Returns:
        Resolved type
    """
# Sentinel for unresolved values
NotResolved: Any

Configuration Sections

# Known configuration sections
known_sections: Container = {
    "sources",
    "destinations",
    "normalize",
    "load",
    "extract",
    "pipelines",
    "dbt_cloud",
    "runtime"
}

Usage Examples

Basic Configuration Access

import dlt

# Access configuration
api_url = dlt.config["api.url"]
timeout = dlt.config.get("api.timeout", 30)  # With default

# Access secrets
api_key = dlt.secrets["api.key"]
db_password = dlt.secrets["database.password"]

Configuration with Source

@dlt.source
def my_api_source(
    api_key: str = dlt.secrets.value,
    base_url: str = dlt.config.value,
    timeout: int = 30
):
    """
    Configuration automatically injected from:
    - secrets.toml: api_key
    - config.toml: base_url
    - Default: timeout
    """
    @dlt.resource
    def data():
        response = requests.get(
            f"{base_url}/data",
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=timeout
        )
        yield response.json()

    return data

TOML Configuration Files

config.toml:

[sources.my_api]
base_url = "https://api.example.com"
timeout = 60

[runtime]
log_level = "INFO"

secrets.toml:

[sources.my_api]
api_key = "sk_live_..."

[destination.postgres.credentials]
password = "secretpassword"

Environment Variables

# dlt automatically reads from environment:
# DLT__SOURCES__MY_API__API_KEY=sk_live_...
# DLT__SOURCES__MY_API__BASE_URL=https://api.example.com

@dlt.source
def my_api(api_key: str = dlt.secrets.value):
    # api_key loaded from DLT__SOURCES__MY_API__API_KEY
    ...

Custom Configuration Class

from dlt.common.configuration import configspec

@configspec
class MyAPIConfig:
    api_key: str = dlt.secrets.value
    base_url: str = dlt.config.value
    timeout: int = 30
    rate_limit: int = 100

@dlt.source
def my_api(config: MyAPIConfig = None):
    """
    Configuration automatically resolved from providers.
    """
    print(f"Connecting to {config.base_url}")
    print(f"Timeout: {config.timeout}s")
    ...

Explicit Configuration Sections

@dlt.source
@dlt.with_config(sections=("sources", "github"))
def github_source(
    access_token: str = dlt.secrets.value,
    owner: str = dlt.config.value,
    repo: str = dlt.config.value
):
    """
    Searches in:
    1. [sources.github] section
    2. [sources] section
    3. Top level
    """
    ...

Pipeline-Specific Configuration

# config.toml
[pipelines.my_pipeline.sources.my_api]
base_url = "https://dev-api.example.com"

[pipelines.prod_pipeline.sources.my_api]
base_url = "https://api.example.com"

# Automatically uses pipeline-specific config
pipeline = dlt.pipeline(pipeline_name="my_pipeline", ...)
pipeline.run(my_api_source())

Credentials Classes

from dlt.sources.credentials import ConnectionStringCredentials

@configspec
class DatabaseConfig:
    credentials: ConnectionStringCredentials
    pool_size: int = 5

# In secrets.toml:
# [sources.mydb.credentials]
# database = "mydb"
# username = "user"
# password = "secret"
# host = "localhost"

Runtime Configuration Override

# Override at runtime
pipeline.run(
    my_source(
        api_key="override_key",
        base_url="https://override.example.com"
    )
)

Nested Configuration

@configspec
class RetryConfig:
    max_attempts: int = 3
    backoff_factor: float = 2.0

@configspec
class APIConfig:
    api_key: str = dlt.secrets.value
    retry: RetryConfig = None

# In config.toml:
# [sources.myapi]
# api_key = "..." # Actually in secrets
#
# [sources.myapi.retry]
# max_attempts = 5
# backoff_factor = 1.5

Check for Configuration

from dlt.common.configuration import NotResolved, resolve_configuration

try:
    config = resolve_configuration(MyAPIConfig)
    print("Configuration resolved successfully")
except Exception:
    print("Missing required configuration")

Optional Configuration

@dlt.source
def flexible_source(
    required_key: str = dlt.secrets.value,
    optional_setting: str = None  # No dlt.config.value
):
    """
    required_key must be provided
    optional_setting can be None
    """
    ...

Configuration Providers Priority

Order (highest to lowest priority):

  1. Explicit arguments
  2. Environment variables (DLT__...)
  3. secrets.toml (for secrets)
  4. config.toml (for config)
  5. Default values

Global vs Pipeline Configuration

# Global config (applies to all pipelines)
[sources.myapi]
base_url = "https://api.example.com"

# Pipeline-specific (overrides global)
[pipelines.dev.sources.myapi]
base_url = "https://dev-api.example.com"

[pipelines.prod.sources.myapi]
base_url = "https://prod-api.example.com"

Configuration Exceptions

from dlt.common.configuration import (
    ConfigFieldMissingException,
    ConfigValueCannotBeCoercedException,
    ConfigFileNotFoundException,
    ConfigurationValueError
)

try:
    pipeline.run(my_source())
except ConfigFieldMissingException as e:
    print(f"Missing config field: {e}")
except ConfigValueCannotBeCoercedException as e:
    print(f"Invalid config value: {e}")

Best Practices

  1. Secrets in secrets.toml: Never commit secrets to version control
  2. Config in config.toml: Commit non-sensitive configuration
  3. Environment variables in production: Use env vars for deployment
  4. Type hints for validation: Use types to validate configuration
  5. Default values: Provide sensible defaults where possible
  6. Use @configspec: For complex configuration structures
  7. Pipeline-specific config: Use pipeline sections for multi-environment