Pipeline (`kadi.kidas.pipeline`)¶

Pipeline is the central orchestrator of the kidas module. It chains cleaning, validation, and normalization steps declaratively, automatically detects source type from file path or URL, and manages result caching via SQLite.

Its API is fluid: each method returns the current instance, allowing method chaining without intermediate variables.

Initialization¶

from kadi.kidas import Pipeline

pipeline = Pipeline()

Pipeline is initialized without arguments. Source and steps are configured via chained methods.

Configuration Methods¶

`add_source(source, **kwargs)`¶

Configures the pipeline data source. Automatically detects source type from file extension or URL prefix.

pipeline.add_source("recoltes_2024.csv")
pipeline.add_source("rapport.xlsx", sheet="Parakou")
pipeline.add_source("https://api.data.bj/agriculture/prices")

It is also possible to pass a Source instance directly:

from kadi.io import CSVSource

source = CSVSource("recoltes.csv", sep=";", encoding="latin-1")
pipeline.add_source(source)

Parameter	Type	Description
`source`	`str` or `Source`	Path, URL, or `Source` instance
`**kwargs`	-	Arguments passed to the `Source` class constructor

Returns: Pipeline (for chaining).

Short alias: load(source, **kwargs)

`add_step(step_name, schema_or_mappings=None, **params)`¶

Adds a processing step to the pipeline queue. Step type (cleaning, validation, or normalization) is automatically determined based on the step name passed.

pipeline.add_step("drop_dupes")
pipeline.add_step("fill_missing", strategy="median")
pipeline.add_step("check_schema", {"culture": "str", "rendement_kg": "float"})
pipeline.add_step("norm_cols", {"crops": "culture"})

Available Cleaning Steps¶

Name	Description
`'drop_dupes'`	Removes duplicate rows
`'fill_missing'`	Imputes missing values (`strategy` parameter)
`'drop_outliers'`	Removes statistical outliers (`method`, `threshold` parameters)
`'parse_dates'`	Normalizes date columns to `datetime64`
`'norm_text'`	Normalizes text strings (accents, case, whitespace)
`'strip_chars'`	Removes unwanted special characters

Available Validation Steps¶

Name	Description
`'check_schema'`	Verifies column types against a schema dict
`'check_types'`	Validates data types column by column
`'check_ranges'`	Checks that numeric values fall within defined ranges
`'check_coords'`	Validates that GPS coordinates are within Benin bounds
`'check_unique'`	Checks uniqueness of values in specified columns
`'check_fk'`	Checks foreign key constraints between columns

Available Normalization Steps¶

Name	Description
`'norm_cols'`	Renames columns according to a mapping dictionary
`'convert_units'`	Converts units (kg/tonne, ha/m2, etc.)
`'convert_currency'`	Converts currencies via `ExchangeRateClient`
`'std_crops'`	Normalizes crop names to KadiPy reference
`'std_markets'`	Normalizes market names
`'std_coords'`	Normalizes GPS coordinates (format, rounding)

Short Aliases:

clean(step, **params): alias for cleaning steps
validate(schema, **params): alias for add_step('check_schema', schema)
normalize(mappings, **params): alias for add_step('norm_cols', mappings)

Execution¶

`run(cache=True)`¶

Executes all configured steps in order. Loads data from source, applies each step, and caches the result if requested.

df, report = pipeline.run(cache=True)

Parameter	Type	Default	Description
`cache`	`bool`	`True`	Attempts cache loading before reading, and saves result after processing

Returns: tuple[pd.DataFrame, dict]

Exceptions:

PipelineError: if no source was configured.
ReadError: if reading source fails.

Returned Report Structure¶

df, report = pipeline.run()

report["nb_rows_in"]      # int   : raw rows loaded
report["nb_rows_out"]     # int   : rows after processing
report["steps_summary"]   # list  : step names in order
report["quality_score"]   # dict | None : quality score (if validation executed)
report["warnings"]        # list  : validation warnings
report["cache_utilise"]   # bool  : True if data originated from cache
report["details"]         # dict  : internal reports for each step

quality_score is a dict produced by Validator. It is None if no validation step was configured.

Utility Methods¶

`config()`¶

Returns full pipeline configuration: configured source and list of defined steps.

cfg = pipeline.config()
print(f"Number of steps: {cfg['nb_steps']}")
for step in cfg["steps"]:
    print(f"  {step['type']}: {step['nom']} ({step['params']})")

Returns: dict with keys source, nb_steps, steps.

`export(filepath)`¶

Exports internal pipeline report to a JSON or HTML file based on destination file extension.

pipeline.export("rapport_pipeline.json")
pipeline.export("rapport_pipeline.html")

Extension	Format
`.json`	Indented JSON, UTF-8 encoded
`.html`	Simple HTML page with pre-formatted JSON

Returns: True if export succeeds.

Exception: PipelineError if file extension is unsupported.

Full Examples¶

Minimal Pipeline¶

from kadi.kidas import Pipeline

df, report = (
    Pipeline()
    .add_source("recoltes_2024.csv")
    .add_step("drop_dupes")
    .add_step("fill_missing", strategy="median")
    .run()
)
print(f"{report['nb_rows_in']} -> {report['nb_rows_out']} rows")

Pipeline with Validation and Normalization¶

from kadi.kidas import Pipeline

df, report = (
    Pipeline()
    .add_source("enquete_prix_2024.xlsx")
    .add_step("drop_dupes")
    .add_step("fill_missing", strategy="median")
    .add_step("drop_outliers", method="iqr")
    .add_step("check_schema", {
        "culture":      "str",
        "marche":       "str",
        "prix_xof_kg":  "float",
        "date":         "datetime",
    })
    .add_step("std_crops", col="culture")
    .add_step("norm_cols", {"prix_xof_kg": "price_xof"})
    .run(cache=True)
)

if report.get("quality_score"):
    print(f"Quality score: {report['quality_score']['overall']:.2f}")

for warning in report.get("warnings", []):
    print(warning)

Fluid API with Short Aliases¶

from kadi.kidas import Pipeline

df, report = (
    Pipeline()
    .load("prix_2024.csv")
    .clean("drop_dupes")
    .clean("fill_missing", strategy="median")
    .validate({"culture": "str", "prix_xof_kg": "float"})
    .normalize({"crops": "culture"})
    .run(cache=True)
)

Reusing an Existing Source¶

from kadi.io import CSVSource
from kadi.kidas import Pipeline

source = CSVSource("export_maep.csv", sep=";", encoding="latin-1")

df, report = (
    Pipeline()
    .add_source(source)
    .add_step("drop_dupes")
    .add_step("parse_dates", cols=["date_recolte"])
    .run()
)

Report Export¶

pipeline = Pipeline()
df, report = (
    pipeline
    .add_source("donnees.csv")
    .add_step("check_schema", {"culture": "str"})
    .run()
)

pipeline.export("rapport.json")

Cache Behavior¶

The pipeline uses an internal SQLite cache (managed by Cache). The cache key is calculated via SHA-256 hash of the source path, truncated to 16 hexadecimal characters.

When cache=True:

Pipeline checks for an existing entry in cache.
If it exists, data is returned directly without re-reading the source or re-executing steps.
If it does not exist, pipeline executes all steps and saves the result.

The report indicates whether data originated from cache via the cache_utilise key.

To force complete re-execution without using cache:

df, report = pipeline.run(cache=False)

Backward Compatibility¶

Legacy method names emit a DeprecationWarning and continue to function until KadiPy v2.0:

