ConfigurableResourceFactory Should Be Public

[dwallace0723]

Currently, the ConfigurableResourceFactory class is not imported in the main dagster module (ref). I am taking this to imply that this class is not meant for public use (Although, the docstrings seem to suggest otherwise (ref)). I'd like to propose that ConfigurableResourceFactory should be imported into the main dagster module and should be available for public use.

My argument centers around the fact that ConfigurableResourceFactory enables a really elegant pattern of separating config/validation from behavior. ConfigurableResourceFactory demands the user override the create_resource abstractmethod (ref) to return the object responsible for providing behavior to Ops/Assets. Thus, ConfigurableResourceFactory simply acts as a config/validation layer responsible for parsing, validating, and forwarding config to the object that ultimately implements the behavior the downstream users will rely on.

As an example, this enables the following pattern:

class Meltano(CliResource):
    ...

    def elt(self, extractor: str, loader: str, timeout: Optional[float] = None, **opts) -> Popen:
        """Run the meltano elt command via the CLI."""
        check.str_param(extractor, "extractor")
        check.str_param(loader, "loader")
        return self.log(self.run("elt", extractor, loader, **opts), timeout)


class MeltanoResource(ConfigurableResourceFactory[Meltano]):
    options: CliOptions = {"log_level": "info"}

    def create_resource(self, context: InitResourceContext) -> Meltano:
        return Meltano(options=self.options, logger=context.log)

@op(description="Run the meltano 'elt' command.", ...)
def meltano_elt(context: OpExecutionContext, meltano: ResourceParam[Meltano]):
    opts = {k: v for k, v in context.op_config.items() if k not in ["extractor", "loader", "timeout"]}
    meltano.elt(
        extractor=context.op_config["extractor"],
        loader=context.op_config["loader"],
        timeout=context.op_config["timeout"],
        **opts,
    )

Whereas if you were to use the ConfigurableResource class the way the docs suggest it be used (ref), you would end up with an implementation looking more like the following:

class Meltano(CliResource):
    ...

    def elt(self, extractor: str, loader: str, timeout: Optional[float] = None, **opts) -> Popen:
        """Run the meltano elt command via the CLI."""
        check.str_param(extractor, "extractor")
        check.str_param(loader, "loader")
        return self.log(self.run("elt", extractor, loader, **opts), timeout)


class MeltanoResource(ConfigurableResource):
    options: CliOptions = {"log_level": "info"}

    def get_client(self) -> Meltano:
         return Meltano(self.options)

@op(description="Run the meltano 'elt' command.", ...)
def meltano_elt(context: OpExecutionContext, meltano: MeltanoResource):
    opts = {k: v for k, v in context.op_config.items() if k not in ["extractor", "loader", "timeout"]}
    meltano.get_client().elt(
        extractor=context.op_config["extractor"],
        loader=context.op_config["loader"],
        timeout=context.op_config["timeout"],
        **opts,
    )

I'd argue that the second implementation adds an unnecessary layer of abstraction, and implements low-value behavior in a proxy class that shouldn't require any behavior in the first place.

[schrockn]

Hey @dwallace0723 thanks for surfacing! I believe your first code example will work if you use change it to use ConfigurableResource instead of ConfigurableResourceFactory.

class Meltano(CliResource):
    ...

    def elt(self, extractor: str, loader: str, timeout: Optional[float] = None, **opts) -> Popen:
        """Run the meltano elt command via the CLI."""
        check.str_param(extractor, "extractor")
        check.str_param(loader, "loader")
        return self.log(self.run("elt", extractor, loader, **opts), timeout)


class MeltanoResource(ConfigurableResource[Meltano]): # change is here
    options: CliOptions = {"log_level": "info"}

    def create_resource(self, context: InitResourceContext) -> Meltano:
        return Meltano(options=self.options, logger=context.log)

@op(description="Run the meltano 'elt' command.", ...)
def meltano_elt(context: OpExecutionContext, meltano: ResourceParam[Meltano]):
    opts = {k: v for k, v in context.op_config.items() if k not in ["extractor", "loader", "timeout"]}
    meltano.elt(
        extractor=context.op_config["extractor"],
        loader=context.op_config["loader"],
        timeout=context.op_config["timeout"],
        **opts,
    )

The naming is slightly confusing since it is "configurable resource" that creates a "resource", but the functionality is identical. Using the "factory" name is more clear when it is used in this fashion, but there is tradeoff in having to expand the surface area here.

I think the sin here is recommending the inner client pattern in the docs you reference.

[dwallace0723]

Okay, yeah. So maybe a docstring update is in order?

[schrockn]

#17613