Rfix/pipeline v2 extract comments

Makefile

@@ -13,15 +13,11 @@ VERSION := ${AUTV}${VERSION_SUFFIX}
 VERSION_MM := ${AUTVMINMAJ}${VERSION_SUFFIX}
 
 
-# dbt runner version info
-DBT_AUTV=$(shell python3 -c "from dlt.dbt_runner._version import __version__;print(__version__)")
-DBT_AUTVMINMAJ=$(shell python3 -c "from dlt.dbt_runner._version import __version__;print('.'.join(__version__.split('.')[:-1]))")
-
 DBT_NAME   := scalevector/dlt-dbt-runner
 DBT_IMG    := ${DBT_NAME}:${TAG}
 DBT_LATEST := ${DBT_NAME}:latest${VERSION_SUFFIX}
-DBT_VERSION := ${DBT_AUTV}${VERSION_SUFFIX}
-DBT_VERSION_MM := ${DBT_AUTVMINMAJ}${VERSION_SUFFIX}
+DBT_VERSION := ${AUTV}${VERSION_SUFFIX}
+DBT_VERSION_MM := ${AUTVMINMAJ}${VERSION_SUFFIX}
 
 install-poetry:
 ifneq ($(VIRTUAL_ENV),)
@@ -38,7 +34,7 @@ dev: has-poetry
 
 lint:
 	./check-package.sh
-	poetry run mypy --config-file mypy.ini dlt examples
+	poetry run mypy --config-file mypy.ini dlt
 	poetry run flake8 --max-line-length=200 examples dlt
 	poetry run flake8 --max-line-length=200 tests
 	# dlt/pipeline dlt/common/schema dlt/common/normalizers
@@ -50,7 +46,7 @@ lint-security:
 reset-test-storage:
 	-rm -r _storage
 	mkdir _storage
-	python3 test/tools/create_storages.py
+	python3 tests/tools/create_storages.py
 
 recreate-compiled-deps:
 	poetry export -f requirements.txt --output _gen_requirements.txt --without-hashes --extras gcp --extras redshift

dlt/__init__.py

@@ -1 +1 @@
-from dlt._version import common_version as __version__
+__version__ = "0.1.0"

dlt/cli/dlt.py

@@ -7,9 +7,8 @@
 from dlt.cli import TRunnerArgs
 from dlt.common.schema import Schema
 from dlt.common.typing import DictStrAny
-from dlt.common.utils import str2bool
 
-from dlt.pipeline import Pipeline, PostgresPipelineCredentials
+from dlt.pipeline import pipeline, restore
 
 
 def add_pool_cli_arguments(parser: argparse.ArgumentParser) -> None:
@@ -27,33 +26,35 @@ def add_pool_cli_arguments(parser: argparse.ArgumentParser) -> None:
 def main() -> None:
     parser = argparse.ArgumentParser(description="Runs various DLT modules", formatter_class=argparse.ArgumentDefaultsHelpFormatter)
     subparsers = parser.add_subparsers(dest="command")
-    normalize = subparsers.add_parser("normalize", help="Runs normalize")
-    add_pool_cli_arguments(normalize)
-    load = subparsers.add_parser("load", help="Runs loader")
-    add_pool_cli_arguments(load)
+
+    # normalize = subparsers.add_parser("normalize", help="Runs normalize")
+    # add_pool_cli_arguments(normalize)
+    # load = subparsers.add_parser("load", help="Runs loader")
+    # add_pool_cli_arguments(load)
+
     dbt = subparsers.add_parser("dbt", help="Executes dbt package")
     add_pool_cli_arguments(dbt)
     schema = subparsers.add_parser("schema", help="Shows, converts and upgrades schemas")
     schema.add_argument("file", help="Schema file name, in yaml or json format, will autodetect based on extension")
     schema.add_argument("--format", choices=["json", "yaml"], default="yaml", help="Display schema in this format")
     schema.add_argument("--remove-defaults", action="store_true", help="Does not show default hint values")
