Sources

Configuration sources module.

Provides various configuration sources: - Defaults: From dataclass model defaults - DotEnv: From .env files - Env: From environment variables - CLI: From command-line arguments - YAML: From YAML files (optional, requires ‘yaml’ extra) - JSON: From JSON files (standard library, no extra required) - TOML: From TOML files (optional, requires ‘toml’ extra for Python < 3.11) - Etcd: From etcd key-value store (optional, requires ‘etcd’ extra)

Note: Optional sources require their respective extras to be installed.

class varlord.sources.Source(model=None, source_id=None)[source]

Bases: object

Base class for all configuration sources.

Each source must implement: - load() -> Mapping[str, Any]: Load configuration snapshot - name: str: Source name for debugging and priority configuration

Optional: - watch() -> Iterator[ChangeEvent]: Stream of changes for dynamic updates

__init__(model=None, source_id=None)[source]

Initialize Source.

Parameters:
  • model (Type[Any] | None) – Optional dataclass model for field filtering. If None, model will be auto-injected by Config when used in Config. If provided, this model will be used (allows override).

  • source_id (str | None) – Optional unique identifier for this source. If None, will be auto-generated based on source type and key parameters.

Note

  • Recommended: Omit model parameter when used in Config (auto-injected).

  • Advanced: Provide model explicitly if using source independently or need different model.

  • source_id: If not provided, will be auto-generated. Provide custom ID for advanced use cases.

__repr__()[source]

Return string representation.

property id: str

Return unique source identifier.

Returns:

Unique identifier string. This is used for precise source identification in PriorityPolicy. Each source instance should have a unique ID.

load()[source]

Load configuration from this source.

Returns:

A mapping of configuration key-value pairs. Keys should be normalized (e.g., “db.host” for nested configs).

Return type:

Mapping[str, Any]

property load_error: str | None

Return load error message if load failed.

Returns:

Error message string or None if load succeeded or file not found. Note: 文件不存在(not_found)不记录错误信息,因为这是正常情况。

property load_status: str

‘success’, ‘failed’, ‘not_found’, ‘unknown’.

Returns:

Status string indicating the last load() call result. - ‘success’: 成功加载 - ‘not_found’: 文件不存在(正常情况,如本地没有 .env 文件) - ‘failed’: 真正的错误(如文件格式错误、权限问题等) - ‘unknown’: 未知状态

Type:

Return load status

property name: str

Return the source name.

Returns:

Source name, used for debugging and priority configuration. This is used for type identification and grouping. Multiple sources can have the same name.

supports_watch()[source]

Check if this source supports watching.

Returns:

False by default. Subclasses that support watching must override this method to return True when watch is enabled.

Return type:

bool

Note

To enable watch support, override this method in your subclass and return True when watch should be enabled.

watch()[source]

Watch for configuration changes (optional).

Yields:

ChangeEvent objects representing configuration changes.

Note

This method is optional. Only sources that support dynamic updates (like etcd) should implement this.

To enable watch support, override this method in your subclass. The default implementation returns an empty iterator, which indicates the source does not support watching.

class varlord.sources.ChangeEvent(key, old_value, new_value, event_type)[source]

Bases: object

Represents a configuration change event.

key

The configuration key that changed

Type:

str

old_value

The old value (None if key was added)

Type:

Any

new_value

The new value (None if key was removed)

Type:

Any

event_type

Type of change (‘added’, ‘modified’, ‘deleted’)

Type:

str

__init__(key, old_value, new_value, event_type)
key: str
old_value: Any
new_value: Any
event_type: str
class varlord.sources.Defaults(model, source_id=None)[source]

Bases: Source

Source that loads default values from a dataclass model.

This source extracts default values from dataclass field definitions. It is automatically used by Config as the first source (base layer).

Example

>>> @dataclass
... class Config:
...     host: str = "localhost"
...     port: int = 8000
...
>>> source = Defaults(model=Config)
>>> source.load()
{'host': 'localhost', 'port': 8000}
__init__(model, source_id=None)[source]

Initialize Defaults source.

Parameters:
  • model (Type[Any]) – The dataclass model to extract defaults from.

  • source_id (str | None) – Optional unique identifier (default: “defaults”)

Raises:

TypeError – If model is not a dataclass

__repr__()[source]

Return string representation.

load()[source]

Load default values from the model.

Returns:

A mapping of normalized keys to their default values. Fields without defaults are excluded. Supports nested fields (e.g., {“db.host”: “localhost”}).