Legacy Name (before v1.2.0)	New Name
`load_data(source)`	`add_source(source)` or `load(source)`
`add_cleaning_step(step)`	`add_step(step)` or `clean(step)`
`add_validation_step(schema)`	`add_step('check_schema', schema)` or `validate(schema)`
`add_normalization_step(mappings)`	`add_step('norm_cols', mappings)` or `normalize(mappings)`
`execute(cache)`	`run(cache)`
`get_pipeline_config()`	`config()`
`export_report(filepath)`	`export(filepath)`

Pipeline ¶

Orchestrateur du flux de traitement de données agricoles kidas.

Pipeline est le point d'entrée unique pour les utilisateurs de kidas. Il permet de définir de manière déclarative et chainable un flux complet : chargement → nettoyage → validation → normalisation → cache.

L'auto-détection du type de source (CSV, Excel, JSON, NetCDF, API) est basée sur l'extension du fichier ou le préfixe 'http' de l'URL.

Attributs

_source (Source | None): Source de données configurée. _df (pd.DataFrame | None): Données courantes dans le pipeline. _steps (list): Liste ordonnée des étapes de traitement configurées. _reports (dict): Rapports agrégés de toutes les étapes. _cache (Cache): Instance du gestionnaire de cache kidas.

Exemple

pipeline = Pipeline() df, rapport = ( ... pipeline ... .add_source('recolte_2024.xlsx') ... .add_step('drop_dupes') ... .add_step('fill_missing', strategy='mean') ... .add_step('check_schema', {'culture': 'str', 'rendement_kg': 'float'}) ... .run(cache=True) ... ) print(rapport['nb_rows_out']) 150 print(rapport['quality_score']['overall']) 0.97

Source code in kadi/kidas/pipeline.py

