Dagster中AWS Secrets Manager资源创建与访问报错排查

在使用Dagster库将AWS Secrets Manager密钥作为资源创建并在assets中访问时遇到错误，文件结构如下：

-proj
    -assets
        __init__.py
        asset_script.py
    -resources 
        __init__.py
        secrets_manager.py
    __init__.py

secrets_manager.py代码：

from dagster import ConfigurableResource
from dagster_aws.secretsmanager import SecretsManagerSecretsResource


class SecretsConnection(ConfigurableResource):
    sm = SecretsManagerSecretsResource(
        region_name="us-east-1",
        add_to_environment=False,
    )

    def get_secrets(self):
        return self.sm.fetch_secrets()

顶层proj/__init__.py代码：

from .assets import (
    asset_script
)
from .resources import (
    secrets_manager
)

assets = load_assets_from_package_module(
    package_module=asset_script,
    group_name="group_name",
)

defs = Definitions(
    assets=assets,
    resources={
        "secrets_manager": secrets_manager.SecretsConnection(),
    },
)

assets中访问资源的代码：

@asset(required_resource_keys={"secrets_manager"})
def fn_name(context: AssetExecutionContext):
    secrets = context.resources.secrets_manager.get_secrets()

运行时触发核心错误：

TypeError: __new__() missing 1 required positional argument: 'fields'

问题出在SecretsConnection类中直接实例化SecretsManagerSecretsResource作为类属性。由于SecretsManagerSecretsResource本身是ConfigurableResource子类，Pydantic（Dagster Pythonic配置的底层依赖）尝试深度复制该实例时，会触发内部初始化错误——这类资源需要通过配置系统完成初始化，而非直接作为类属性实例化。

提供两种可行修复方式：

方式一：直接使用原生资源类 #

无需自定义SecretsConnection，直接在Definitions中注册SecretsManagerSecretsResource：

修改顶层proj/__init__.py：

from dagster_aws.secretsmanager import SecretsManagerSecretsResource
from .assets import asset_script
from dagster import load_assets_from_package_module, Definitions

assets = load_assets_from_package_module(
    package_module=asset_script,
    group_name="group_name",
)

defs = Definitions(
    assets=assets,
    resources={
        "secrets_manager": SecretsManagerSecretsResource(
            region_name="us-east-1",
            add_to_environment=False,
        ),
    },
)

修改asset代码，直接调用原生方法：

@asset(required_resource_keys={"secrets_manager"})
def fn_name(context: AssetExecutionContext):
    secrets = context.resources.secrets_manager.fetch_secrets()

方式二：正确封装自定义资源 #

若需保留自定义类，需将SecretsManagerSecretsResource定义为类的配置字段，而非直接实例化：

修改secrets_manager.py：

from dagster import ConfigurableResource
from dagster_aws.secretsmanager import SecretsManagerSecretsResource


class SecretsConnection(ConfigurableResource):
    sm: SecretsManagerSecretsResource

    def get_secrets(self):
        return self.sm.fetch_secrets()

修改顶层proj/__init__.py，初始化时传入资源实例：

from .assets import asset_script
from .resources.secrets_manager import SecretsConnection
from dagster_aws.secretsmanager import SecretsManagerSecretsResource
from dagster import load_assets_from_package_module, Definitions

assets = load_assets_from_package_module(
    package_module=asset_script,
    group_name="group_name",
)

defs = Definitions(
    assets=assets,
    resources={
        "secrets_manager": SecretsConnection(
            sm=SecretsManagerSecretsResource(
                region_name="us-east-1",
                add_to_environment=False,
            )
        ),
    },
)

asset代码保持原逻辑即可。

修复后重新运行Dagster，资源初始化与资产访问可正常执行，不再触发TypeError。

内容的提问来源于stack exchange，提问作者MAA