企业官网建设流程全解析

dlt 配置注入与密钥管理实战：在代码中访问与编写配置、自定义 Spec 与分层布局

【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy 🛠️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt

dlt（data load tool）在@dlt.source、@dlt.resource、@dlt.destination装饰的函数上自动生成配置规范（spec），并依据注入规则从环境变量、secrets.toml/config.toml、Vault 等配置提供方（provider）中注入缺失参数。本文以 docs/website/docs/general-usage/credentials/advanced.md 为主线，深入讲解注入规则的细节、sections 分层布局、dlt.config/dlt.secrets的字典式读写、代码内配置目标凭据，以及如何通过自定义 Spec（基于BaseConfiguration与CredentialsConfiguration）完全掌控配置解析，并给出对应源码级实现依据。

一、dlt 装饰函数中的配置注入机制

dlt会为@dlt.source、@dlt.resource、@dlt.destination装饰的函数自动生成配置spec，无需额外编写代码。这些函数可以使用标准配置方法（环境变量、TOML 文件、Vault、自定义 provider）进行配置。调用时，对于任何未显式提供的参数，dlt会从配置提供方中注入对应值；也可以像普通 Python 函数一样显式传参——注入机制完全可选。

从源码看，这一机制的核心实现在 dlt/common/configuration/inject.py 的with_config装饰器中：它根据函数签名合成 spec，在每次调用时通过resolve_configuration解析配置，再用update_bound_args将解析结果回填到函数参数。spec 本身由 dlt/common/reflection/spec.py 的spec_from_signature从签名动态合成：只有「带默认值且类型合法」的参数才会进入 spec，spec 的类名由函数限定名（qualname）推断并注册到函数所在模块，同名函数在模块内唯一缓存。

注入规则（Injection rules）

规则 1：显式传入的参数永远不会被注入。这让注入机制变成可选项。以 Pipedrive source 为例：

import os from typing import Iterator from dlt.extract import DltResource @dlt.source(name="pipedrive") def pipedrive_source( pipedrive_api_key: str = dlt.secrets.value, since_timestamp: pendulum.DateTime | str | None = "1970-01-01 00:00:00", ) -> Iterator[DltResource]: ... my_key = os.environ["MY_PIPEDRIVE_KEY"] my_source = pipedrive_source(pipedrive_api_key=my_key)

如果你不想走标准凭据处理流程，可以像上面这样显式指定pipedrive_api_key。源码中，显式参数被作为explicit_value传入resolve_configuration，解析时会优先采用、不再查找 provider（见 dlt/common/configuration/resolve.py）。

规则 2：无默认值的必填参数永远不会被注入，调用时必须显式指定。例如：

@dlt.source def slack_data(channels_list: list[str], api_key: str = dlt.secrets.value): ...

channels_list不会被注入，若不显式传入将直接报错。原因见spec_from_signature中的过滤逻辑：只有p.default != Parameter.empty（即带默认值）的参数才进入合成 spec（dlt/common/reflection/spec.py）。

规则 3：带默认值的参数，若能在配置提供方中找到则注入，否则回退到函数签名中的默认值。例如：

from dlt.common.typing import TAnyDateTime START_DATE: pendulum.DateTime = pendulum.DateTime(2024, 1, 1) @dlt.source def slack_source( page_size: int = 100, access_token: str = dlt.secrets.value, start_date: TAnyDateTime | None = START_DATE ): ...

dlt会按照特定顺序先在 provider 中查找page_size、access_token、start_date，找不到再使用默认值。include_defaults=True是with_config的默认行为，即带默认值的参数默认会纳入合成 spec（dlt/common/configuration/inject.py）。

规则 4：默认值为dlt.secrets.value或dlt.config.value的参数必须被注入（或显式传入）。若在 provider 中找不到，dlt会抛出异常。此外，dlt.secrets.value向dlt表明该值是密钥，只会从安全配置提供方（secrets 类 provider）注入。在源码中，这类参数经spec_from_signature处理后默认值被替换为None并标记为必填；若参数本身带类型注解，还会被包装为Annotated[field_type, SecretSentinel]以标记密钥语义（dlt/common/reflection/spec.py）。

为 source 和 resource 添加类型注解

强烈建议为函数签名添加类型注解，成本极低、收益显著：