-    pipeline = subparsers.add_parser("pipeline", help="Operations on the pipelines")
-    pipeline.add_argument("name", help="Pipeline name")
-    pipeline.add_argument("workdir", help="Pipeline working directory")
-    pipeline.add_argument("operation", choices=["failed_loads"], default="failed_loads", help="Show failed loads for a pipeline")
+    pipe_cmd = subparsers.add_parser("pipeline", help="Operations on the pipelines")
+    pipe_cmd.add_argument("name", help="Pipeline name")
+    pipe_cmd.add_argument("operation", choices=["failed_loads", "drop"], default="failed_loads", help="Show failed loads for a pipeline")
+    pipe_cmd.add_argument("--workdir", help="Pipeline working directory", default=None)
 
     # TODO: consider using fire: https://github.com/google/python-fire
     # TODO: this also looks better https://click.palletsprojects.com/en/8.1.x/complex/#complex-guide
     args = parser.parse_args()
     run_f: Callable[[TRunnerArgs], None] = None
 
-    if args.command == "normalize":
-        from dlt.normalize.normalize import run_main as normalize_run
-        run_f = normalize_run
-    elif args.command == "load":
-        from dlt.load.load import run_main as loader_run
-        run_f = loader_run
-    elif args.command == "dbt":
+    # if args.command == "normalize":
+    #     from dlt.normalize.normalize import run_main as normalize_run
+    #     run_f = normalize_run
+    # elif args.command == "load":
+    #     from dlt.load.load import run_main as loader_run
+    #     run_f = loader_run
+    if args.command == "dbt":
         from dlt.dbt_runner.runner import run_main as dbt_run
         run_f = dbt_run
     elif args.command == "schema":
@@ -70,13 +71,21 @@ def main() -> None:
         print(schema_str)
         exit(0)
     elif args.command == "pipeline":
-        p = Pipeline(args.name)
-        p.restore_pipeline(PostgresPipelineCredentials("dummy"), args.workdir)
-        completed_loads = p.list_completed_loads()
-        for load_id in completed_loads:
-            print(f"Checking failed jobs in {load_id}")
-            for job, failed_message in p.list_failed_jobs(load_id):
-                print(f"JOB: {job}\nMSG: {failed_message}")
+        # from dlt.load import dummy
+
+        p = restore(pipeline_name=args.name, working_dir=args.workdir)
+        print(f"Found pipeline {p.pipeline_name} ({args.name}) in {p.working_dir} ({args.workdir}) with state {p._get_state()}")
+
+        if args.operation == "failed_loads":
+            completed_loads = p.list_completed_load_packages()
+            for load_id in completed_loads:
+                print(f"Checking failed jobs in load id '{load_id}'")
+                for job, failed_message in p.list_failed_jobs_in_package(load_id):
+                    print(f"JOB: {os.path.abspath(job)}\nMSG: {failed_message}")
+
+        if args.operation == "drop":
+            p.drop()
+
         exit(0)
     else:
         parser.print_help()

dlt/common/__init__.py

@@ -3,4 +3,3 @@
 from .pendulum import pendulum  # noqa: F401
 from .json import json  # noqa: F401, I251
 from .time import sleep   # noqa: F401
-from dlt._version import common_version as __version__

dlt/common/configuration/__init__.py

@@ -1,11 +1,6 @@
-from .run_configuration import RunConfiguration, BaseConfiguration, CredentialsConfiguration  # noqa: F401
-from .normalize_volume_configuration import NormalizeVolumeConfiguration, ProductionNormalizeVolumeConfiguration  # noqa: F401
-from .load_volume_configuration import LoadVolumeConfiguration, ProductionLoadVolumeConfiguration  # noqa: F401
-from .schema_volume_configuration import SchemaVolumeConfiguration, ProductionSchemaVolumeConfiguration  # noqa: F401
-from .pool_runner_configuration import PoolRunnerConfiguration, TPoolType  # noqa: F401
-from .gcp_client_credentials import GcpClientCredentials  # noqa: F401
-from .postgres_credentials import PostgresCredentials  # noqa: F401
-from .utils import make_configuration  # noqa: F401
+from .specs.base_configuration import configspec, is_valid_hint  # noqa: F401
+from .resolve import resolve_configuration, inject_namespace  # noqa: F401
+from .inject import with_config, last_config, get_fun_spec
 
 from .exceptions import (  # noqa: F401
-    ConfigEntryMissingException, ConfigEnvValueCannotBeCoercedException, ConfigIntegrityException, ConfigFileNotFoundException)
+    ConfigFieldMissingException, ConfigValueCannotBeCoercedException, ConfigIntegrityException, ConfigFileNotFoundException)

