Skip to content

Commit

Permalink
[Tidy first] Task config type (#9874)
Browse files Browse the repository at this point in the history
  • Loading branch information
gshank authored Apr 10, 2024
1 parent d03292e commit a1f0057
Show file tree
Hide file tree
Showing 11 changed files with 61 additions and 65 deletions.
3 changes: 1 addition & 2 deletions core/dbt/cli/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -399,7 +399,6 @@ def debug(ctx, **kwargs):

task = DebugTask(
ctx.obj["flags"],
None,
)

results = task.run()
Expand Down Expand Up @@ -464,7 +463,7 @@ def init(ctx, **kwargs):
"""Initialize a new dbt project."""
from dbt.task.init import InitTask

task = InitTask(ctx.obj["flags"], None)
task = InitTask(ctx.obj["flags"])

results = task.run()
success = task.interpret_results(results)
Expand Down
15 changes: 6 additions & 9 deletions core/dbt/deps/resolver.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
from dataclasses import dataclass, field
from typing import Dict, List, NoReturn, Union, Type, Iterator, Set, Any
from typing import Dict, List, NoReturn, Type, Iterator, Set, Any

from dbt.exceptions import (
DuplicateDependencyToRootError,
Expand All @@ -17,14 +17,13 @@
from dbt.deps.registry import RegistryUnpinnedPackage

from dbt.contracts.project import (
PackageSpec,
LocalPackage,
TarballPackage,
GitPackage,
RegistryPackage,
)

PackageContract = Union[LocalPackage, TarballPackage, GitPackage, RegistryPackage]


@dataclass
class PackageListing:
Expand Down Expand Up @@ -68,7 +67,7 @@ def incorporate(self, package: UnpinnedPackage):
else:
self.packages[key] = package

def update_from(self, src: List[PackageContract]) -> None:
def update_from(self, src: List[PackageSpec]) -> None:
pkg: UnpinnedPackage
for contract in src:
if isinstance(contract, LocalPackage):
Expand All @@ -84,9 +83,7 @@ def update_from(self, src: List[PackageContract]) -> None:
self.incorporate(pkg)

@classmethod
def from_contracts(
cls: Type["PackageListing"], src: List[PackageContract]
) -> "PackageListing":
def from_contracts(cls: Type["PackageListing"], src: List[PackageSpec]) -> "PackageListing":
self = cls({})
self.update_from(src)
return self
Expand Down Expand Up @@ -114,7 +111,7 @@ def _check_for_duplicate_project_names(


def resolve_packages(
packages: List[PackageContract],
packages: List[PackageSpec],
project: Project,
cli_vars: Dict[str, Any],
) -> List[PinnedPackage]:
Expand All @@ -137,7 +134,7 @@ def resolve_packages(
return resolved


def resolve_lock_packages(packages: List[PackageContract]) -> List[PinnedPackage]:
def resolve_lock_packages(packages: List[PackageSpec]) -> List[PinnedPackage]:
lock_packages = PackageListing.from_contracts(packages)
final = PackageListing()

Expand Down
57 changes: 22 additions & 35 deletions core/dbt/task/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,14 +6,14 @@
from contextlib import nullcontext
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Type, Union, Set
from typing import Any, Dict, List, Optional, Set

from dbt.compilation import Compiler
import dbt_common.exceptions.base
import dbt.exceptions
from dbt import tracking
from dbt.cli.flags import Flags
from dbt.config import RuntimeConfig, Project
from dbt.config import RuntimeConfig
from dbt.config.profile import read_profile
from dbt.constants import DBT_PROJECT_FILE_NAME
from dbt.contracts.graph.manifest import Manifest
Expand Down Expand Up @@ -50,12 +50,6 @@
from dbt.task.printer import print_run_result_error


class NoneConfig:
@classmethod
def from_args(cls, args: Flags):
return None


def read_profiles(profiles_dir=None):
"""This is only used for some error handling"""
if profiles_dir is None:
Expand All @@ -72,12 +66,8 @@ def read_profiles(profiles_dir=None):


class BaseTask(metaclass=ABCMeta):
ConfigType: Union[Type[NoneConfig], Type[Project]] = NoneConfig

def __init__(self, args: Flags, config, project=None) -> None:
def __init__(self, args: Flags) -> None:
self.args = args
self.config = config
self.project = config if isinstance(config, Project) else project

@classmethod
def pre_init_hook(cls, args: Flags):
Expand All @@ -94,23 +84,6 @@ def set_log_format(cls):
else:
log_manager.format_text()

@classmethod
def from_args(cls, args, *pargs, **kwargs):
try:
# This is usually RuntimeConfig
config = cls.ConfigType.from_args(args)
except dbt.exceptions.DbtProjectError as exc:
fire_event(LogDbtProjectError(exc=str(exc)))

tracking.track_invalid_invocation(args=args, result_type=exc.result_type)
raise dbt_common.exceptions.DbtRuntimeError("Could not run dbt") from exc
except dbt.exceptions.DbtProfileError as exc:
all_profile_names = list(read_profiles(get_flags().PROFILES_DIR).keys())
fire_event(LogDbtProfileError(exc=str(exc), profiles=all_profile_names))
tracking.track_invalid_invocation(args=args, result_type=exc.result_type)
raise dbt_common.exceptions.DbtRuntimeError("Could not run dbt") from exc
return cls(args, config, *pargs, **kwargs)

@abstractmethod
def run(self):
raise dbt_common.exceptions.base.NotImplementedError("Not Implemented")
Expand Down Expand Up @@ -154,10 +127,11 @@ def move_to_nearest_project_dir(project_dir: Optional[str]) -> Path:
# produce the same behavior. currently this class only contains manifest compilation,
# holding a manifest, and moving direcories.
class ConfiguredTask(BaseTask):
ConfigType = RuntimeConfig

def __init__(self, args: Flags, config, manifest: Optional[Manifest] = None) -> None:
super().__init__(args, config)
def __init__(
self, args: Flags, config: RuntimeConfig, manifest: Optional[Manifest] = None
) -> None:
super().__init__(args)
self.config = config
self.graph: Optional[Graph] = None
self.manifest = manifest
self.compiler = Compiler(self.config)
Expand All @@ -177,7 +151,20 @@ def compile_manifest(self):
@classmethod
def from_args(cls, args: Flags, *pargs, **kwargs):
move_to_nearest_project_dir(args.project_dir)
return super().from_args(args, *pargs, **kwargs)
try:
# This is usually RuntimeConfig
config = RuntimeConfig.from_args(args)
except dbt.exceptions.DbtProjectError as exc:
fire_event(LogDbtProjectError(exc=str(exc)))

tracking.track_invalid_invocation(args=args, result_type=exc.result_type)
raise dbt_common.exceptions.DbtRuntimeError("Could not run dbt") from exc
except dbt.exceptions.DbtProfileError as exc:
all_profile_names = list(read_profiles(get_flags().PROFILES_DIR).keys())
fire_event(LogDbtProfileError(exc=str(exc), profiles=all_profile_names))
tracking.track_invalid_invocation(args=args, result_type=exc.result_type)
raise dbt_common.exceptions.DbtRuntimeError("Could not run dbt") from exc
return cls(args, config, *pargs, **kwargs)


class ExecutionContext:
Expand Down
4 changes: 3 additions & 1 deletion core/dbt/task/build.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@
from dbt.artifacts.schemas.results import NodeStatus, RunStatus
from dbt.artifacts.schemas.run import RunResult
from dbt.cli.flags import Flags
from dbt.config.runtime import RuntimeConfig
from dbt.contracts.graph.manifest import Manifest
from dbt.graph import ResourceTypeSelector, GraphQueue, Graph
from dbt.node_types import NodeType
from dbt.task.test import TestSelector
Expand Down Expand Up @@ -75,7 +77,7 @@ class BuildTask(RunTask):
}
ALL_RESOURCE_VALUES = frozenset({x for x in RUNNER_MAP.keys()})

def __init__(self, args: Flags, config, manifest) -> None:
def __init__(self, args: Flags, config: RuntimeConfig, manifest: Manifest) -> None:
super().__init__(args, config, manifest)
self.selected_unit_tests: Set = set()
self.model_to_unit_test_map: Dict[str, List] = {}
Expand Down
7 changes: 7 additions & 0 deletions core/dbt/task/clean.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,13 +9,20 @@
FinishedCleanPaths,
)
from dbt_common.exceptions import DbtRuntimeError
from dbt.cli.flags import Flags
from dbt.config.project import Project
from dbt.task.base import (
BaseTask,
move_to_nearest_project_dir,
)


class CleanTask(BaseTask):
def __init__(self, args: Flags, config: Project):
super().__init__(args)
self.config = config
self.project = config

def run(self):
"""
This function takes all the paths in the target file
Expand Down
11 changes: 2 additions & 9 deletions core/dbt/task/debug.py
Original file line number Diff line number Diff line change
Expand Up @@ -78,8 +78,8 @@ class DebugRunStatus(Flag):


class DebugTask(BaseTask):
def __init__(self, args: Flags, config) -> None:
super().__init__(args, config)
def __init__(self, args: Flags) -> None:
super().__init__(args)
self.profiles_dir = args.PROFILES_DIR
self.profile_path = os.path.join(self.profiles_dir, "profiles.yml")
try:
Expand All @@ -98,13 +98,6 @@ def __init__(self, args: Flags, config) -> None:
self.profile: Optional[Profile] = None
self.raw_profile_data: Optional[Dict[str, Any]] = None
self.profile_name: Optional[str] = None
self.project: Optional[Project] = None

@property
def project_profile(self):
if self.project is None:
return None
return self.project.profile_name

def run(self) -> bool:
# WARN: this is a legacy workflow that is not compatible with other runtime flags
Expand Down
7 changes: 4 additions & 3 deletions core/dbt/task/deps.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
from dbt.deps.base import downloads_directory
from dbt.deps.resolver import resolve_lock_packages, resolve_packages
from dbt.deps.registry import RegistryPinnedPackage
from dbt.contracts.project import Package
from dbt.contracts.project import PackageSpec


from dbt_common.events.functions import fire_event
Expand Down Expand Up @@ -44,7 +44,7 @@ def increase_indent(self, flow=False, indentless=False):
return super(dbtPackageDumper, self).increase_indent(flow, False)


def _create_sha1_hash(packages: List[Package]) -> str:
def _create_sha1_hash(packages: List[PackageSpec]) -> str:
"""Create a SHA1 hash of the packages list,
this is used to determine if the packages for current execution matches
the previous lock.
Expand Down Expand Up @@ -94,14 +94,15 @@ def _create_packages_yml_entry(package: str, version: Optional[str], source: str

class DepsTask(BaseTask):
def __init__(self, args: Any, project: Project) -> None:
super().__init__(args=args)
# N.B. This is a temporary fix for a bug when using relative paths via
# --project-dir with deps. A larger overhaul of our path handling methods
# is needed to fix this the "right" way.
# See GH-7615
project.project_root = str(Path(project.project_root).resolve())
self.project = project

move_to_nearest_project_dir(project.project_root)
super().__init__(args=args, config=None, project=project)
self.cli_vars = args.vars

def track_package_install(
Expand Down
4 changes: 3 additions & 1 deletion core/dbt/task/list.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@
UnitTestDefinition,
)
from dbt.cli.flags import Flags
from dbt.config.runtime import RuntimeConfig
from dbt.contracts.graph.manifest import Manifest
from dbt.flags import get_flags
from dbt.graph import ResourceTypeSelector
from dbt.task.base import resource_types_from_args
Expand Down Expand Up @@ -55,7 +57,7 @@ class ListTask(GraphRunnableTask):
)
)

def __init__(self, args: Flags, config, manifest) -> None:
def __init__(self, args: Flags, config: RuntimeConfig, manifest: Manifest) -> None:
super().__init__(args, config, manifest)
if self.args.models:
if self.args.select:
Expand Down
2 changes: 1 addition & 1 deletion core/dbt/task/retry.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@


class RetryTask(ConfiguredTask):
def __init__(self, args: Flags, config) -> None:
def __init__(self, args: Flags, config: RuntimeConfig) -> None:
# load previous run results
state_path = args.state or config.target_path
self.previous_results = load_result_state(
Expand Down
4 changes: 3 additions & 1 deletion core/dbt/task/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,10 @@
from dbt.adapters.base import BaseRelation
from dbt.cli.flags import Flags
from dbt.clients.jinja import MacroGenerator
from dbt.config.runtime import RuntimeConfig
from dbt.context.providers import generate_runtime_model_context
from dbt.contracts.graph.nodes import HookNode, ResultNode
from dbt.contracts.graph.manifest import Manifest
from dbt.artifacts.schemas.results import NodeStatus, RunStatus, RunningStatus, BaseResult
from dbt.artifacts.schemas.run import RunResult
from dbt.artifacts.resources import Hook
Expand Down Expand Up @@ -306,7 +308,7 @@ def execute(self, model, manifest):


class RunTask(CompileTask):
def __init__(self, args: Flags, config, manifest) -> None:
def __init__(self, args: Flags, config: RuntimeConfig, manifest: Manifest) -> None:
super().__init__(args, config, manifest)
self.ran_hooks: List[HookNode] = []
self._total_executed = 0
Expand Down
12 changes: 9 additions & 3 deletions core/dbt/task/runnable.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
from dbt.adapters.base import BaseRelation
from dbt.adapters.factory import get_adapter
from dbt.cli.flags import Flags
from dbt.config.runtime import RuntimeConfig
from dbt.contracts.graph.manifest import Manifest
from dbt.contracts.graph.nodes import ResultNode
from dbt.artifacts.schemas.results import NodeStatus, RunningStatus, RunStatus, BaseResult
Expand Down Expand Up @@ -66,8 +67,9 @@
class GraphRunnableTask(ConfiguredTask):
MARK_DEPENDENT_ERRORS_STATUSES = [NodeStatus.Error]

def __init__(self, args: Flags, config, manifest) -> None:
def __init__(self, args: Flags, config: RuntimeConfig, manifest: Manifest) -> None:
super().__init__(args, config, manifest)
self.config = config
self._flattened_nodes: Optional[List[ResultNode]] = None
self._raise_next_tick: Optional[DbtRuntimeError] = None
self._skipped_children: Dict[str, Optional[RunResult]] = {}
Expand Down Expand Up @@ -124,7 +126,9 @@ def get_selection_spec(self) -> SelectionSpec:
# This is what's used with no default selector and no selection
# use --select and --exclude args
spec = parse_difference(self.selection_arg, self.exclusion_arg, indirect_selection)
return spec
# mypy complains because the return values of get_selector and parse_difference
# are different
return spec # type: ignore

@abstractmethod
def get_node_selector(self) -> NodeSelector:
Expand Down Expand Up @@ -624,7 +628,9 @@ def create_schema(relation: BaseRelation) -> None:
list_futures = []
create_futures = []

with dbt_common.utils.executor(self.config) as tpe:
# TODO: following has a mypy issue because profile and project config
# defines threads as int and HasThreadingConfig defines it as Optional[int]
with dbt_common.utils.executor(self.config) as tpe: # type: ignore
for req in required_databases:
if req.database is None:
name = "list_schemas"
Expand Down

0 comments on commit a1f0057

Please sign in to comment.