1. 你不会在代码中收到非法数据类型；
2. dlt自动解析并转换类型，无需手动解析；
3. dlt可以自动为 source 生成示例 config 和 secret 文件；
4. 可以请求内置与自定义凭据（连接字符串、AWS/GCP/Azure 凭据）；
5. 可通过Union指定多种可能类型，例如 OAuth 或 API Key 两种鉴权方式。

示例：

from dlt.common.configuration.specs import GcpServiceAccountCredentials @dlt.source def google_sheets( spreadsheet_id: str = dlt.config.value, tab_names: list[str] = dlt.config.value, credentials: GcpServiceAccountCredentials = dlt.secrets.value, only_strings: bool = False ): ...

收益：

1. 你会得到类型正确的tab_names字符串列表；
2. 你会得到配置正确的 Google 凭据（详见 GCP Credential Configuration），用户可以用多种形式提供：
   • service.json字符串或字典（代码中或通过配置提供方）；
   • 连接字符串（用于 SQL Alchemy）；
   • 不传任何值时的默认凭据（例如 Cloud Function 运行时自带的凭据）。

类型解析与转换由resolve.py中的deserialize_value等逻辑完成：list[str]会从 provider 值（TOML 数组或环境变量的 Python 字面量）反序列化为列表；GcpServiceAccountCredentials这类凭据 spec 则会通过initialize_credentials从原生表示（连接字符串或 service.json）实例化。

二、用 sections 组织配置与密钥

dlt将配置与密钥 section 组织成与注入机制集成的配置布局（configuration layout），该结构适用于所有配置提供方，包括 TOML 文件、环境变量等。

这种层级结构既能高效处理简单场景，也支持复杂场景，例如多个 source 使用不同凭据，或同一项目内多个 pipeline 共享配置：

pipeline_name | |-sources |-<source 1 module name> |-<source function 1 name> |- {all source and resource options and secrets} |-<source function 2 name> |- {all source and resource options and secrets} |-<source 2 module> |... |-extract |- extract options for resources i.e., parallelism settings, maybe retries |-destination |- <destination name> |- {destination options} |-credentials |-{credentials options} |-schema |-<schema name> |-schema settings: not implemented but I'll let people set nesting level, name convention, normalizer, etc. here |-load |-normalize

在 TOML 文件中，该结构表现为带点号的嵌套 section；对于环境变量等 provider，布局用双下划线展平（例如PIPELINE_NAME__SOURCES__MODULE_NAME__FUNCTION_NAME__OPTION）。pipeline 名称优先作为顶层 section：dlt会先带 pipeline 名前缀查找，再不带前缀查找，因此可以在同一项目里为多个 pipeline 维护隔离的配置。

紧凑 source 布局（Compact sources layout）

当 source 的section（通常是模块名）与其name（函数名）不同时，dlt还接受一条更短的、直接在sources下使用 source name 的配置路径：

sources.<name>.<key>

这是对完整路径sources.<section>.<name>.<key>的补充。完整路径与 section 路径（sources.<section>.<key>）都优先于紧凑路径。这在通过.clone()重命名 source 时特别有用：

# compact layout — just the source name [sources.my_db.credentials] password="..." # full layout — section + name (takes precedence) [sources.my_db_module.my_db.credentials] password="..."

clone()的典型用法是sql_database.clone(name="my_db", section="my_db_module")，为同一模块的多个实例建立不同配置 section（参见 docs/website/docs/general-usage/source.md）。从源码看，紧凑布局的支持定义在 dlt/common/configuration/resolve.py：

COMPACT_LAYOUT_SECTIONS: Set[str] = {known_sections.SOURCES} """Top-level sections that support compact config layout (top.name as shortcut for top.section.name)."""

即只有sources顶层 section 支持「top.name 作为 top.section.name 的快捷方式」这一紧凑写法。查找顺序（以notion.py中的notion_databases为例）为：

1. sources.notion.notion_databases.api_key
2. sources.notion.api_key
3. sources.api_key
4. api_key

当 section 与 name 不同（例如clone(name="my_db", section="my_db_module")）时：

1. sources.my_db_module.my_db.api_key（完整路径）
2. sources.my_db_module.api_key
3. sources.my_db.api_key（紧凑路径）
4. sources.api_key
5. api_key

destination 凭据类似，但credentialssection 被视为必选分组、不会被消除：

1. destination.postgres.credentials.password
2. destination.credentials.password
3. credentials.password

三、在代码中访问配置与密钥