class Pipeline:
    """Orchestrateur du flux de traitement de données agricoles kidas.

    Pipeline est le point d'entrée unique pour les utilisateurs de kidas.
    Il permet de définir de manière déclarative et chainable un flux complet :
    chargement → nettoyage → validation → normalisation → cache.

    L'auto-détection du type de source (CSV, Excel, JSON, NetCDF, API) est
    basée sur l'extension du fichier ou le préfixe 'http' de l'URL.

    Attributs:
        _source (Source | None): Source de données configurée.
        _df (pd.DataFrame | None): Données courantes dans le pipeline.
        _steps (list): Liste ordonnée des étapes de traitement configurées.
        _reports (dict): Rapports agrégés de toutes les étapes.
        _cache (Cache): Instance du gestionnaire de cache kidas.

    Exemple:
        >>> pipeline = Pipeline()
        >>> df, rapport = (
        ...     pipeline
        ...     .add_source('recolte_2024.xlsx')
        ...     .add_step('drop_dupes')
        ...     .add_step('fill_missing', strategy='mean')
        ...     .add_step('check_schema', {'culture': 'str', 'rendement_kg': 'float'})
        ...     .run(cache=True)
        ... )
        >>> print(rapport['nb_rows_out'])
        150
        >>> print(rapport['quality_score']['overall'])
        0.97
    """

    def __init__(self) -> None:
        """Initialise un pipeline vide prêt à recevoir des étapes de traitement."""
        # Source de données (sera configurée par add_source())
        self._source: Optional[Source] = None

        # DataFrame courant (None jusqu'à l'appel de run())
        self._df: Optional[pd.DataFrame] = None

        # Nombre de lignes brutes avant toute transformation
        self._nrows_in: int = 0
        self._lignes_avant: int = 0  # Alias de rétrocompatibilité

        # Liste ordonnée des étapes : {'type', 'nom', 'params'}
        self._steps: List[Dict[str, Any]] = []

        # Rapports agrégés internes des étapes de traitement
        self._reports: Dict[str, Any] = {
            "source": None,
            "nettoyage": None,
            "validation": None,
            "normalisation": None,
            "cache_utilise": False,
        }

        # Instance du cache kidas
        self._cache: Cache = Cache()

    @staticmethod
    def _detect_kind(source: Union[str, os.PathLike]) -> str:
        """Détecte le type de source de données à partir du chemin ou de l'URL.

        Args:
            source (str | os.PathLike): Chemin vers le fichier ou URL de l'API.

        Returns:
            str: Type détecté parmi 'csv', 'excel', 'json', 'netcdf', 'api'.

        Raises:
            PipelineError: Si le type de source ne peut pas être déterminé.
        """
        source_str = str(source)

        # Détection des APIs par préfixe HTTP/HTTPS
        if source_str.startswith("http://") or source_str.startswith("https://"):
            return "api"

        # Détection par extension de fichier
        _, extension = os.path.splitext(source_str.lower())

        if extension in _EXT_CSV:
            return "csv"
        elif extension in _EXT_EXCEL:
            return "excel"
        elif extension in _EXT_JSON:
            return "json"
        elif extension in _EXT_NETCDF:
            return "netcdf"
        else:
            raise PipelineError(
                f"Extension '{extension}' non reconnue pour la source '{source}'. "
                f"Extensions supportées : CSV, Excel, JSON, NetCDF, API."
            )

    # Alias statique de rétrocompatibilité
    _detecter_type_source = _detect_kind

    def add_source(
        self,
        source: Union[str, os.PathLike, Source],
        **kwargs: Any,
    ) -> "Pipeline":
        """Configure la source de données du pipeline.

        Auto-détecte le type de source si un chemin de fichier est fourni,
        ou utilise directement une instance Source existante.

        Args:
            source (str | os.PathLike | Source): Chemin vers le fichier, URL de l'API,
                ou instance Source directement.
            **kwargs: Arguments optionnels transmis au constructeur de la
                Source (ex: encoding, sheet_name, use_dask).

        Returns:
            Pipeline: L'instance courante (pour le chaînage).

        Raises:
            PipelineError: Si le type de source est indéterminable.
        """
        if isinstance(source, Source):
            # Utilisation directe d'une Source existante
            self._source = source
            type_source = source.kind
        else:
            # Auto-détection et instanciation de la Source appropriée
            type_source = self._detecter_type_source(source)

            if type_source == "csv":
                self._source = CSVSource(source, **kwargs)
            elif type_source == "excel":
                self._source = ExcelSource(source, **kwargs)
            elif type_source == "json":
                self._source = JSONSource(source, **kwargs)
            elif type_source == "netcdf":
                self._source = NetCDFSource(source, **kwargs)
            elif type_source == "api":
                self._source = APISource(source, **kwargs)

        # Enregistrement dans le rapport
        self._reports["source"] = {
            "path": str(source),
            "type": type_source,
        }

        logger.info(
            "Pipeline kidas : source '%s' configurée (type: %s).",
            source,
            type_source,
        )

        return self

    def add_step(
        self,
        step_name: str,
        schema_or_mappings: Any = None,
        **params: Any,
    ) -> "Pipeline":
        """Ajoute une étape de traitement à la file du pipeline.

        Accepte des étapes de nettoyage, de validation ou de normalisation
        en un seul point d'entrée. Le type de l'étape est déterminé
        automatiquement selon le nom passé.

        Args:
            step_name (str): Nom de l'étape. Valeurs acceptées :
                - Nettoyage : 'drop_dupes', 'fill_missing', 'drop_outliers',
                  'parse_dates', 'norm_text', 'strip_chars'.
                - Validation : 'check_schema', 'check_types',
                  'check_ranges', 'check_coords', 'check_unique', 'check_fk'.
                - Normalisation : 'norm_cols', 'convert_units',
                  'convert_currency', 'std_crops', 'std_markets', 'std_coords'.
            schema_or_mappings: Argument positionnel optionnel pour les étapes
                de validation (schema dict) ou normalisation (mappings dict).
            **params: Paramètres supplémentaires à passer à la méthode
                (ex: strategy='mean', cols=['id', 'date']).

        Returns:
            Pipeline: L'instance courante (pour le chaînage).
        """
        # Détermination automatique du type d'étape
        _cleaning = {
            "drop_dupes", "fill_missing", "drop_outliers",
            "parse_dates", "norm_text", "strip_chars",
            "check_decimals", "report",
            "remove_duplicates", "handle_missing_values", "remove_outliers",
            "fix_dates", "standardize_text", "remove_special_chars",
            "detect_inconsistent_decimals", "get_cleaning_report",
        }
        _validation = {
            "check_schema", "check_types", "check_ranges",
            "check_coords", "check_unique", "check_fk",
            "quality_score", "validate_schema", "validate_types",
            "validate_ranges", "validate_coordinates", "validate_uniqueness",
        }
        _normalisation = {
            "norm_cols", "convert_units", "convert_currency",
            "std_crops", "std_markets", "std_coords", "mappings",
            "normalize_column_names", "normalize_units", "normalize_currencies",
            "normalize_crop_names", "normalize_market_names", "normalize_geometry",
            "get_normalization_mapping", "normalize",
        }

        if step_name in _cleaning:
            type_etape = "cleaning"
        elif step_name in _validation:
            type_etape = "validation"
        elif step_name in _normalisation:
            type_etape = "normalization"
        else:
            type_etape = "cleaning"  # Fallback tolérant

        # Gestion du premier argument positionnel
        if schema_or_mappings is not None:
            if type_etape == "validation":
                params["schema"] = schema_or_mappings
            elif type_etape == "normalization":
                params["mappings"] = schema_or_mappings

        # Enregistrement de l'étape dans la file
        self._steps.append({
            "type": type_etape,
            "nom": step_name,
            "params": params,
        })

        logger.debug(
            "Etape '%s' (type: %s) ajoutée (params: %s).",
            step_name, type_etape, params,
        )

        return self

    def load(self, source: Union[str, Source], **kwargs: Any) -> "Pipeline":
        """Configure la source de données du pipeline (alias court de add_source).

        Args:
            source (str | Source): Chemin vers le fichier, URL ou instance Source.
            **kwargs: Arguments transmis au constructeur de la Source.

        Returns:
            Pipeline: L'instance courante pour le chaînage.
        """
        return self.add_source(source, **kwargs)

    def clean(self, step: str, **params: Any) -> "Pipeline":
        """Ajoute une étape de nettoyage au pipeline (alias court de add_step).

        Args:
            step (str): Nom de l'étape de nettoyage (ex: 'drop_dupes').
            **params: Paramètres de l'étape de nettoyage.

        Returns:
            Pipeline: L'instance courante pour le chaînage.
        """
        return self.add_step(step, **params)

    def validate(self, schema: Any = None, **params: Any) -> "Pipeline":
        """Ajoute une étape de validation au pipeline (alias court de add_step).

        Args:
            schema (dict | str): Schéma de validation ou nom de l'étape.
            **params: Paramètres de validation.

        Returns:
            Pipeline: L'instance courante pour le chaînage.
        """
        if isinstance(schema, str):
            return self.add_step(schema, **params)
        return self.add_step("check_schema", schema, **params)

    def normalize(self, mappings: Any = None, **params: Any) -> "Pipeline":
        """Ajoute une étape de normalisation au pipeline (alias court de add_step).

        Args:
            mappings (dict | str): Mappings de normalisation ou nom de l'étape.
            **params: Paramètres de normalisation.

        Returns:
            Pipeline: L'instance courante pour le chaînage.
        """
        if isinstance(mappings, str):
            return self.add_step(mappings, **params)
        return self.add_step("norm_cols", mappings, **params)

    def run(
        self,
        cache: bool = True,
    ) -> Tuple[pd.DataFrame, Dict]:
        """Exécute toutes les étapes configurées du pipeline.

        Charge les données depuis la source, applique les étapes de
        nettoyage, validation et normalisation dans l'ordre, puis
        met le résultat en cache si demandé.

        Args:
            cache (bool): Si True, tente de charger depuis le cache avant
                la lecture et sauvegarde le résultat final. Par défaut True.

        Returns:
            tuple[pd.DataFrame, dict]: Tuple contenant :
                - Le DataFrame traité et prêt à l'emploi.
                - Le rapport structuré selon la documentation :
                    - ``nb_rows_in``    : nombre de lignes brutes chargées.
                    - ``nb_rows_out``   : nombre de lignes après traitement.
                    - ``steps_summary`` : liste des noms d'étapes appliquées.
                    - ``quality_score`` : score qualité (dict) ou None.
                    - ``warnings``      : liste des avertissements détectés.
                    - ``cache_utilise`` : True si les données viennent du cache.
                    - ``details``       : rapports internes (source, nettoyage,
                      validation, normalisation).

        Raises:
            PipelineError: Si aucune source n'a été configurée.
            ReadError: Si la lecture de la source échoue.
        """
        # Vérification qu'une source a été configurée
        if self._source is None:
            raise PipelineError(
                "Aucune source configurée. Appelez add_source() avant run()."
            )

        # Génération d'une clé de cache sécurisée : hachage SHA-256 du chemin brut,
        # tronqué à 16 caractères hexadécimaux pour éviter les collisions et les
        # erreurs d'encodage avec des chemins contenant des espaces ou des accents.
        _empreinte = hashlib.sha256(
            str(self._source.path).encode("utf-8")
        ).hexdigest()[:16]
        cle_cache = f"pipeline_{_empreinte}"

        # Tentative de chargement depuis le cache
        if cache:
            df_cached, _ = self._cache.get(cle_cache)
            if df_cached is not None:
                logger.info(
                    "Pipeline kidas : données chargées depuis le cache (clé: '%s').",
                    cle_cache,
                )
                # Construction du rapport minimal pour un retour depuis le cache
                rapport_cache = {
                    "nb_rows_in":    len(df_cached),
                    "nb_rows_out":   len(df_cached),
                    "steps_summary": [e["nom"] for e in self._steps],
                    "quality_score": None,
                    "warnings":      [],
                    "cache_utilise": True,
                    "details": {
                        "source":        self._reports.get("source"),
                        "nettoyage":     None,
                        "validation":    None,
                        "normalisation": None,
                    },
                }
                return df_cached, rapport_cache

        # Lecture des données depuis la source et mémorisation du nombre de lignes brutes
        try:
            self._df = self._source.read()
            # Capture du nombre de lignes avant tout traitement
            self._lignes_avant = len(self._df)
            logger.info(
                "Pipeline kidas : %d lignes chargées depuis '%s'.",
                self._lignes_avant,
                self._source.path,
            )
        except Exception as erreur:
            raise ReadError(
                f"Echec de lecture dans le pipeline : {erreur}"
            ) from erreur

        # Exécution des étapes dans l'ordre
        for etape in self._steps:
            self._df = self._exec_step(etape)

        # Sauvegarde en cache du résultat final
        if cache and self._df is not None:
            self._cache.set(cle_cache, self._df)
            self._reports["cache_utilise"] = True

        # Calcul du nombre de lignes en sortie
        nb_lignes_out = len(self._df) if self._df is not None else 0

        # Construction du rapport final aligné sur la documentation publique
        rapport_final = {
            "nb_rows_in":    self._lignes_avant,
            "nb_rows_out":   nb_lignes_out,
            "steps_summary": [e["nom"] for e in self._steps],
            "quality_score": self._reports.get("quality_score"),
            "warnings":      self._collect_warnings(),
            "cache_utilise": self._reports.get("cache_utilise", False),
            "details": {
                "source":        self._reports.get("source"),
                "nettoyage":     self._reports.get("nettoyage"),
                "validation":    self._reports.get("validation"),
                "normalisation": self._reports.get("normalisation"),
            },
        }

        return self._df, rapport_final

    def _collect_warnings(self) -> List[str]:
        """Collecte les avertissements depuis les rapports internes.

        Parcourt la liste des validations du rapport de validation pour
        extraire les erreurs de schéma et les expose sous forme de liste
        plate à la racine du rapport final. Cela permet à l'utilisateur
        d'accéder aux avertissements sans naviguer dans la hiérarchie
        interne du rapport.

        Returns:
            list[str]: Liste des messages d'avertissement. Vide si aucune
                étape de validation n'a été exécutée ou si aucune anomalie
                n'a été détectée.
        """
        avertissements: List[str] = []

        # Extraction des erreurs issues du rapport de validation
        rapport_validation = self._reports.get("validation")
        if rapport_validation and isinstance(rapport_validation, dict):
            # 'validations' est une liste de dicts, chacun représentant
            # le résultat d'une vérification (schema, ranges, coordinates…)
            validations = rapport_validation.get("validations", [])
            for entree in validations:
                if not isinstance(entree, dict):
                    continue
                # Seules les validations de schéma portent une clé 'erreurs'
                erreurs = entree.get("erreurs", [])
                if isinstance(erreurs, list) and erreurs:
                    type_validation = entree.get("type", "validation")
                    avertissements.extend(
                        f"[{type_validation}] {msg}" for msg in erreurs
                    )

        return avertissements

    def _exec_step(self, etape: Dict) -> pd.DataFrame:
        """Exécute une étape individuelle du pipeline sur le DataFrame courant.

        Args:
            etape (dict): Dictionnaire décrivant l'étape :
                {'type', 'nom', 'params'}.

        Returns:
            pd.DataFrame: Le DataFrame résultant de l'étape.

        Raises:
            PipelineError: Si la méthode de l'étape est inconnue.
        """
        type_etape = etape["type"]
        nom_methode = etape["nom"]
        params = etape["params"]

        try:
            if type_etape == "cleaning":
                # Instanciation du nettoyeur et appel dynamique de la méthode
                cleaner = Cleaner(self._df)

                if not hasattr(cleaner, nom_methode):
                    raise PipelineError(
                        f"Méthode de nettoyage '{nom_methode}' inconnue. "
                        f"Méthodes disponibles : drop_dupes, fill_missing, "
                        f"drop_outliers, parse_dates, norm_text, strip_chars."
                    )

                # Appel de la méthode avec les paramètres
                resultat = getattr(cleaner, nom_methode)(**params)

                # Certaines méthodes retournent un tuple (df, outliers_df)
                if isinstance(resultat, tuple):
                    self._df = resultat[0]
                else:
                    self._df = resultat

                # Mise à jour cumulée du rapport de nettoyage
                nouveau_rep = cleaner.report()
                if self._reports["nettoyage"] is None:
                    self._reports["nettoyage"] = nouveau_rep
                else:
                    rep_existant = self._reports["nettoyage"]
                    for key in ("doublons_supprimes", "nan_traites", "outliers_detectes", "dates_corrigees"):
                        rep_existant[key] += nouveau_rep.get(key, 0)
                    rep_existant["operations"].extend(nouveau_rep.get("operations", []))
                    rep_existant["lignes_finales"] = len(self._df)
                    rep_existant["colonnes_finales"] = len(self._df.columns)

            elif type_etape == "validation":
                # Validation du schéma et calcul du score qualité
                validator = Validator(self._df)
                est_valide, erreurs = validator.check_schema(params.get("schema", {}))
                score = validator.quality_score()

                # Journalisation du résultat de validation
                if not est_valide:
                    logger.warning(
                        "Validation du schéma : %d erreur(s) détectée(s).", len(erreurs)
                    )
                    for erreur in erreurs:
                        logger.warning("  - %s", erreur)

                nouveau_val_rep = validator.report()
                if self._reports["validation"] is None:
                    self._reports["validation"] = nouveau_val_rep
                else:
                    rep_existant = self._reports["validation"]
                    rep_existant["validations"].extend(nouveau_val_rep.get("validations", []))
                    rep_existant["lignes"] = len(self._df)
                    rep_existant["colonnes"] = len(self._df.columns)
                    rep_existant["quality_score"] = score
                self._reports["quality_score"] = score

            elif type_etape == "normalization":
                # Application des normalisations demandées
                normalizer = Normalizer(self._df)

                # Méthodes disponibles directement par nom via add_step
                _methodes_directes = {
                    "norm_cols", "convert_units", "convert_currency",
                    "std_crops", "std_markets", "std_coords",
                }

                if nom_methode in _methodes_directes:
                    # Dispatch direct : appel de la méthode par son nom
                    if not hasattr(normalizer, nom_methode):
                        raise PipelineError(
                            f"Méthode de normalisation '{nom_methode}' inconnue."
                        )
                    resultat = getattr(normalizer, nom_methode)(**params)
                    # mappings() retourne un dict, les autres retournent un DataFrame
                    if isinstance(resultat, pd.DataFrame):
                        self._df = resultat

                else:
                    # Ancien mécanisme via le dict 'mappings' (rétrocompatibilité)
                    mappings = params.get("mappings", {})

                    # Normalisation des noms de colonnes si demandé
                    if "columns" in mappings or mappings.get("normalize_columns"):
                        normalizer.norm_cols()

                    # Normalisation des noms de cultures si demandé
                    if "crops" in mappings:
                        normalizer.std_crops(col=mappings["crops"])

                    # Normalisation des unités si demandé
                    if "units" in mappings:
                        normalizer.convert_units(unit_map=mappings["units"])

                    # Normalisation des marchés si demandé
                    if "markets" in mappings:
                        normalizer.std_markets(col=mappings["markets"])

                    self._df = normalizer.df

                self._reports["normalisation"] = normalizer.mappings()

        except PipelineError:
            raise
        except Exception as erreur:
            raise PipelineError(
                f"Erreur lors de l'exécution de l'étape '{nom_methode}' : {erreur}"
            ) from erreur

        return self._df

    # Alias interne pour conformité renommage.md
    _run_step = _exec_step

    def config(self) -> dict:
        """Retourne la configuration complète du pipeline (étapes définies).

        Returns:
            dict: Dictionnaire décrivant la source et les étapes configurées.
        """
        steps_list = [
            {"type": e["type"], "nom": e["nom"], "params": e["params"]}
            for e in self._steps
        ]
        return {
            "source": self._reports.get("source"),
            "nb_steps": len(self._steps),
            "nb_etapes": len(self._steps),  # Rétrocompatibilité
            "steps": steps_list,
            "etapes": steps_list,  # Rétrocompatibilité
        }

    def export(
        self,
        filepath: str,
        **kwargs,
    ) -> bool:
        """Exporte le rapport de pipeline dans un fichier JSON ou HTML.

        Args:
            filepath (str): Chemin de destination du rapport. L'extension
                détermine le format : '.json' ou '.html'.

        Returns:
            bool: True si l'export s'est déroulé avec succès.

        Raises:
            PipelineError: Si l'extension n'est pas supportée.
        """
        import json

        _, extension = os.path.splitext(filepath.lower())

        try:
            if extension == ".json":
                # Export au format JSON
                with open(filepath, "w", encoding="utf-8") as f:
                    json.dump(self._reports, f, ensure_ascii=False, indent=2, default=str)

            elif extension == ".html":
                # Export au format HTML simplifié
                contenu_json = json.dumps(
                    self._reports, ensure_ascii=False, indent=2, default=str
                )
                html = (
                    "<html><head><meta charset='utf-8'>"
                    "<title>Rapport kidas Pipeline</title></head>"
                    "<body><h1>Rapport kidas Pipeline</h1>"
                    f"<pre>{contenu_json}</pre></body></html>"
                )
                with open(filepath, "w", encoding="utf-8") as f:
                    f.write(html)

            else:
                raise PipelineError(
                    f"Format d'export '{extension}' non supporté. "
                    f"Utilisez '.json' ou '.html'."
                )

            logger.info("Rapport pipeline exporté vers '%s'.", filepath)
            return True

        except OSError as erreur:
            raise PipelineError(
                f"Impossible d'écrire le rapport vers '{filepath}' : {erreur}"
            ) from erreur

    # --- Méthodes de rétrocompatibilité ---

    def load_data(
        self,
        source: Union[str, Source],
        **kwargs: Any,
    ) -> "Pipeline":
        """Alias de rétrocompatibilité pour add_source()."""
        warnings.warn(
            "Pipeline.load_data() est obsolète et sera supprimé dans KadiPy v2.0. "
            "Utilisez Pipeline.add_source() à la place.",
            category=DeprecationWarning,
            stacklevel=2,
        )
        return self.add_source(source, **kwargs)

    def add_cleaning_step(
        self,
        step_name: str,
        **params: Any,
    ) -> "Pipeline":
        """Alias de rétrocompatibilité pour add_step()."""
        warnings.warn(
            "Pipeline.add_cleaning_step() est obsolète et sera supprimé dans KadiPy v2.0. "
            "Utilisez Pipeline.add_step() à la place.",
            category=DeprecationWarning,
            stacklevel=2,
        )
        return self.add_step(step_name, **params)

    def add_validation_step(
        self,
        schema: Dict[str, str],
    ) -> "Pipeline":
        """Alias de rétrocompatibilité pour add_step('check_schema', schema)."""
        warnings.warn(
            "Pipeline.add_validation_step() est obsolète et sera supprimé dans KadiPy v2.0. "
            "Utilisez Pipeline.add_step() à la place.",
            category=DeprecationWarning,
            stacklevel=2,
        )
        return self.add_step("check_schema", schema=schema)

    def add_normalization_step(
        self,
        mappings: Dict[str, Any],
    ) -> "Pipeline":
        """Alias de rétrocompatibilité pour add_step('mappings', mappings)."""
        warnings.warn(
            "Pipeline.add_normalization_step() est obsolète et sera supprimé dans KadiPy v2.0. "
            "Utilisez Pipeline.add_step() à la place.",
            category=DeprecationWarning,
            stacklevel=2,
        )
        return self.add_step("mappings", mappings=mappings)

    def execute(
        self,
        cache: bool = True,
    ) -> Tuple[pd.DataFrame, Dict]:
        """Alias de rétrocompatibilité pour run()."""
        warnings.warn(
            "Pipeline.execute() est obsolète et sera supprimé dans KadiPy v2.0. "
            "Utilisez Pipeline.run() à la place.",
            category=DeprecationWarning,
            stacklevel=2,
        )
        return self.run(cache=cache)

    def get_pipeline_config(self) -> dict:
        """Alias de rétrocompatibilité pour config()."""
        warnings.warn(
            "Pipeline.get_pipeline_config() est obsolète et sera supprimé dans KadiPy v2.0. "
            "Utilisez Pipeline.config() à la place.",
            category=DeprecationWarning,
            stacklevel=2,
        )
        return self.config()

    def export_report(
        self,
        filepath: str,
        **kwargs,
    ) -> bool:
        """Alias de rétrocompatibilité pour export()."""
        warnings.warn(
            "Pipeline.export_report() est obsolète et sera supprimé dans KadiPy v2.0. "
            "Utilisez Pipeline.export() à la place.",
            category=DeprecationWarning,
            stacklevel=2,
        )
        return self.export(filepath, **kwargs)

init ¶

__init__() -> None

Initialise un pipeline vide prêt à recevoir des étapes de traitement.

Source code in kadi/kidas/pipeline.py

def __init__(self) -> None:
    """Initialise un pipeline vide prêt à recevoir des étapes de traitement."""
    # Source de données (sera configurée par add_source())
    self._source: Optional[Source] = None

    # DataFrame courant (None jusqu'à l'appel de run())
    self._df: Optional[pd.DataFrame] = None

    # Nombre de lignes brutes avant toute transformation
    self._nrows_in: int = 0
    self._lignes_avant: int = 0  # Alias de rétrocompatibilité

    # Liste ordonnée des étapes : {'type', 'nom', 'params'}
    self._steps: List[Dict[str, Any]] = []

    # Rapports agrégés internes des étapes de traitement
    self._reports: Dict[str, Any] = {
        "source": None,
        "nettoyage": None,
        "validation": None,
        "normalisation": None,
        "cache_utilise": False,
    }

    # Instance du cache kidas
    self._cache: Cache = Cache()

add_source ¶

add_source(source: Union[str, PathLike, Source], **kwargs: Any) -> Pipeline

Configure la source de données du pipeline.

Auto-détecte le type de source si un chemin de fichier est fourni, ou utilise directement une instance Source existante.