Return type:

Mapping[str, Any]

property name: str

Return source name.

class varlord.sources.Env(model=None, prefix=None, source_id=None)[source]

Bases: Source

Source that loads configuration from environment variables.

Only loads environment variables that map to fields defined in the model. Model is required and will be auto-injected by Config if not provided.

Example

>>> @dataclass
... class Config:
...     api_key: str = field()  # Required by default
>>> # Environment: API_KEY=value1 OTHER_VAR=ignored
>>> source = Env(model=Config)
>>> source.load()
{'api_key': 'value1'}  # OTHER_VAR is ignored
__init__(model=None, prefix=None, source_id=None)[source]

Initialize Env source.

Parameters:
  • model (Type[Any] | None) – Optional model to filter environment variables. Only variables that map to model fields will be loaded. If None, model will be auto-injected by Config when used in Config. If provided, this model will be used (allows override).

  • prefix (str | None) – Optional prefix for environment variables (e.g., TITAN__). If None, matches all environment variables that map to model fields. If provided, only variables starting with this prefix (uppercase) are considered.

  • source_id (str | None) – Optional unique identifier (default: “env” or “env:{prefix}”)

Note

  • Prefix filtering is optional. If prefix is None, all env vars are checked against model fields.

  • Recommended: Omit model parameter when used in Config (auto-injected).

  • Advanced: Provide model explicitly if using source independently.

__repr__()[source]

Return string representation.

load()[source]

Load configuration from environment variables, filtered by model fields.

Returns:

A mapping of normalized keys to environment variable values. Only includes variables that map to model fields.

Raises:

ValueError – If model is not provided

Return type:

Mapping[str, Any]

property name: str

Return source name.

class varlord.sources.CLI(model=None, argv=None, source_id=None)[source]

Bases: Source

Source that loads configuration from command-line arguments.

Uses argparse to parse command-line arguments. Only adds arguments for fields defined in the model. Model is required and will be auto-injected by Config.

Mapping rules: - Double dashes (–) in CLI arguments become dots (.) in normalized keys - Single dashes (-) in CLI arguments become underscores (_) in normalized keys

Examples: - –host → host - –k8s-pod-name → k8s_pod_name - –db–host → db.host - –aaa–bbb–ccc-dd → aaa.bbb.ccc_dd

Supports: - Automatic type inference from model fields - Boolean flags (–flag / –no-flag) - Nested keys via double dashes (–db–host maps to db.host) - Help text from field metadata

Example

>>> @dataclass
... class Config:
...     host: str = field()
>>> # Command line: python app.py --host 0.0.0.0
>>> source = CLI(model=Config)
>>> source.load()
{'host': '0.0.0.0'}
__init__(model=None, argv=None, source_id=None)[source]

Initialize CLI source.

Parameters:
  • model (Type[Any] | None) – Optional model to filter CLI arguments. Only arguments that map to model fields will be parsed. If None, model will be auto-injected by Config when used in Config. If provided, this model will be used (allows override).

  • argv (List[str] | None) – Command-line arguments (default: sys.argv[1:])

  • source_id (str | None) – Optional unique identifier (default: “cli”)

Note

  • Recommended: Omit model parameter when used in Config (auto-injected).

  • Advanced: Provide model explicitly if using source independently.

__repr__()[source]

Return string representation.

format_help(prog=None)[source]

Generate help text for all CLI arguments based on model fields.

This method generates help text without using argparse’s built-in help, giving varlord complete control over the help output format.

Parameters:

prog (str | None) – Program name (default: script name from sys.argv[0])

Returns:

Formatted help text string

Return type:

str

get_field_help(field_key)[source]

Get help text for a specific field.

Parameters:

field_key (str) – Normalized field key (e.g., “docx_path” or “db.host”)

Returns:

Help text if found, None otherwise

Return type:

str | None

load()[source]

Load configuration from command-line arguments, filtered by model fields.

Returns:

A mapping of normalized keys (using dot notation) to their values. Only includes arguments for model fields.

Raises:

ValueError – If model is not provided

Return type:

Mapping[str, Any]

property name: str

Return source name.

class varlord.sources.JSON(file_path, model=None, source_id=None, required=False)[source]

Bases: FileSource

Source that loads configuration from JSON files.

Uses standard library json module (no extra dependencies).

Only loads keys that map to fields defined in the model. Model is required and will be auto-injected by Config.

Example