dlt会自动处理凭据，但你也可以直接在代码中访问它们。dlt.secrets与dlt.config对象提供类似字典的访问方式，可读取配置值与密钥，用于自定义预处理；你也可以在同一个配置文件中存放自定义设置。

# Use `dlt.secrets` and `dlt.config` to explicitly retrieve values from providers source_instance = google_sheets( dlt.config["sheet_id"], dlt.config["my_section.tabs"], dlt.secrets["my_section.gcp_credentials"] ) source_instance.run(destination="bigquery")

dlt.config与dlt.secrets的行为类似字典：dlt会检查所有配置提供方——环境变量、TOML 文件等——来填充这些字典。还可以用dlt.config.get()或dlt.secrets.get()取回值并转换成指定类型：

from dlt.common.configuration.specs import GcpServiceAccountCredentials credentials = dlt.secrets.get("my_section.gcp_credentials", GcpServiceAccountCredentials)

这会从my_section.gcp_credentials键下存储的值创建出GcpServiceAccountCredentials实例。其实现位于 dlt/common/configuration/accessors.py：dlt.config与dlt.secrets本质上是配置访问器，内部依次查询已注册的 provider，并支持「读取 + 类型转换 + 写入」三种操作；dlt.secrets在get时会按TSecretValue类型处理并只会查询支持密钥的 provider。

四、在代码中编写配置与密钥

你也可以用dlt.config与dlt.secrets以编程方式设置值：

dlt.config["sheet_id"] = "23029402349032049" dlt.secrets["destination.postgres.credentials"] = BaseHook.get_connection('postgres_dsn').extra

这实际上是用你指定的值「模拟」了 TOML provider：写入的值会被后续配置解析当作来自config.toml（对dlt.config）或secrets.toml（对dlt.secrets）处理（见 dlt/common/configuration/accessors.py 的说明）。因此dlt.secrets只能写入安全 provider，而dlt.config写入非敏感配置。适合在测试、Airflow 等需要动态注入凭据的环境中组合使用。

五、在代码中配置 destination 凭据

你可以在需要时以编程方式设置 destination 凭据。下面的例子演示了如何将 GcpServiceAccountCredentialsspec用于 BigQuery destination：

import os import dlt from dlt.sources.credentials import GcpServiceAccountCredentials from dlt.destinations import bigquery # Retrieve credentials from environment variable creds_dict = os.getenv('BIGQUERY_CREDENTIALS') # Create and initialize credentials instance gcp_credentials = GcpServiceAccountCredentials() gcp_credentials.parse_native_representation(creds_dict) # Pass credentials to the BigQuery destination pipeline = dlt.pipeline(destination=bigquery(credentials=gcp_credentials)) pipeline.run([{"key1": "value1"}], table_name="temp")

这里的关键调用是parse_native_representation：它把 GCP service account 凭据的原生表示（通常是 JSON 字符串或字典）解析为 spec 的字段。该方法定义在 dlt/common/configuration/specs/base_configuration.py，默认抛出NotImplementedError，由各具体凭据类（如GcpServiceAccountCredentials）覆写实现实际解析逻辑；from_init_value（L325-L341）则会在内部调用_apply_init_value：字典走update，其余值走parse_native_representation，解析成功后自动标记为已解析（resolved）。

Google Sheets source 完整示例

下面的示例演示了一个读取 Google Sheets 指定 tab 的google_sheetssource 函数：

@dlt.source def google_sheets( spreadsheet_id=dlt.config.value, tab_names=dlt.config.value, credentials=dlt.secrets.value, only_strings=False ): # Handle credentials as either dictionary or string if isinstance(credentials, str): credentials = json.loads(credentials) # Handle tabs as either list or comma-separated string if isinstance(tab_names, str): tab_names = tab_names.split(",") sheets = build('sheets', 'v4', credentials=ServiceAccountCredentials.from_service_account_info(credentials)) # ty: ignore tabs = [] for tab_name in tab_names: data = _get_sheet(sheets, spreadsheet_id, tab_name) # ty: ignore[unresolved-reference] tabs.append(dlt.resource(data, name=tab_name)) return tabs

@dlt.source装饰器让函数所有参数都可配置。特殊默认值dlt.secrets.value与dlt.config.value告诉dlt这些参数是必填的，要么显式传入、要么存在于配置中；其中dlt.secrets.value额外将参数标记为密钥。

本示例中各参数的角色：

• spreadsheet_id—必填 config参数；
• tab_names—必填 config参数；
• credentials—必填 secret参数（Google Sheets 凭据，字典形式）；
• only_strings—可选 config参数，带默认值。

