BODConfigLoader

BODConfigLoader

Chargeur de configurations BOD depuis différentes sources.

Functions

from_dict staticmethod

Python

from_dict(config_dict: Dict[str, Any]) -> BODConfig

Crée un objet :class:BODConfig à partir d'un dictionnaire.

Cette méthode résout les transformateurs référencés par nom via le registre de transformateurs.

Parameters:

Name	Type	Description	Default
config_dict	Dict[str, Any]	Dictionnaire de configuration.	required

Returns:

Name	Type	Description
BODConfig	BODConfig	Instance configurée.

Raises:

Type	Description
ValueError	Si un transformateur référencé est introuvable.

Source code in cmpparis/bod_config_loader.py

@staticmethod
def from_dict(config_dict: Dict[str, Any]) -> BODConfig:
    """Crée un objet :class:`BODConfig` à partir d'un dictionnaire.

    Cette méthode résout les transformateurs référencés par nom via
    le registre de transformateurs.

    Args:
        config_dict (Dict[str, Any]): Dictionnaire de configuration.

    Returns:
        BODConfig: Instance configurée.

    Raises:
        ValueError: Si un transformateur référencé est introuvable.
    """
    # Extract transformers and convert from names to functions
    header_transformers = None
    line_transformers = None

    if 'header_transformers' in config_dict:
        header_transformers = {}
        for field, transformer_name in config_dict['header_transformers'].items():
            header_transformers[field] = get_transformer(transformer_name)

    if 'line_transformers' in config_dict:
        line_transformers = {}
        for field, transformer_name in config_dict['line_transformers'].items():
            line_transformers[field] = get_transformer(transformer_name)

    return BODConfig(
        header_xpath=config_dict['header_xpath'],
        lines_xpath=config_dict['lines_xpath'],
        header_mapping=config_dict['header_mapping'],
        line_mapping=config_dict['line_mapping'],
        header_transformers=header_transformers,
        line_transformers=line_transformers,
        csv_fieldnames=config_dict.get('csv_fieldnames'),
        flatten_mode=config_dict.get('flatten_mode', 'duplicate_header')
    )

from_yaml staticmethod

Python

from_yaml(file_path: str) -> BODConfig

Charge une configuration depuis un fichier YAML local.

Parameters:

Name	Type	Description	Default
file_path	str	Chemin du fichier YAML.	required

Returns:

Name	Type	Description
BODConfig	BODConfig	Instance de configuration.

Source code in cmpparis/bod_config_loader.py

@staticmethod
def from_yaml(file_path: str) -> BODConfig:
    """Charge une configuration depuis un fichier YAML local.

    Args:
        file_path (str): Chemin du fichier YAML.

    Returns:
        BODConfig: Instance de configuration.
    """
    logger.info(f"Loading BOD config from YAML: {file_path}")

    with open(file_path, 'r', encoding='utf-8') as f:
        config_dict = yaml.safe_load(f)

    return BODConfigLoader.from_dict(config_dict)

from_json staticmethod

Python

from_json(file_path: str) -> BODConfig

Charge une configuration depuis un fichier JSON local.

Parameters:

Name	Type	Description	Default
file_path	str	Chemin du fichier JSON.	required

Returns:

Name	Type	Description
BODConfig	BODConfig	Instance de configuration.

Source code in cmpparis/bod_config_loader.py

@staticmethod
def from_json(file_path: str) -> BODConfig:
    """Charge une configuration depuis un fichier JSON local.

    Args:
        file_path (str): Chemin du fichier JSON.

    Returns:
        BODConfig: Instance de configuration.
    """
    logger.info(f"Loading BOD config from JSON: {file_path}")

    with open(file_path, 'r', encoding='utf-8') as f:
        config_dict = json.load(f)

    return BODConfigLoader.from_dict(config_dict)

from_s3 staticmethod

Python

from_s3(bucket: str, key: str, s3_client: Optional[Any] = None) -> BODConfig

Charge une configuration depuis S3.

Parameters:

Name	Type	Description	Default
bucket	str	Nom du bucket S3.	required
key	str	Clé de l'objet S3.	required
s3_client	Optional[Any]	Client S3 optionnel (utilise cmpparis.S3 si non fourni).	None