>>> @dataclass
... class Config:
...     host: str = field()
>>> # config.json: {"host": "0.0.0.0"}
>>> source = JSON("config.json", model=Config)
>>> source.load()
{'host': '0.0.0.0'}
__init__(file_path, model=None, source_id=None, required=False)[source]

Initialize JSON source.

Parameters:
  • file_path (str) – Path to JSON file

  • model (Type[Any] | None) – Optional model for field filtering

  • source_id (str | None) – Optional unique identifier (default: auto-generated from path)

  • required (bool) – If True, raise error when file not found

property name: str

Return source name.

class varlord.sources.DotEnv(dotenv_path='.env', model=None, encoding=None, source_id=None)[source]

Bases: Source

Source that loads configuration from .env files.

Requires the ‘dotenv’ extra: pip install varlord[dotenv]

Only loads variables that map to fields defined in the model. Model is required and will be auto-injected by Config.

Example

>>> @dataclass
... class Config:
...     api_key: str = field()  # Required by default
>>> # .env file: API_KEY=value1 OTHER_VAR=ignored
>>> source = DotEnv(".env", model=Config)
>>> source.load()
{'api_key': 'value1'}  # OTHER_VAR is ignored
__init__(dotenv_path='.env', model=None, encoding=None, source_id=None)[source]

Initialize DotEnv source.

Parameters:
  • dotenv_path (str) – Path to .env file

  • model (Type[Any] | None) – Model to filter .env variables. Only variables that map to model fields will be loaded. Model is required and will be auto-injected by Config.

  • encoding (str | None) – File encoding (default: None, uses system default)

  • source_id (str | None) – Optional unique identifier (default: auto-generated from path)

Raises:

ImportError – If python-dotenv is not installed

__repr__()[source]

Return string representation.

load()[source]

Load configuration from .env file, filtered by model fields.

Returns:

A mapping of normalized keys to their values. Only includes variables that map to model fields.

Raises:

ValueError – If model is not provided

Return type:

Mapping[str, Any]

property name: str

Return source name.

class varlord.sources.YAML(file_path, model=None, source_id=None, required=False)[source]

Bases: FileSource

Source that loads configuration from YAML files.

Requires the ‘yaml’ extra: pip install varlord[yaml]

Only loads keys that map to fields defined in the model. Model is required and will be auto-injected by Config.

Example

>>> @dataclass
... class Config:
...     host: str = field()
>>> # config.yaml: host: 0.0.0.0
>>> source = YAML("config.yaml", model=Config)
>>> source.load()
{'host': '0.0.0.0'}
__init__(file_path, model=None, source_id=None, required=False)[source]

Initialize YAML source.

Parameters:
  • file_path (str) – Path to YAML file

  • model (Type[Any] | None) – Optional model for field filtering

  • source_id (str | None) – Optional unique identifier (default: auto-generated from path)

  • required (bool) – If True, raise error when file not found

Raises:

ImportError – If pyyaml is not installed

property name: str

Return source name.

class varlord.sources.TOML(file_path, model=None, source_id=None, required=False)[source]

Bases: FileSource

Source that loads configuration from TOML files.

Requires the ‘toml’ extra for Python < 3.11: pip install varlord[toml] Python 3.11+ has built-in tomllib support.

Only loads keys that map to fields defined in the model. Model is required and will be auto-injected by Config.

Example

>>> @dataclass
... class Config:
...     host: str = field()
>>> # config.toml: host = "0.0.0.0"
>>> source = TOML("config.toml", model=Config)
>>> source.load()
{'host': '0.0.0.0'}
__init__(file_path, model=None, source_id=None, required=False)[source]

Initialize TOML source.

Parameters:
  • file_path (str) – Path to TOML file

  • model (Type[Any] | None) – Optional model for field filtering

  • source_id (str | None) – Optional unique identifier (default: auto-generated from path)

  • required (bool) – If True, raise error when file not found

Raises:

ImportError – If tomli is not installed (Python < 3.11)

property name: str

Return source name.

class varlord.sources.Etcd(host='127.0.0.1', port=2379, prefix='/', watch=False, timeout=None, model=None, ca_cert=None, cert_key=None, cert_cert=None, user=None, password=None, source_id=None)[source]

Bases: Source

Source that loads configuration from etcd.

Requires the etcd extra: pip install varlord[etcd]

Supports: - Loading configuration from a prefix - TLS/SSL certificate authentication - User authentication - Watching for changes (dynamic updates) - Automatic reconnection on connection loss