提示：dlt.resource的工作方式相同，因此独立 resource（未作为 source 内部函数定义）遵循同样的注入规则。

六、编写自定义 Spec：完全掌控注入行为

自定义规范（custom specifications）让你完全掌控函数参数：

• 控制哪些值应被注入、其类型与默认值；
• 指定可选（optional）与 final 字段；
• 构建层级配置（spec 内嵌 spec）；
• 提供自定义的on_partial（在因缺少配置键而失败前调用）或on_resolved处理器；
• 提供自定义的原生值解析器（native value parsers）；
• 提供自定义的默认凭据逻辑；
• 利用 Python dataclass 功能；
• 利用 Pythondict功能（spec 实例可从字典创建、也可序列化为字典）。

事实上，dlt会为每个被装饰的函数合成一个唯一的 spec。以google_sheets为例，会生成如下类：

from dlt.common.configuration import configspec, with_config, BaseConfiguration @configspec class GoogleSheetsConfiguration(BaseConfiguration): tab_names: list[str] = None # mandatory credentials: GcpServiceAccountCredentials = None # mandatory secret only_strings: bool | None = False

合成的完整流程见 dlt/common/reflection/spec.py：spec_from_signature遍历函数签名，过滤出带默认值的合法类型参数，标记dlt.config.value/dlt.secrets.value，然后用type(name, (base,), new_fields)动态创建 spec 类，最后通过configspec()将其转换为带字典接口的 dataclass，并注册到函数所在模块。

所有 Spec 派生自 BaseConfiguration

BaseConfiguration（见 dlt/common/configuration/specs/base_configuration.py）是创建配置对象的基类，提供以下能力：

• 以原生形式解析和表示配置的方法：parse_native_representation与to_native_representation（L349-L371）；
• 访问与操作配置字段的方法；
• 在 dataclass 之上实现的字典兼容接口——实例可当作字典使用（__getitem__、__setitem__、__delitem__、__iter__、__len__、update，见 L452-L490），因此既可从字典创建、也可序列化为字典；
• 判断某个属性是否存在、字段是否有效，以及按 MRO 调用方法（call_method_in_mro，见 L499-L511）的辅助函数；
• 解析状态追踪：is_resolved()、is_partial()、resolve()（L406-L423），其中is_partial检查是否有必填字段缺失，resolve会调用on_resolved处理器并把实例标记为已解析。

configspec装饰器（L176-L299）会把任何被装饰类转换为可用作配置解析 spec 的 Python dataclass：所有字段必须有默认值（缺失的自动补None并告警），未注解的属性会抛出ConfigFieldMissingTypeHintException，不支持的注解类型抛出ConfigFieldTypeHintNotSupported，并自动生成__init__（除非类自定义了__init__）。更详细的说明可参阅该类 docstring。

所有凭据派生自 CredentialsConfiguration

CredentialsConfiguration是BaseConfiguration的子类，作为各类凭据的基类（见 dlt/common/configuration/specs/base_configuration.py）：

@configspec class CredentialsConfiguration(BaseConfiguration): """Base class for all credentials. Credentials are configurations that may be stored only by providers supporting secrets.""" __section__: ClassVar[str] = "credentials" def to_native_credentials(self) -> Any: return self.to_native_representation() def __str__(self) -> str: """Get string representation of credentials to be displayed, with all secret parts removed""" return super().__str__()

它定义了初始化凭据、转换为原生表示、生成字符串表示的方法，并在生成字符串表示时确保敏感信息被剔除（__str__去除所有 secret 部分）。其__section__固定为"credentials"，这也是 destination 配置中credentialssection 成为必选分组、查找时不会被消除的根源。所有内置凭据类型（连接字符串、AWS/GCP/Azure 凭据等）都以此类为基类，并覆写parse_native_representation实现各自的原生表示解析。更详细说明可参阅类 docstring。

七、总结与下一步

本文覆盖了 dlt 配置系统的完整编程接口：

• 配置存放位置、provider 优先级与布局查找顺序，见配置总览；
• 内置与自定义凭据类型（连接字符串、AWS/Azure/GCP 凭据、Union多类型鉴权），见复杂凭据类型；
• Vault（Google Secrets Manager、Airflow Variables 等）集成，见Vault 指南；
• 实际为 source/destination 添加凭据的完整示例，可参考添加凭据实战。

【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy 🛠️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt