dlt 配置注入与密钥管理实战在代码中访问与编写配置、自定义 Spec 与分层布局【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dltdltdata 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从签名动态合成只有「带默认值且类型合法」的参数才会进入 specspec 的类名由函数限定名qualname推断并注册到函数所在模块同名函数在模块内唯一缓存。注入规则Injection rules规则 1显式传入的参数永远不会被注入。这让注入机制变成可选项。以 Pipedrive source 为例import os from typing import Iterator from dlt.extract import DltResource dlt.source(namepipedrive) 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_keymy_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即带默认值的参数才进入合成 specdlt/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_defaultsTrue是with_config的默认行为即带默认值的参数默认会纳入合成 specdlt/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 添加类型注解强烈建议为函数签名添加类型注解成本极低、收益显著你不会在代码中收到非法数据类型dlt自动解析并转换类型无需手动解析dlt可以自动为 source 生成示例 config 和 secret 文件可以请求内置与自定义凭据连接字符串、AWS/GCP/Azure 凭据可通过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 ): ...收益你会得到类型正确的tab_names字符串列表你会得到配置正确的 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 Ill let people set nesting level, name convention, normalizer, etc. here |-load |-normalize在 TOML 文件中该结构表现为带点号的嵌套 section对于环境变量等 provider布局用双下划线展平例如PIPELINE_NAME__SOURCES__MODULE_NAME__FUNCTION_NAME__OPTION。pipeline 名称优先作为顶层 sectiondlt会先带 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(namemy_db, sectionmy_db_module)为同一模块的多个实例建立不同配置 section参见 docs/website/docs/general-usage/source.md。从源码看紧凑布局的支持定义在 dlt/common/configuration/resolve.pyCOMPACT_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为例为sources.notion.notion_databases.api_keysources.notion.api_keysources.api_keyapi_key当 section 与 name 不同例如clone(namemy_db, sectionmy_db_module)时sources.my_db_module.my_db.api_key完整路径sources.my_db_module.api_keysources.my_db.api_key紧凑路径sources.api_keyapi_keydestination 凭据类似但credentialssection 被视为必选分组、不会被消除destination.postgres.credentials.passworddestination.credentials.passwordcredentials.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(destinationbigquery)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.pydlt.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 destinationimport 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(destinationbigquery(credentialsgcp_credentials)) pipeline.run([{key1: value1}], table_nametemp)这里的关键调用是parse_native_representation它把 GCP service account 凭据的原生表示通常是 JSON 字符串或字典解析为 spec 的字段。该方法定义在 dlt/common/configuration/specs/base_configuration.py默认抛出NotImplementedError由各具体凭据类如GcpServiceAccountCredentials覆写实现实际解析逻辑from_init_valueL325-L341则会在内部调用_apply_init_value字典走update其余值走parse_native_representation解析成功后自动标记为已解析resolved。Google Sheets source 完整示例下面的示例演示了一个读取 Google Sheets 指定 tab 的google_sheetssource 函数dlt.source def google_sheets( spreadsheet_iddlt.config.value, tab_namesdlt.config.value, credentialsdlt.secrets.value, only_stringsFalse ): # 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, credentialsServiceAccountCredentials.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, nametab_name)) return tabsdlt.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.pyspec_from_signature遍历函数签名过滤出带默认值的合法类型参数标记dlt.config.value/dlt.secrets.value然后用type(name, (base,), new_fields)动态创建 spec 类最后通过configspec()将其转换为带字典接口的 dataclass并注册到函数所在模块。所有 Spec 派生自 BaseConfigurationBaseConfiguration见 dlt/common/configuration/specs/base_configuration.py是创建配置对象的基类提供以下能力以原生形式解析和表示配置的方法parse_native_representation与to_native_representationL349-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。所有凭据派生自 CredentialsConfigurationCredentialsConfiguration是BaseConfiguration的子类作为各类凭据的基类见 dlt/common/configuration/specs/base_configuration.pyconfigspec 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 配置系统的完整编程接口能力关键 API / 机制源码位置自动注入dlt.source/dlt.resource/dlt.destination合成 specdlt/common/configuration/inject.py签名→spec 合成spec_from_signatureconfigspecdlt/common/reflection/spec.py配置解析与回退resolve_configuration、紧凑布局dlt/common/configuration/resolve.py字典式读写dlt.config/dlt.secrets含get类型转换、写入模拟 TOML providerdlt/common/configuration/accessors.py配置基类BaseConfiguration字典接口、on_resolved、原生表示dlt/common/configuration/specs/base_configuration.py凭据基类CredentialsConfiguration密钥安全输出、credentialssectiondlt/common/configuration/specs/base_configuration.py配置存放位置、provider 优先级与布局查找顺序见配置总览内置与自定义凭据类型连接字符串、AWS/Azure/GCP 凭据、Union多类型鉴权见复杂凭据类型VaultGoogle 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