Parameters:

Name	Type	Description	Default
`source`	`str \| PathLike \| Source`	Chemin vers le fichier, URL de l'API, ou instance Source directement.	required
`**kwargs`	`Any`	Arguments optionnels transmis au constructeur de la Source (ex: encoding, sheet_name, use_dask).	`{}`

Returns:

Name	Type	Description
`Pipeline`	`Pipeline`	L'instance courante (pour le chaînage).

Raises:

Type	Description
`PipelineError`	Si le type de source est indéterminable.

Source code in kadi/kidas/pipeline.py

def add_source(
    self,
    source: Union[str, os.PathLike, Source],
    **kwargs: Any,
) -> "Pipeline":
    """Configure la source de données du pipeline.

    Auto-détecte le type de source si un chemin de fichier est fourni,
    ou utilise directement une instance Source existante.

    Args:
        source (str | os.PathLike | Source): Chemin vers le fichier, URL de l'API,
            ou instance Source directement.
        **kwargs: Arguments optionnels transmis au constructeur de la
            Source (ex: encoding, sheet_name, use_dask).

    Returns:
        Pipeline: L'instance courante (pour le chaînage).

    Raises:
        PipelineError: Si le type de source est indéterminable.
    """
    if isinstance(source, Source):
        # Utilisation directe d'une Source existante
        self._source = source
        type_source = source.kind
    else:
        # Auto-détection et instanciation de la Source appropriée
        type_source = self._detecter_type_source(source)

        if type_source == "csv":
            self._source = CSVSource(source, **kwargs)
        elif type_source == "excel":
            self._source = ExcelSource(source, **kwargs)
        elif type_source == "json":
            self._source = JSONSource(source, **kwargs)
        elif type_source == "netcdf":
            self._source = NetCDFSource(source, **kwargs)
        elif type_source == "api":
            self._source = APISource(source, **kwargs)

    # Enregistrement dans le rapport
    self._reports["source"] = {
        "path": str(source),
        "type": type_source,
    }

    logger.info(
        "Pipeline kidas : source '%s' configurée (type: %s).",
        source,
        type_source,
    )

    return self

add_step ¶

add_step(step_name: str, schema_or_mappings: Any = None, **params: Any) -> Pipeline

Ajoute une étape de traitement à la file du pipeline.

Accepte des étapes de nettoyage, de validation ou de normalisation en un seul point d'entrée. Le type de l'étape est déterminé automatiquement selon le nom passé.

Parameters:

Name	Type	Description	Default
`step_name`	`str`	Nom de l'étape. Valeurs acceptées : - Nettoyage : 'drop_dupes', 'fill_missing', 'drop_outliers', 'parse_dates', 'norm_text', 'strip_chars'. - Validation : 'check_schema', 'check_types', 'check_ranges', 'check_coords', 'check_unique', 'check_fk'. - Normalisation : 'norm_cols', 'convert_units', 'convert_currency', 'std_crops', 'std_markets', 'std_coords'.	required
`schema_or_mappings`	`Any`	Argument positionnel optionnel pour les étapes de validation (schema dict) ou normalisation (mappings dict).	`None`
`**params`	`Any`	Paramètres supplémentaires à passer à la méthode (ex: strategy='mean', cols=['id', 'date']).	`{}`

Returns:

Name	Type	Description
`Pipeline`	`Pipeline`	L'instance courante (pour le chaînage).

Source code in kadi/kidas/pipeline.py

def add_step(
    self,
    step_name: str,
    schema_or_mappings: Any = None,
    **params: Any,
) -> "Pipeline":
    """Ajoute une étape de traitement à la file du pipeline.

    Accepte des étapes de nettoyage, de validation ou de normalisation
    en un seul point d'entrée. Le type de l'étape est déterminé
    automatiquement selon le nom passé.

    Args:
        step_name (str): Nom de l'étape. Valeurs acceptées :
            - Nettoyage : 'drop_dupes', 'fill_missing', 'drop_outliers',
              'parse_dates', 'norm_text', 'strip_chars'.
            - Validation : 'check_schema', 'check_types',
              'check_ranges', 'check_coords', 'check_unique', 'check_fk'.
            - Normalisation : 'norm_cols', 'convert_units',
              'convert_currency', 'std_crops', 'std_markets', 'std_coords'.
        schema_or_mappings: Argument positionnel optionnel pour les étapes
            de validation (schema dict) ou normalisation (mappings dict).
        **params: Paramètres supplémentaires à passer à la méthode
            (ex: strategy='mean', cols=['id', 'date']).

    Returns:
        Pipeline: L'instance courante (pour le chaînage).
    """
    # Détermination automatique du type d'étape
    _cleaning = {
        "drop_dupes", "fill_missing", "drop_outliers",
        "parse_dates", "norm_text", "strip_chars",
        "check_decimals", "report",
        "remove_duplicates", "handle_missing_values", "remove_outliers",
        "fix_dates", "standardize_text", "remove_special_chars",
        "detect_inconsistent_decimals", "get_cleaning_report",
    }
    _validation = {
        "check_schema", "check_types", "check_ranges",
        "check_coords", "check_unique", "check_fk",
        "quality_score", "validate_schema", "validate_types",
        "validate_ranges", "validate_coordinates", "validate_uniqueness",
    }
    _normalisation = {
        "norm_cols", "convert_units", "convert_currency",
        "std_crops", "std_markets", "std_coords", "mappings",
        "normalize_column_names", "normalize_units", "normalize_currencies",
        "normalize_crop_names", "normalize_market_names", "normalize_geometry",
        "get_normalization_mapping", "normalize",
    }

    if step_name in _cleaning:
        type_etape = "cleaning"
    elif step_name in _validation:
        type_etape = "validation"
    elif step_name in _normalisation:
        type_etape = "normalization"
    else:
        type_etape = "cleaning"  # Fallback tolérant

    # Gestion du premier argument positionnel
    if schema_or_mappings is not None:
        if type_etape == "validation":
            params["schema"] = schema_or_mappings
        elif type_etape == "normalization":
            params["mappings"] = schema_or_mappings

    # Enregistrement de l'étape dans la file
    self._steps.append({
        "type": type_etape,
        "nom": step_name,
        "params": params,
    })

    logger.debug(
        "Etape '%s' (type: %s) ajoutée (params: %s).",
        step_name, type_etape, params,
    )

    return self

load ¶

load(source: Union[str, Source], **kwargs: Any) -> Pipeline

Configure la source de données du pipeline (alias court de add_source).

Parameters:

Name	Type	Description	Default
`source`	`str \| Source`	Chemin vers le fichier, URL ou instance Source.	required
`**kwargs`	`Any`	Arguments transmis au constructeur de la Source.	`{}`

Returns:

Name	Type	Description
`Pipeline`	`Pipeline`	L'instance courante pour le chaînage.

Source code in kadi/kidas/pipeline.py

def load(self, source: Union[str, Source], **kwargs: Any) -> "Pipeline":
    """Configure la source de données du pipeline (alias court de add_source).

    Args:
        source (str | Source): Chemin vers le fichier, URL ou instance Source.
        **kwargs: Arguments transmis au constructeur de la Source.

    Returns:
        Pipeline: L'instance courante pour le chaînage.
    """
    return self.add_source(source, **kwargs)

clean ¶

clean(step: str, **params: Any) -> Pipeline

Ajoute une étape de nettoyage au pipeline (alias court de add_step).

Parameters:

Name	Type	Description	Default
`step`	`str`	Nom de l'étape de nettoyage (ex: 'drop_dupes').	required
`**params`	`Any`	Paramètres de l'étape de nettoyage.	`{}`

Returns:

Name	Type	Description
`Pipeline`	`Pipeline`	L'instance courante pour le chaînage.

Source code in kadi/kidas/pipeline.py

def clean(self, step: str, **params: Any) -> "Pipeline":
    """Ajoute une étape de nettoyage au pipeline (alias court de add_step).

    Args:
        step (str): Nom de l'étape de nettoyage (ex: 'drop_dupes').
        **params: Paramètres de l'étape de nettoyage.

    Returns:
        Pipeline: L'instance courante pour le chaînage.
    """
    return self.add_step(step, **params)

validate ¶

validate(schema: Any = None, **params: Any) -> Pipeline

Ajoute une étape de validation au pipeline (alias court de add_step).

Parameters:

Name	Type	Description	Default
`schema`	`dict \| str`	Schéma de validation ou nom de l'étape.	`None`
`**params`	`Any`	Paramètres de validation.	`{}`

Returns:

Name	Type	Description
`Pipeline`	`Pipeline`	L'instance courante pour le chaînage.

Source code in kadi/kidas/pipeline.py

def validate(self, schema: Any = None, **params: Any) -> "Pipeline":
    """Ajoute une étape de validation au pipeline (alias court de add_step).

    Args:
        schema (dict | str): Schéma de validation ou nom de l'étape.
        **params: Paramètres de validation.

    Returns:
        Pipeline: L'instance courante pour le chaînage.
    """
    if isinstance(schema, str):
        return self.add_step(schema, **params)
    return self.add_step("check_schema", schema, **params)

