Merge branch 'master' into DOG-4448-file-extractor-simplify-extractio…

…n-failure-analysis

cognite/extractorutils/_inner_util.py

@@ -25,6 +25,17 @@ def _resolve_log_level(level: str) -> int:
     return {"NOTSET": 0, "DEBUG": 10, "INFO": 20, "WARNING": 30, "ERROR": 40, "CRITICAL": 50}[level.upper()]
 
 
+def resolve_log_level_for_httpx(level: str) -> str:
+    return {
+        None: "WARNING",
+        "INFO": "WARNING",
+        "WARNING": "WARNING",
+        "ERROR": "ERROR",
+        "CRITICAL": "CRITICAL",
+        "DEBUG": "DEBUG",
+    }.get(level, "WARNING")
+
+
 class _DecimalEncoder(json.JSONEncoder):
     def default(self, obj: Any) -> Dict[str, str]:
         if isinstance(obj, Decimal):

cognite/extractorutils/configtools/elements.py

@@ -32,6 +32,7 @@
     OAuthClientCredentials,
 )
 from cognite.client.data_classes import Asset, DataSet, ExtractionPipeline
+from cognite.extractorutils._inner_util import resolve_log_level_for_httpx
 from cognite.extractorutils.configtools._util import _load_certificate_data
 from cognite.extractorutils.exceptions import InvalidConfigError
 from cognite.extractorutils.metrics import (
@@ -491,6 +492,14 @@ def setup_logging(self, suppress_console: bool = False) -> None:
             if root.getEffectiveLevel() > file_handler.level:
                 root.setLevel(file_handler.level)
 
+            log_level = logging.getLevelName(root.getEffectiveLevel())
+            httpx_log_level = resolve_log_level_for_httpx(log_level)
+            httpx_logger = logging.getLogger("httpx")
+            httpx_logger.setLevel(httpx_log_level)
+
+            http_core_logger = logging.getLogger("httpcore")
+            http_core_logger.setLevel(httpx_log_level)
+
 
 @dataclass
 class _PushGatewayConfig:

cognite/extractorutils/configtools/loaders.py

@@ -163,6 +163,9 @@ def ignore_unknown(self, node: yaml.Node) -> None:
     SafeLoaderIgnoreUnknown.add_constructor(None, SafeLoaderIgnoreUnknown.ignore_unknown)  # type: ignore
     initial_load = yaml.load(source, Loader=SafeLoaderIgnoreUnknown)  # noqa: S506
 
+    if not isinstance(initial_load, dict):
+        raise InvalidConfigError("The root node of the YAML document must be an object")
+
     if not isinstance(source, str):
         source.seek(0)
 

cognite/extractorutils/unstable/configuration/exceptions.py

@@ -0,0 +1,24 @@
+from typing import List, Optional
+
+
+class InvalidConfigError(Exception):
+    """
+    Exception thrown from ``load_yaml`` and ``load_yaml_dict`` if config file is invalid. This can be due to
+
+      * Missing fields
+      * Incompatible types
+      * Unkown fields
+    """
+
+    def __init__(self, message: str, details: Optional[List[str]] = None):
+        super(InvalidConfigError, self).__init__()
+        self.message = message
+        self.details = details
+
+        self.attempted_revision: int | None = None
+
+    def __str__(self) -> str:
+        return f"Invalid config: {self.message}"
+
+    def __repr__(self) -> str:
+        return self.__str__()

cognite/extractorutils/unstable/configuration/loaders.py

@@ -8,9 +8,13 @@
 
 from cognite.client import CogniteClient
 from cognite.extractorutils.configtools.loaders import _load_yaml_dict_raw
-from cognite.extractorutils.exceptions import InvalidConfigError
+from cognite.extractorutils.exceptions import InvalidConfigError as OldInvalidConfigError
+from cognite.extractorutils.unstable.configuration.exceptions import InvalidConfigError
 from cognite.extractorutils.unstable.configuration.models import ConfigModel
 
+__all__ = ["ConfigFormat", "load_file", "load_from_cdf", "load_io", "load_dict"]
+
+
 _T = TypeVar("_T", bound=ConfigModel)
 
 
@@ -44,7 +48,17 @@ def load_from_cdf(
     )
     response.raise_for_status()
     data = response.json()
-    return load_io(StringIO(data["config"]), ConfigFormat.YAML, schema), data["revision"]
+
+    try:
+        return load_io(StringIO(data["config"]), ConfigFormat.YAML, schema), data["revision"]
+
+    except InvalidConfigError as e:
+        e.attempted_revision = data["revision"]
+        raise e
+    except OldInvalidConfigError as e:
+        new_e = InvalidConfigError(e.message)
+        new_e.attempted_revision = data["revision"]
+        raise new_e from e
 
 
 def load_io(stream: TextIO, format: ConfigFormat, schema: Type[_T]) -> _T:
@@ -100,12 +114,9 @@ def load_dict(data: dict, schema: Type[_T]) -> _T:
             if "ctx" in err and "error" in err["ctx"]:
                 exc = err["ctx"]["error"]
                 if isinstance(exc, ValueError) or isinstance(exc, AssertionError):
-                    messages.append(f"{loc_str}: {str(exc)}")
+                    messages.append(f"{str(exc)}: {loc_str}")
                     continue
 
-            if err.get("type") == "json_invalid":
-                messages.append(f"{err.get('msg')}: {loc_str}")
-            else:
-                messages.append(f"{loc_str}: {err.get('msg')}")
+            messages.append(f"{err.get('msg')}: {loc_str}")
 
         raise InvalidConfigError(", ".join(messages), details=messages) from e

cognite/extractorutils/unstable/configuration/models.py

@@ -19,6 +19,21 @@
 from cognite.extractorutils.configtools._util import _load_certificate_data
 from cognite.extractorutils.exceptions import InvalidConfigError
 
+__all__ = [
+    "ConfigModel",
+    "AuthenticationConfig",
+    "TimeIntervalConfig",
+    "ConnectionConfig",
+    "CronConfig",
+    "IntervalConfig",
+    "ScheduleConfig",
+    "LogLevel",
+    "LogFileHandlerConfig",
+    "LogConsoleHandlerConfig",
+    "LogHandlerConfig",
+    "ExtractorConfig",
+]
+
 
 class ConfigModel(BaseModel):
     model_config = ConfigDict(
@@ -139,7 +154,7 @@ class ConnectionConfig(ConfigModel):
     project: str
     base_url: str
 
-    extraction_pipeline: str
+    integration: str
 
     authentication: AuthenticationConfig
 
@@ -223,17 +238,24 @@ class LogLevel(Enum):
 
 
 class LogFileHandlerConfig(ConfigModel):
+    type: Literal["file"]
     path: Path
     level: LogLevel
     retention: int = 7
 
 
 class LogConsoleHandlerConfig(ConfigModel):
+    type: Literal["console"]
     level: LogLevel
 
 
-LogHandlerConfig = Union[LogFileHandlerConfig, LogConsoleHandlerConfig]
+LogHandlerConfig = Annotated[LogFileHandlerConfig | LogConsoleHandlerConfig, Field(discriminator="type")]
+
+
+# Mypy BS
+def _log_handler_default() -> List[LogHandlerConfig]:
+    return [LogConsoleHandlerConfig(type="console", level=LogLevel.INFO)]
 
 
 class ExtractorConfig(ConfigModel):
-    log_handlers: List[LogHandlerConfig] = Field(default_factory=lambda: [LogConsoleHandlerConfig(level=LogLevel.INFO)])
+    log_handlers: List[LogHandlerConfig] = Field(default_factory=_log_handler_default)

cognite/extractorutils/unstable/core/_dto.py

@@ -32,3 +32,13 @@ class TaskUpdate(CogniteModel):
     type: Literal["started"] | Literal["ended"]
     name: str
     timestamp: int
+
+
+class Error(CogniteModel):
+    external_id: str
+    level: str
+    description: str
+    details: str | None
+    start_time: int
+    end_time: int | None
+    task: str | None