Basic Example:
>>> source = Etcd(
...     host="127.0.0.1",
...     port=2379,
...     prefix="/app/",
... )
>>> source.load()
{'host': '0.0.0.0', 'port': '9000'}
With TLS:
>>> source = Etcd(
...     host="192.168.0.220",
...     port=2379,
...     prefix="/app/",
...     ca_cert="./cert/ca.cert.pem",
...     cert_key="./cert/key.pem",
...     cert_cert="./cert/cert.pem",
... )

Note

⚠️ 重要变更:不再提供 from_env() 类方法。 所有参数都通过 __init__ 传递,调用方负责获取初始配置信息。 可以使用 Env source 或其他方式获取连接参数。

__init__(host='127.0.0.1', port=2379, prefix='/', watch=False, timeout=None, model=None, ca_cert=None, cert_key=None, cert_cert=None, user=None, password=None, source_id=None)[source]

Initialize Etcd source.

Parameters:
  • host (str) – Etcd host

  • port (int) – Etcd port

  • prefix (str) – Key prefix to load (e.g., “/app/”)

  • watch (bool) – Whether to enable watch support

  • timeout (int | None) – Connection timeout in seconds

  • model (Type[Any] | None) – Model to filter etcd keys. Only keys that map to model fields will be loaded. Model is required and will be auto-injected by Config.

  • ca_cert (str | None) – Path to CA certificate file for TLS

  • cert_key (str | None) – Path to client key file for TLS

  • cert_cert (str | None) – Path to client certificate file for TLS

  • user (str | None) – Username for authentication (optional)

  • password (str | None) – Password for authentication (optional)

  • source_id (str | None) – Optional unique identifier (default: auto-generated)

Raises:

ImportError – If etcd3 is not installed

Note

⚠️ 重要变更:不再提供 from_env() 类方法。 所有参数都通过 __init__ 传递,调用方负责获取初始配置信息。

__repr__()[source]

Return string representation.

load()[source]

Load configuration from etcd, filtered by model fields.

Returns:

A mapping of configuration keys to values. Keys are normalized (prefix removed, converted to dot notation). Only includes keys that map to model fields.

Raises:

ValueError – If model is not provided

Return type:

Mapping[str, Any]

property name: str

Return source name.

supports_watch()[source]

Check if etcd source supports watching.

Returns:

True if watch is enabled.

Return type:

bool

watch()[source]

Watch for configuration changes in etcd.

Yields:

ChangeEvent objects representing configuration changes.

Note

This method blocks and yields events as they occur. It should be run in a separate thread.

Default Sources

Defaults source.

Loads default values from dataclass model fields. This is now an internal source used automatically by Config.

class varlord.sources.defaults.Defaults(model, source_id=None)[source]

Bases: Source

Source that loads default values from a dataclass model.

This source extracts default values from dataclass field definitions. It is automatically used by Config as the first source (base layer).

Example

>>> @dataclass
... class Config:
...     host: str = "localhost"
...     port: int = 8000
...
>>> source = Defaults(model=Config)
>>> source.load()
{'host': 'localhost', 'port': 8000}
__init__(model, source_id=None)[source]

Initialize Defaults source.

Parameters:
  • model (Type[Any]) – The dataclass model to extract defaults from.

  • source_id (str | None) – Optional unique identifier (default: “defaults”)

Raises:

TypeError – If model is not a dataclass

property name: str

Return source name.

load()[source]

Load default values from the model.

Returns:

A mapping of normalized keys to their default values. Fields without defaults are excluded. Supports nested fields (e.g., {“db.host”: “localhost”}).

Return type:

Mapping[str, Any]

__repr__()[source]

Return string representation.

Environment variable source.

Loads configuration from environment variables, filtered by model fields.

class varlord.sources.env.Env(model=None, prefix=None, source_id=None)[source]

Bases: Source

Source that loads configuration from environment variables.

Only loads environment variables that map to fields defined in the model. Model is required and will be auto-injected by Config if not provided.

Example

>>> @dataclass
... class Config:
...     api_key: str = field()  # Required by default
>>> # Environment: API_KEY=value1 OTHER_VAR=ignored
>>> source = Env(model=Config)
>>> source.load()
{'api_key': 'value1'}  # OTHER_VAR is ignored
__init__(model=None, prefix=None, source_id=None)[source]

Initialize Env source.