normalize ¶

normalize(mappings: Any = None, **params: Any) -> Pipeline

Ajoute une étape de normalisation au pipeline (alias court de add_step).

Parameters:

Name	Type	Description	Default
`mappings`	`dict \| str`	Mappings de normalisation ou nom de l'étape.	`None`
`**params`	`Any`	Paramètres de normalisation.	`{}`

Returns:

Name	Type	Description
`Pipeline`	`Pipeline`	L'instance courante pour le chaînage.

Source code in kadi/kidas/pipeline.py

def normalize(self, mappings: Any = None, **params: Any) -> "Pipeline":
    """Ajoute une étape de normalisation au pipeline (alias court de add_step).

    Args:
        mappings (dict | str): Mappings de normalisation ou nom de l'étape.
        **params: Paramètres de normalisation.

    Returns:
        Pipeline: L'instance courante pour le chaînage.
    """
    if isinstance(mappings, str):
        return self.add_step(mappings, **params)
    return self.add_step("norm_cols", mappings, **params)

run ¶

run(cache: bool = True) -> Tuple[pd.DataFrame, Dict]

Exécute toutes les étapes configurées du pipeline.

Charge les données depuis la source, applique les étapes de nettoyage, validation et normalisation dans l'ordre, puis met le résultat en cache si demandé.

Parameters:

Name	Type	Description	Default
`cache`	`bool`	Si True, tente de charger depuis le cache avant la lecture et sauvegarde le résultat final. Par défaut True.	`True`

Returns:

Type Description

Tuple[DataFrame, Dict]

tuple[pd.DataFrame, dict]: Tuple contenant : - Le DataFrame traité et prêt à l'emploi. - Le rapport structuré selon la documentation : - nb_rows_in : nombre de lignes brutes chargées. - nb_rows_out : nombre de lignes après traitement. - steps_summary : liste des noms d'étapes appliquées. - quality_score : score qualité (dict) ou None. - warnings : liste des avertissements détectés. - cache_utilise : True si les données viennent du cache. - details : rapports internes (source, nettoyage, validation, normalisation).

Raises:

Type	Description
`PipelineError`	Si aucune source n'a été configurée.
`ReadError`	Si la lecture de la source échoue.

Source code in kadi/kidas/pipeline.py

def run(
    self,
    cache: bool = True,
) -> Tuple[pd.DataFrame, Dict]:
    """Exécute toutes les étapes configurées du pipeline.

    Charge les données depuis la source, applique les étapes de
    nettoyage, validation et normalisation dans l'ordre, puis
    met le résultat en cache si demandé.

    Args:
        cache (bool): Si True, tente de charger depuis le cache avant
            la lecture et sauvegarde le résultat final. Par défaut True.

    Returns:
        tuple[pd.DataFrame, dict]: Tuple contenant :
            - Le DataFrame traité et prêt à l'emploi.
            - Le rapport structuré selon la documentation :
                - ``nb_rows_in``    : nombre de lignes brutes chargées.
                - ``nb_rows_out``   : nombre de lignes après traitement.
                - ``steps_summary`` : liste des noms d'étapes appliquées.
                - ``quality_score`` : score qualité (dict) ou None.
                - ``warnings``      : liste des avertissements détectés.
                - ``cache_utilise`` : True si les données viennent du cache.
                - ``details``       : rapports internes (source, nettoyage,
                  validation, normalisation).

    Raises:
        PipelineError: Si aucune source n'a été configurée.
        ReadError: Si la lecture de la source échoue.
    """
    # Vérification qu'une source a été configurée
    if self._source is None:
        raise PipelineError(
            "Aucune source configurée. Appelez add_source() avant run()."
        )

    # Génération d'une clé de cache sécurisée : hachage SHA-256 du chemin brut,
    # tronqué à 16 caractères hexadécimaux pour éviter les collisions et les
    # erreurs d'encodage avec des chemins contenant des espaces ou des accents.
    _empreinte = hashlib.sha256(
        str(self._source.path).encode("utf-8")
    ).hexdigest()[:16]
    cle_cache = f"pipeline_{_empreinte}"

    # Tentative de chargement depuis le cache
    if cache:
        df_cached, _ = self._cache.get(cle_cache)
        if df_cached is not None:
            logger.info(
                "Pipeline kidas : données chargées depuis le cache (clé: '%s').",
                cle_cache,
            )
            # Construction du rapport minimal pour un retour depuis le cache
            rapport_cache = {
                "nb_rows_in":    len(df_cached),
                "nb_rows_out":   len(df_cached),
                "steps_summary": [e["nom"] for e in self._steps],
                "quality_score": None,
                "warnings":      [],
                "cache_utilise": True,
                "details": {
                    "source":        self._reports.get("source"),
                    "nettoyage":     None,
                    "validation":    None,
                    "normalisation": None,
                },
            }
            return df_cached, rapport_cache

    # Lecture des données depuis la source et mémorisation du nombre de lignes brutes
    try:
        self._df = self._source.read()
        # Capture du nombre de lignes avant tout traitement
        self._lignes_avant = len(self._df)
        logger.info(
            "Pipeline kidas : %d lignes chargées depuis '%s'.",
            self._lignes_avant,
            self._source.path,
        )
    except Exception as erreur:
        raise ReadError(
            f"Echec de lecture dans le pipeline : {erreur}"
        ) from erreur

    # Exécution des étapes dans l'ordre
    for etape in self._steps:
        self._df = self._exec_step(etape)

    # Sauvegarde en cache du résultat final
    if cache and self._df is not None:
        self._cache.set(cle_cache, self._df)
        self._reports["cache_utilise"] = True

    # Calcul du nombre de lignes en sortie
    nb_lignes_out = len(self._df) if self._df is not None else 0

    # Construction du rapport final aligné sur la documentation publique
    rapport_final = {
        "nb_rows_in":    self._lignes_avant,
        "nb_rows_out":   nb_lignes_out,
        "steps_summary": [e["nom"] for e in self._steps],
        "quality_score": self._reports.get("quality_score"),
        "warnings":      self._collect_warnings(),
        "cache_utilise": self._reports.get("cache_utilise", False),
        "details": {
            "source":        self._reports.get("source"),
            "nettoyage":     self._reports.get("nettoyage"),
            "validation":    self._reports.get("validation"),
            "normalisation": self._reports.get("normalisation"),
        },
    }

    return self._df, rapport_final

config ¶

config() -> dict

Retourne la configuration complète du pipeline (étapes définies).

Returns:

Name	Type	Description
`dict`	`dict`	Dictionnaire décrivant la source et les étapes configurées.

Source code in kadi/kidas/pipeline.py

def config(self) -> dict:
    """Retourne la configuration complète du pipeline (étapes définies).

    Returns:
        dict: Dictionnaire décrivant la source et les étapes configurées.
    """
    steps_list = [
        {"type": e["type"], "nom": e["nom"], "params": e["params"]}
        for e in self._steps
    ]
    return {
        "source": self._reports.get("source"),
        "nb_steps": len(self._steps),
        "nb_etapes": len(self._steps),  # Rétrocompatibilité
        "steps": steps_list,
        "etapes": steps_list,  # Rétrocompatibilité
    }

export ¶

export(filepath: str, **kwargs) -> bool

Exporte le rapport de pipeline dans un fichier JSON ou HTML.

Parameters:

Name	Type	Description	Default
`filepath`	`str`	Chemin de destination du rapport. L'extension détermine le format : '.json' ou '.html'.	required