dlt/common/configuration/container.py

@@ -0,0 +1,66 @@
+from contextlib import contextmanager
+from typing import Dict, Iterator, Type, TypeVar
+
+from dlt.common.configuration.specs.base_configuration import ContainerInjectableContext
+from dlt.common.configuration.exceptions import ContainerInjectableContextMangled, ContextDefaultCannotBeCreated
+
+TConfiguration = TypeVar("TConfiguration", bound=ContainerInjectableContext)
+
+
+class Container:
+
+    _INSTANCE: "Container" = None
+
+    contexts: Dict[Type[ContainerInjectableContext], ContainerInjectableContext]
+
+    def __new__(cls: Type["Container"]) -> "Container":
+        if not cls._INSTANCE:
+            cls._INSTANCE = super().__new__(cls)
+            cls._INSTANCE.contexts = {}
+        return cls._INSTANCE
+
+    def __init__(self) -> None:
+        pass
+
+    def __getitem__(self, spec: Type[TConfiguration]) -> TConfiguration:
+        # return existing config object or create it from spec
+        if not issubclass(spec, ContainerInjectableContext):
+            raise KeyError(f"{spec.__name__} is not a context")
+
+        item = self.contexts.get(spec)
+        if item is None:
+            if spec.can_create_default:
+                item = spec()
+                self.contexts[spec] = item
+            else:
+                raise ContextDefaultCannotBeCreated(spec)
+
+        return item  # type: ignore
+
+    def __setitem__(self, spec: Type[TConfiguration], value: TConfiguration) -> None:
+        self.contexts[spec] = value
+
+    def __contains__(self, spec: Type[TConfiguration]) -> bool:
+        return spec in self.contexts
+
+    @contextmanager
+    def injectable_context(self, config: TConfiguration) -> Iterator[TConfiguration]:
+        spec = type(config)
+        previous_config: ContainerInjectableContext = None
+        if spec in self.contexts:
+            previous_config = self.contexts[spec]
+        # set new config and yield context
+        try:
+            self.contexts[spec] = config
+            yield config
+        finally:
+            # before setting the previous config for given spec, check if there was no overlapping modification
+            if self.contexts[spec] is config:
+                # config is injected for spec so restore previous
+                if previous_config is None:
+                    del self.contexts[spec]
+                else:
+                    self.contexts[spec] = previous_config
+            else:
+                # value was modified in the meantime and not restored
+                raise ContainerInjectableContextMangled(spec, self.contexts[spec], config)

dlt/common/configuration/exceptions.py

@@ -1,34 +1,59 @@
-from typing import Iterable, Union
+from typing import Any, Mapping, Type, Union, NamedTuple, Sequence
 
 from dlt.common.exceptions import DltException
 
 
+class LookupTrace(NamedTuple):
+    provider: str
+    namespaces: Sequence[str]
+    key: str
+    value: Any
+
+
 class ConfigurationException(DltException):
     def __init__(self, msg: str) -> None:
         super().__init__(msg)
 
 
-class ConfigEntryMissingException(ConfigurationException):
-    """thrown when not all required config elements are present"""
 
-    def __init__(self, missing_set: Iterable[str], namespace: str = None) -> None:
-        self.missing_set = missing_set
-        self.namespace = namespace
+class ContainerException(ConfigurationException):
+    """base exception for all exceptions related to injectable container"""
+    pass
+
+
+class ConfigProviderException(ConfigurationException):
+    """base exceptions for all exceptions raised by config providers"""
+    pass
+
+
+class ConfigurationWrongTypeException(ConfigurationException):
+    def __init__(self, _typ: type) -> None:
+        super().__init__(f"Invalid configuration instance type {_typ}. Configuration instances must derive from BaseConfiguration.")
+
 