Parameters:
  • model (Type[Any] | None) – Optional model to filter environment variables. Only variables that map to model fields will be loaded. If None, model will be auto-injected by Config when used in Config. If provided, this model will be used (allows override).

  • prefix (str | None) – Optional prefix for environment variables (e.g., TITAN__). If None, matches all environment variables that map to model fields. If provided, only variables starting with this prefix (uppercase) are considered.

  • source_id (str | None) – Optional unique identifier (default: “env” or “env:{prefix}”)

Note

  • Prefix filtering is optional. If prefix is None, all env vars are checked against model fields.

  • Recommended: Omit model parameter when used in Config (auto-injected).

  • Advanced: Provide model explicitly if using source independently.

property name: str

Return source name.

load()[source]

Load configuration from environment variables, filtered by model fields.

Returns:

A mapping of normalized keys to environment variable values. Only includes variables that map to model fields.

Raises:

ValueError – If model is not provided

Return type:

Mapping[str, Any]

__repr__()[source]

Return string representation.

Command-line argument source.

Loads configuration from command-line arguments using argparse. Only parses arguments for fields defined in the model.

varlord.sources.cli.normalized_key_to_cli_arg(normalized_key)[source]

Convert normalized key to CLI argument format.

Mapping rules: - Dots (.) become double dashes (–) - Underscores (_) become single dashes (-)

Examples: - “host” -> “host” - “k8s_pod_name” -> “k8s-pod-name” - “db.host” -> “db–host” - “aaa.bbb.ccc_dd” -> “aaa–bbb–ccc-dd”

varlord.sources.cli.cli_arg_to_normalized_key(cli_arg)[source]

Convert CLI argument to normalized key.

Mapping rules: - Double dashes (–) become dots (.) - Single dashes (-) become underscores (_)

Examples: - “host” -> “host” - “k8s-pod-name” -> “k8s_pod_name” - “db–host” -> “db.host” - “aaa–bbb–ccc-dd” -> “aaa.bbb.ccc_dd”

class varlord.sources.cli.CLI(model=None, argv=None, source_id=None)[source]

Bases: Source

Source that loads configuration from command-line arguments.

Uses argparse to parse command-line arguments. Only adds arguments for fields defined in the model. Model is required and will be auto-injected by Config.

Mapping rules: - Double dashes (–) in CLI arguments become dots (.) in normalized keys - Single dashes (-) in CLI arguments become underscores (_) in normalized keys

Examples: - –host → host - –k8s-pod-name → k8s_pod_name - –db–host → db.host - –aaa–bbb–ccc-dd → aaa.bbb.ccc_dd

Supports: - Automatic type inference from model fields - Boolean flags (–flag / –no-flag) - Nested keys via double dashes (–db–host maps to db.host) - Help text from field metadata

Example

>>> @dataclass
... class Config:
...     host: str = field()
>>> # Command line: python app.py --host 0.0.0.0
>>> source = CLI(model=Config)
>>> source.load()
{'host': '0.0.0.0'}
__init__(model=None, argv=None, source_id=None)[source]

Initialize CLI source.

Parameters:
  • model (Type[Any] | None) – Optional model to filter CLI arguments. Only arguments that map to model fields will be parsed. If None, model will be auto-injected by Config when used in Config. If provided, this model will be used (allows override).

  • argv (List[str] | None) – Command-line arguments (default: sys.argv[1:])

  • source_id (str | None) – Optional unique identifier (default: “cli”)

Note

  • Recommended: Omit model parameter when used in Config (auto-injected).

  • Advanced: Provide model explicitly if using source independently.

property name: str

Return source name.

load()[source]

Load configuration from command-line arguments, filtered by model fields.

Returns:

A mapping of normalized keys (using dot notation) to their values. Only includes arguments for model fields.

Raises:

ValueError – If model is not provided

Return type:

Mapping[str, Any]

format_help(prog=None)[source]

Generate help text for all CLI arguments based on model fields.

This method generates help text without using argparse’s built-in help, giving varlord complete control over the help output format.

Parameters:

prog (str | None) – Program name (default: script name from sys.argv[0])

Returns:

Formatted help text string

Return type:

str

get_field_help(field_key)[source]

Get help text for a specific field.

Parameters:

field_key (str) – Normalized field key (e.g., “docx_path” or “db.host”)

Returns:

Help text if found, None otherwise

Return type:

str | None

__repr__()[source]

Return string representation.

DotEnv source.

Loads configuration from .env files using python-dotenv. Only loads variables that map to fields defined in the model.

class varlord.sources.dotenv.DotEnv(dotenv_path='.env', model=None, encoding=None, source_id=None)[source]

Bases: Source

Source that loads configuration from .env files.