Returns:

Name	Type	Description
BODConfig	BODConfig	Instance de configuration.

Raises:

Type	Description
ValueError	Si l'extension de fichier n'est pas supportée.

Source code in cmpparis/bod_config_loader.py

@staticmethod
def from_s3(bucket: str, key: str, s3_client: Optional[Any] = None) -> BODConfig:
    """Charge une configuration depuis S3.

    Args:
        bucket (str): Nom du bucket S3.
        key (str): Clé de l'objet S3.
        s3_client (Optional[Any]): Client S3 optionnel (utilise ``cmpparis.S3`` si non fourni).

    Returns:
        BODConfig: Instance de configuration.

    Raises:
        ValueError: Si l'extension de fichier n'est pas supportée.
    """
    logger.info(f"Loading BOD config from S3: s3://{bucket}/{key}")

    # Use cmpparis S3 class if no client provided
    if s3_client is None:
        from .s3 import S3
        s3_client = S3()

    # Download file content
    content = s3_client.download_file_as_string(bucket, key)

    # Detect format by extension
    if key.endswith('.yaml') or key.endswith('.yml'):
        config_dict = yaml.safe_load(content)
    elif key.endswith('.json'):
        config_dict = json.loads(content)
    else:
        raise ValueError(f"Unsupported file format: {key}. Use .yaml, .yml, or .json")

    return BODConfigLoader.from_dict(config_dict)

from_string staticmethod

Python

from_string(content: str, format: str = 'yaml') -> BODConfig

Charge une configuration depuis une chaîne.

Parameters:

Name	Type	Description	Default
content	str	Contenu de la configuration.	required
format	str	'yaml' ou 'json'.	'yaml'

Returns:

Name	Type	Description
BODConfig	BODConfig	Instance de configuration.

Raises:

Type	Description
ValueError	Si le format n'est pas supporté.

Source code in cmpparis/bod_config_loader.py

@staticmethod
def from_string(content: str, format: str = 'yaml') -> BODConfig:
    """Charge une configuration depuis une chaîne.

    Args:
        content (str): Contenu de la configuration.
        format (str): ``'yaml'`` ou ``'json'``.

    Returns:
        BODConfig: Instance de configuration.

    Raises:
        ValueError: Si le format n'est pas supporté.
    """
    if format.lower() == 'yaml':
        config_dict = yaml.safe_load(content)
    elif format.lower() == 'json':
        config_dict = json.loads(content)
    else:
        raise ValueError(f"Unsupported format: {format}. Use 'yaml' or 'json'")

    return BODConfigLoader.from_dict(config_dict)

validate_config staticmethod

Python

validate_config(config_dict: Dict[str, Any]) -> bool

Valide un dictionnaire de configuration.

Parameters:

Name	Type	Description	Default
config_dict	Dict[str, Any]	Configuration à valider.	required

Returns:

Name	Type	Description
bool	bool	True si la configuration est valide.

Raises:

Type	Description
ValueError	Si des champs obligatoires manquent ou si un transformateur est introuvable.

Source code in cmpparis/bod_config_loader.py

@staticmethod
def validate_config(config_dict: Dict[str, Any]) -> bool:
    """Valide un dictionnaire de configuration.

    Args:
        config_dict (Dict[str, Any]): Configuration à valider.

    Returns:
        bool: ``True`` si la configuration est valide.

    Raises:
        ValueError: Si des champs obligatoires manquent ou si un transformateur est introuvable.
    """
    required_fields = ['header_xpath', 'lines_xpath', 'header_mapping', 'line_mapping']

    for field in required_fields:
        if field not in config_dict:
            raise ValueError(f"Missing required field: {field}")

    # Validate transformers exist
    if 'header_transformers' in config_dict:
        for transformer_name in config_dict['header_transformers'].values():
            get_transformer(transformer_name)  # Will raise if not found

    if 'line_transformers' in config_dict:
        for transformer_name in config_dict['line_transformers'].values():
            get_transformer(transformer_name)  # Will raise if not found

    return True