-        msg = 'Missing config keys: ' + str(missing_set)
-        if namespace:
-            msg += ". Note that required namespace for that keys is " + namespace + " and namespace separator is two underscores"
+class ConfigFieldMissingException(ConfigurationException):
+    """thrown when not all required config fields are present"""
+
+    def __init__(self, spec_name: str, traces: Mapping[str, Sequence[LookupTrace]]) -> None:
+        self.traces = traces
+        self.spec_name = spec_name
+
+        msg = f"Following fields are missing: {str(list(traces.keys()))} in configuration with spec {spec_name}\n"
+        for f, field_traces in traces.items():
+            msg += f'\tfor field "{f}" config providers and keys were tried in following order:\n'
+            for tr in field_traces:
+                msg += f'\t\tIn {tr.provider} key {tr.key} was not found.\n'
         super().__init__(msg)
 
 
-class ConfigEnvValueCannotBeCoercedException(ConfigurationException):
-    """thrown when value from ENV cannot be coerced to hinted type"""
+class ConfigValueCannotBeCoercedException(ConfigurationException):
+    """thrown when value returned by config provider cannot be coerced to hinted type"""
 
-    def __init__(self, attr_name: str, env_value: str, hint: type) -> None:
-        self.attr_name = attr_name
-        self.env_value = env_value
+    def __init__(self, field_name: str, field_value: Any, hint: type) -> None:
+        self.field_name = field_name
+        self.field_value = field_value
         self.hint = hint
-        super().__init__('env value %s cannot be coerced into type %s in attr %s' % (env_value, str(hint), attr_name))
+        super().__init__('env value %s cannot be coerced into type %s in attr %s' % (field_value, str(hint), field_name))
 
 
 class ConfigIntegrityException(ConfigurationException):
@@ -46,3 +71,55 @@ class ConfigFileNotFoundException(ConfigurationException):
 
     def __init__(self, path: str) -> None:
         super().__init__(f"Missing config file in {path}")
+
+
+class ConfigFieldMissingTypeHintException(ConfigurationException):
+    """thrown when configuration specification does not have type hint"""
+
+    def __init__(self, field_name: str, spec: Type[Any]) -> None:
+        self.field_name = field_name
+        self.typ_ = spec
+        super().__init__(f"Field {field_name} on configspec {spec} does not provide required type hint")
+
+
+class ConfigFieldTypeHintNotSupported(ConfigurationException):
+    """thrown when configuration specification uses not supported type in hint"""
+
+    def __init__(self, field_name: str, spec: Type[Any], typ_: Type[Any]) -> None:
+        self.field_name = field_name
+        self.typ_ = spec
+        super().__init__(f"Field {field_name} on configspec {spec} has hint with unsupported type {typ_}")
+
+
+class ValueNotSecretException(ConfigurationException):
+    def __init__(self, provider_name: str, key: str) -> None:
+        self.provider_name = provider_name
+        self.key = key
+        super().__init__(f"Provider {provider_name} cannot hold secret values but key {key} with secret value is present")
+
+
+class InvalidInitialValue(ConfigurationException):
+    def __init__(self, spec: Type[Any], initial_value_type: Type[Any]) -> None:
+        self.spec = spec
+        self.initial_value_type = initial_value_type
+        super().__init__(f"Initial value of type {initial_value_type} is not valid for {spec.__name__}")
+
+
+class ContainerInjectableContextMangled(ContainerException):
+    def __init__(self, spec: Type[Any], existing_config: Any, expected_config: Any) -> None:
+        self.spec = spec
+        self.existing_config = existing_config
+        self.expected_config = expected_config
+        super().__init__(f"When restoring context {spec.__name__}, instance {expected_config} was expected, instead instance {existing_config} was found.")
+
+
+class ContextDefaultCannotBeCreated(ContainerException, KeyError):
+    def __init__(self, spec: Type[Any]) -> None:
+        self.spec = spec
+        super().__init__(f"Container cannot create the default value of context {spec.__name__}.")
+
+
+class DuplicateConfigProviderException(ConfigProviderException):
+    def __init__(self, provider_name: str) -> None:
+        self.provider_name = provider_name
+        super().__init__(f"Provider with name {provider_name} already present in ConfigProvidersContext")