Requires the ‘dotenv’ extra: pip install varlord[dotenv]

Only loads variables that map to fields defined in the model. Model is required and will be auto-injected by Config.

Example

>>> @dataclass
... class Config:
...     api_key: str = field()  # Required by default
>>> # .env file: API_KEY=value1 OTHER_VAR=ignored
>>> source = DotEnv(".env", model=Config)
>>> source.load()
{'api_key': 'value1'}  # OTHER_VAR is ignored
__init__(dotenv_path='.env', model=None, encoding=None, source_id=None)[source]

Initialize DotEnv source.

Parameters:
  • dotenv_path (str) – Path to .env file

  • model (Type[Any] | None) – Model to filter .env variables. Only variables that map to model fields will be loaded. Model is required and will be auto-injected by Config.

  • encoding (str | None) – File encoding (default: None, uses system default)

  • source_id (str | None) – Optional unique identifier (default: auto-generated from path)

Raises:

ImportError – If python-dotenv is not installed

property name: str

Return source name.

load()[source]

Load configuration from .env file, filtered by model fields.

Returns:

A mapping of normalized keys to their values. Only includes variables that map to model fields.

Raises:

ValueError – If model is not provided

Return type:

Mapping[str, Any]

__repr__()[source]

Return string representation.

Etcd Source

Etcd source.

Loads configuration from etcd with optional watch support for dynamic updates. This is an optional source that requires the etcd extra.

class varlord.sources.etcd.Etcd(host='127.0.0.1', port=2379, prefix='/', watch=False, timeout=None, model=None, ca_cert=None, cert_key=None, cert_cert=None, user=None, password=None, source_id=None)[source]

Bases: Source

Source that loads configuration from etcd.

Requires the etcd extra: pip install varlord[etcd]

Supports: - Loading configuration from a prefix - TLS/SSL certificate authentication - User authentication - Watching for changes (dynamic updates) - Automatic reconnection on connection loss

Basic Example:
>>> source = Etcd(
...     host="127.0.0.1",
...     port=2379,
...     prefix="/app/",
... )
>>> source.load()
{'host': '0.0.0.0', 'port': '9000'}
With TLS:
>>> source = Etcd(
...     host="192.168.0.220",
...     port=2379,
...     prefix="/app/",
...     ca_cert="./cert/ca.cert.pem",
...     cert_key="./cert/key.pem",
...     cert_cert="./cert/cert.pem",
... )

Note

⚠️ 重要变更:不再提供 from_env() 类方法。 所有参数都通过 __init__ 传递,调用方负责获取初始配置信息。 可以使用 Env source 或其他方式获取连接参数。

__init__(host='127.0.0.1', port=2379, prefix='/', watch=False, timeout=None, model=None, ca_cert=None, cert_key=None, cert_cert=None, user=None, password=None, source_id=None)[source]

Initialize Etcd source.

Parameters:
  • host (str) – Etcd host

  • port (int) – Etcd port

  • prefix (str) – Key prefix to load (e.g., “/app/”)

  • watch (bool) – Whether to enable watch support

  • timeout (int | None) – Connection timeout in seconds

  • model (Type[Any] | None) – Model to filter etcd keys. Only keys that map to model fields will be loaded. Model is required and will be auto-injected by Config.

  • ca_cert (str | None) – Path to CA certificate file for TLS

  • cert_key (str | None) – Path to client key file for TLS

  • cert_cert (str | None) – Path to client certificate file for TLS

  • user (str | None) – Username for authentication (optional)

  • password (str | None) – Password for authentication (optional)

  • source_id (str | None) – Optional unique identifier (default: auto-generated)

Raises:

ImportError – If etcd3 is not installed

Note

⚠️ 重要变更:不再提供 from_env() 类方法。 所有参数都通过 __init__ 传递,调用方负责获取初始配置信息。

property name: str

Return source name.

load()[source]

Load configuration from etcd, filtered by model fields.

Returns:

A mapping of configuration keys to values. Keys are normalized (prefix removed, converted to dot notation). Only includes keys that map to model fields.

Raises:

ValueError – If model is not provided

Return type:

Mapping[str, Any]

supports_watch()[source]

Check if etcd source supports watching.

Returns:

True if watch is enabled.

Return type:

bool

watch()[source]

Watch for configuration changes in etcd.

Yields:

ChangeEvent objects representing configuration changes.

Note

This method blocks and yields events as they occur. It should be run in a separate thread.

__repr__()[source]

Return string representation.