WIP

Author: z3z1ma

src/dbt_core_interface/project.py

@@ -237,73 +237,6 @@ class WrappedDbtAdapterCompilationResponse:
     injected_code: t.Optional[str] = None
 
 
-class DbtTaskConfiguration:
-    """A container for task configuration with sane defaults.
-
-    Users should enforce an interface for their tasks via a factory method that returns an instance of this class.
-    """
-
-    def __init__(self, profile: str, target: str, **kwargs: t.Any) -> None:
-        """Initialize the task configuration."""
-        self.profile: str = profile
-        self.target: str = target
-        self.kwargs: t.Dict[str, t.Any] = kwargs or {}
-        self.threads: int = kwargs.get("threads", 1)
-        self.single_threaded: bool = kwargs.get("single_threaded", self.threads == 1)
-        self.state_id: t.Optional[str] = kwargs.get("state_id")
-        self.version_check: bool = kwargs.get("version_check", False)
-        self.resource_types: t.Optional[t.List[str]] = kwargs.get("resource_types")
-        self.models: t.Union[None, str, t.List[str]] = kwargs.get("models")
-        self.select: t.Union[None, str, t.List[str]] = kwargs.get("select")
-        self.exclude: t.Union[None, str, t.List[str]] = kwargs.get("exclude")
-        self.selector_name: t.Optional[str] = kwargs.get("selector_name")
-        self.state: t.Optional[str] = kwargs.get("state")
-        self.defer: bool = kwargs.get("defer", False)
-        self.fail_fast: bool = kwargs.get("fail_fast", False)
-        self.full_refresh: bool = kwargs.get("full_refresh", False)
-        self.store_failures: bool = kwargs.get("store_failures", False)
-        self.indirect_selection: bool = kwargs.get("indirect_selection", "eager")
-        self.data: bool = kwargs.get("data", False)
-        self.schema: bool = kwargs.get("schema", False)
-        self.show: bool = kwargs.get("show", False)
-        self.output: str = kwargs.get("output", "name")
-        self.output_keys: t.Union[None, str, t.List[str]] = kwargs.get("output_keys")
-        self.macro: t.Optional[str] = kwargs.get("macro")
-        self.args: str = kwargs.get("args", "{}")
-        self.quiet: bool = kwargs.get("quiet", True)
-
-    def __getattribute__(self, __name: str) -> t.Any:
-        """Force all attribute access to be lower case."""
-        return object.__getattribute__(self, __name.lower())
-
-    def __getattr__(self, name: str) -> t.Any:
-        """Get an attribute from the kwargs if it does not exist on the class.
-
-        This is useful for passing through arbitrary arguments to dbt while still
-        being able to manage some semblance of a sane interface with defaults.
-        """
-        return self.kwargs.get(name)
-
-    @classmethod
-    def from_runtime_config(cls, config: RuntimeConfig, **kwargs: t.Any) -> DbtTaskConfiguration:
-        """Create a task configuration container from a DbtProject's runtime config.
-
-        This is a good example of where static typing is not necessary. Developers can just
-        pass in whatever they want and it will be passed through to the task configuration container.
-        Users of the library are free to pass in any mapping derived from their own implementation for
-        their own custom task.
-        """
-        threads = kwargs.pop("threads", config.threads)
-        kwargs.pop("single_threaded", None)  # This is a derived property
-        return cls(
-            config.profile_name,
-            config.target_name,
-            threads=threads,
-            single_threaded=threads == 1,
-            **kwargs,
-        )
-
-
 @t.final
 class DbtProject:
     """Container for a dbt project.
@@ -814,13 +747,6 @@ def macro_parser(self) -> SqlMacroParser:
             self._macro_parser = SqlMacroParser(self.config, self.manifest)
         return self._macro_parser
 
-    def get_task_config(self, **kwargs) -> DbtTaskConfiguration:
-        """Get a dbt-core task configuration."""
-        threads = kwargs.pop("threads", self.config.threads)
-        return DbtTaskConfiguration.from_runtime_config(
-            config=self.config, threads=threads, **kwargs
-        )
-
 
 class DbtProjectContainer:
     """Manages multiple DbtProjects.