cdm_data_loaders.core.settings ¶

class CdmDataLoadersBase(BaseSettings):
    """Base for all CDM Data Loaders settings classes.

    - Sets up some basic CLI parsing defaults
    - Adds a class method to check for conflicting aliases
    """

    model_config = SettingsConfigDict(**DEFAULT_SETTINGS_CONFIG_DICT)

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
        """Validate the subclass's cli shortcuts and aliases."""
        super().__pydantic_init_subclass__(**kwargs)
        cls.check_aliases()

    @classmethod
    def _cli_name(cls, name: str) -> str:
        """Get the CLI argument name pydantic-settings registers for a field or alias name."""
        return name.replace("_", "-") if cls.model_config.get("cli_kebab_case") else name

    @classmethod
    def _field_cli_names(cls, field_name: str, field_info: FieldInfo) -> set[str]:
        """CLI argument names a field may register: its name plus any aliases.

        Deliberately over-inclusive -- reserving a name that is never registered can only cause
        a false positive, never a missed collision.
        """
        names = {field_name} | _alias_names(field_info.alias) | _alias_names(field_info.validation_alias)
        return {cls._cli_name(name) for name in names}

    @classmethod
    def check_aliases(cls) -> None:
        """Raise `SettingsError` if the class's CLI argument names or shortcuts are invalid or collide.

        pydantic-settings silently drops shortcuts whose target does not match a registered argument
        name, and silently drops any name already registered by an earlier argument, so both are
        checked here. Names differing only in '_' vs '-' are treated as the same name.
        """
        owners: dict[str, str] = {}
        errors = []

        def claim(name: str, owner: str, errors: list[str]) -> None:
            key = name.replace("_", "-")
            if key in RESERVED_CLI_NAMES:
                err_msg = f"{cls.__name__}: CLI name {name!r} for {owner!r} is reserved by argparse"
                errors.append(err_msg)
                return
            other = owners.setdefault(key, owner)
            if other == owner and key in claimed:
                err_msg = f"{cls.__name__}: CLI name {name!r} is claimed twice by {owner!r}"
                errors.append(err_msg)
                return
            if other != owner:
                both = " and ".join(sorted(repr(o) for o in (other, owner)))
                err_msg = f"{cls.__name__}: CLI name {name!r} is claimed by both {both}"
                errors.append(err_msg)
                return
            claimed.add(key)

        claimed: set[str] = set()
        registered: dict[str, str] = {}
        for field_name, field_info in cls.model_fields.items():
            cli_names = cls._field_cli_names(field_name, field_info)
            # claim each normalised name once per field, so a field's own name and alias can coincide
            for cli_name in {name.replace("_", "-"): name for name in cli_names}.values():
                claim(cli_name, field_name, errors)
            registered.update(dict.fromkeys(cli_names, field_name))

        cli_shortcuts: Mapping[str, str | list[str]] = cls.model_config.get("cli_shortcuts") or {}
        for target, shortcuts in cli_shortcuts.items():
            owner = registered.get(target)
            if owner is None:
                suggestion = cls._cli_name(target)
                hint = f" (did you mean {suggestion!r}?)" if suggestion in registered else ""
                err_msg = f"{cls.__name__}: cli_shortcuts target {target!r} does not match any CLI argument{hint}"
                errors.append(err_msg)
                continue
            for shortcut in [shortcuts] if isinstance(shortcuts, str) else shortcuts:
                if not shortcut or shortcut.startswith("-"):
                    err_msg = f"{cls.__name__}: invalid shortcut {shortcut!r} for {owner!r}; omit the leading dashes"
                    errors.append(err_msg)
                    continue
                claim(shortcut, owner, errors)
        if errors:
            raise SettingsError("\n".join(errors))

class LoggerSettings(CdmDataLoadersBase):
    """Configuration for a class with a logger config."""

    log_config_file: LogConfigFile

class InputLocalOutputSettings(LoggerSettings):
    """Configuration for a DLT pipeline with an input directory or URL and a local output directory."""

    model_config = default_settings_with_shortcuts(cli_shortcuts={INPUT_DIR: "i", OUTPUT_DIR: "o"})

    input_dir: InputDir
    output_dir: LocalOutputDir

class CtsSettings(InputLocalOutputSettings):
    """Configuration for running a basic DLT pipeline."""

    model_config: SettingsConfigDict = default_settings_with_shortcuts(cli_shortcuts=CLI_SHORTCUTS)

    buffer_size: BufferSize
    dlt_dev_mode: DltDevMode
    log_interval: LogInterval
    use_destination: UseDestination
    use_output_dir_for_pipeline_metadata: UseOutputDirForPipelineMetadata

    @computed_field
    @property
    def pipeline_dir(self) -> Path | None:
        """Custom directory to save pipeline metadata to.

        If use_output_dir_for_pipeline_metadata is true, this is ``<output_dir>/<PIPELINE_DIR_NAME>``;
        otherwise None, and dlt uses its default.
        """
        if self.use_output_dir_for_pipeline_metadata:
            return self.output_dir / PIPELINE_DIR_NAME
        return None

    @computed_field
    @property
    def report_path(self) -> Path:
        """Path to the report file.

        If use_output_dir_for_pipeline_metadata is true, the report is saved in the pipelines dir;
        Otherwise, it is saved directly to the output directory.
        """
        if self.pipeline_dir:
            return self.pipeline_dir / REPORT_FILE_NAME
        return self.output_dir / REPORT_FILE_NAME

class FileFormatIngestSettings(CtsSettings):
    """Configuration for DLT pipelines that import a particular file format."""

    model_config: SettingsConfigDict = default_settings_with_shortcuts(cli_shortcuts=FILE_FORMAT_CLI_SHORTCUTS)

    dataset_name: DatasetName
    loader_file_format: LoaderFileFormat

    @property
    def use_pyarrow(self) -> bool:
        """Return whether this pipeline yields ``pyarrow.Table`` pages instead of row dicts."""
        return uses_pyarrow_format(self)

class XmlIngestSettings(FileFormatIngestSettings):
    """Configuration for DLT pipelines that import XML files."""

    model_config: SettingsConfigDict = default_settings_with_shortcuts(cli_shortcuts=XML_CLI_SHORTCUTS)

    file_glob: Annotated[
        FileGlob,
        Field(
            default=DEFAULT_XML_FILE_GLOB,
            description=file_glob_description("Glob pattern for XML files"),
        ),
    ]

class SingleTableXmlIngestSettings(XmlIngestSettings, TransformSettingsMixin):
    """Configuration for DLT pipelines that import XML elements into a single target table."""

    table_name: TableName
    xml_tag: XmlTag