Returns:

Name	Type	Description
`bool`	`bool`	True si l'export s'est déroulé avec succès.

Raises:

Type	Description
`PipelineError`	Si l'extension n'est pas supportée.

Source code in kadi/kidas/pipeline.py

def export(
    self,
    filepath: str,
    **kwargs,
) -> bool:
    """Exporte le rapport de pipeline dans un fichier JSON ou HTML.

    Args:
        filepath (str): Chemin de destination du rapport. L'extension
            détermine le format : '.json' ou '.html'.

    Returns:
        bool: True si l'export s'est déroulé avec succès.

    Raises:
        PipelineError: Si l'extension n'est pas supportée.
    """
    import json

    _, extension = os.path.splitext(filepath.lower())

    try:
        if extension == ".json":
            # Export au format JSON
            with open(filepath, "w", encoding="utf-8") as f:
                json.dump(self._reports, f, ensure_ascii=False, indent=2, default=str)

        elif extension == ".html":
            # Export au format HTML simplifié
            contenu_json = json.dumps(
                self._reports, ensure_ascii=False, indent=2, default=str
            )
            html = (
                "<html><head><meta charset='utf-8'>"
                "<title>Rapport kidas Pipeline</title></head>"
                "<body><h1>Rapport kidas Pipeline</h1>"
                f"<pre>{contenu_json}</pre></body></html>"
            )
            with open(filepath, "w", encoding="utf-8") as f:
                f.write(html)

        else:
            raise PipelineError(
                f"Format d'export '{extension}' non supporté. "
                f"Utilisez '.json' ou '.html'."
            )

        logger.info("Rapport pipeline exporté vers '%s'.", filepath)
        return True

    except OSError as erreur:
        raise PipelineError(
            f"Impossible d'écrire le rapport vers '{filepath}' : {erreur}"
        ) from erreur

load_data ¶

load_data(source: Union[str, Source], **kwargs: Any) -> Pipeline

Alias de rétrocompatibilité pour add_source().

Source code in kadi/kidas/pipeline.py

def load_data(
    self,
    source: Union[str, Source],
    **kwargs: Any,
) -> "Pipeline":
    """Alias de rétrocompatibilité pour add_source()."""
    warnings.warn(
        "Pipeline.load_data() est obsolète et sera supprimé dans KadiPy v2.0. "
        "Utilisez Pipeline.add_source() à la place.",
        category=DeprecationWarning,
        stacklevel=2,
    )
    return self.add_source(source, **kwargs)

add_cleaning_step ¶

add_cleaning_step(step_name: str, **params: Any) -> Pipeline

Alias de rétrocompatibilité pour add_step().

Source code in kadi/kidas/pipeline.py

def add_cleaning_step(
    self,
    step_name: str,
    **params: Any,
) -> "Pipeline":
    """Alias de rétrocompatibilité pour add_step()."""
    warnings.warn(
        "Pipeline.add_cleaning_step() est obsolète et sera supprimé dans KadiPy v2.0. "
        "Utilisez Pipeline.add_step() à la place.",
        category=DeprecationWarning,
        stacklevel=2,
    )
    return self.add_step(step_name, **params)

add_validation_step ¶

add_validation_step(schema: Dict[str, str]) -> Pipeline

Alias de rétrocompatibilité pour add_step('check_schema', schema).

Source code in kadi/kidas/pipeline.py

def add_validation_step(
    self,
    schema: Dict[str, str],
) -> "Pipeline":
    """Alias de rétrocompatibilité pour add_step('check_schema', schema)."""
    warnings.warn(
        "Pipeline.add_validation_step() est obsolète et sera supprimé dans KadiPy v2.0. "
        "Utilisez Pipeline.add_step() à la place.",
        category=DeprecationWarning,
        stacklevel=2,
    )
    return self.add_step("check_schema", schema=schema)

add_normalization_step ¶

add_normalization_step(mappings: Dict[str, Any]) -> Pipeline

Alias de rétrocompatibilité pour add_step('mappings', mappings).

Source code in kadi/kidas/pipeline.py

def add_normalization_step(
    self,
    mappings: Dict[str, Any],
) -> "Pipeline":
    """Alias de rétrocompatibilité pour add_step('mappings', mappings)."""
    warnings.warn(
        "Pipeline.add_normalization_step() est obsolète et sera supprimé dans KadiPy v2.0. "
        "Utilisez Pipeline.add_step() à la place.",
        category=DeprecationWarning,
        stacklevel=2,
    )
    return self.add_step("mappings", mappings=mappings)

execute ¶

execute(cache: bool = True) -> Tuple[pd.DataFrame, Dict]

Alias de rétrocompatibilité pour run().

Source code in kadi/kidas/pipeline.py

def execute(
    self,
    cache: bool = True,
) -> Tuple[pd.DataFrame, Dict]:
    """Alias de rétrocompatibilité pour run()."""
    warnings.warn(
        "Pipeline.execute() est obsolète et sera supprimé dans KadiPy v2.0. "
        "Utilisez Pipeline.run() à la place.",
        category=DeprecationWarning,
        stacklevel=2,
    )
    return self.run(cache=cache)

get_pipeline_config ¶

get_pipeline_config() -> dict

Alias de rétrocompatibilité pour config().

Source code in kadi/kidas/pipeline.py

def get_pipeline_config(self) -> dict:
    """Alias de rétrocompatibilité pour config()."""
    warnings.warn(
        "Pipeline.get_pipeline_config() est obsolète et sera supprimé dans KadiPy v2.0. "
        "Utilisez Pipeline.config() à la place.",
        category=DeprecationWarning,
        stacklevel=2,
    )
    return self.config()

export_report ¶

export_report(filepath: str, **kwargs) -> bool

Alias de rétrocompatibilité pour export().

Source code in kadi/kidas/pipeline.py

def export_report(
    self,
    filepath: str,
    **kwargs,
) -> bool:
    """Alias de rétrocompatibilité pour export()."""
    warnings.warn(
        "Pipeline.export_report() est obsolète et sera supprimé dans KadiPy v2.0. "
        "Utilisez Pipeline.export() à la place.",
        category=DeprecationWarning,
        stacklevel=2,
    )
    return self.export(filepath, **kwargs)

Pipeline (kadi.kidas.pipeline)¶

Initialization¶

Configuration Methods¶

add_source(source, **kwargs)¶

add_step(step_name, schema_or_mappings=None, **params)¶

Available Cleaning Steps¶

Available Validation Steps¶

Available Normalization Steps¶

Execution¶

run(cache=True)¶

Returned Report Structure¶

Utility Methods¶

config()¶

export(filepath)¶

Full Examples¶

Minimal Pipeline¶

Pipeline with Validation and Normalization¶

Fluid API with Short Aliases¶

Reusing an Existing Source¶

Report Export¶

Cache Behavior¶

Backward Compatibility¶

Pipeline ¶

__init__ ¶

add_source ¶

add_step ¶

load ¶

clean ¶

validate ¶

normalize ¶

run ¶

config ¶

export ¶

load_data ¶

add_cleaning_step ¶

add_validation_step ¶

add_normalization_step ¶

execute ¶

get_pipeline_config ¶

export_report ¶

Pipeline (`kadi.kidas.pipeline`)¶

`add_source(source, **kwargs)`¶

`add_step(step_name, schema_or_mappings=None, **params)`¶

`run(cache=True)`¶

`config()`¶

`export(filepath)`¶

init ¶