IngestionPlugin

Purpose: Data loading from external sources Location: packages/floe-core/src/floe_core/plugins/ingestion.py Entry Point: floe.ingestion ADR: ADR-0020: Ingestion Plugins

IngestionPlugin abstracts data ingestion/EL (Extract-Load) tools, enabling platform teams to choose between embedded (dlt) or external (Airbyte) ingestion patterns while maintaining consistent pipeline definitions.

Interface Definition

The live ABC is IngestionPlugin in packages/floe-core/src/floe_core/plugins/ingestion.py. The snippet below is a conceptual excerpt for the public contract.

# Conceptual excerpt; see packages/floe-core/src/floe_core/plugins/ingestion.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any

@dataclass
class IngestionConfig:
    """Configuration for an ingestion pipeline."""
    name: str                           # Pipeline name
    source: str                         # Source identifier
    destination_table: str              # Target Iceberg table
    write_disposition: str              # append | replace | merge
    incremental: dict | None            # Incremental config
    secret_refs: dict[str, str] | None  # Secret references

@dataclass
class IngestionResult:
    """Result of an ingestion run."""
    success: bool
    rows_loaded: int
    tables_created: list[str]
    duration_seconds: float
    error: str | None

class IngestionPlugin(ABC):
    """Interface for data ingestion/EL plugins (dlt, Airbyte)."""

    name: str
    version: str
    is_external: bool

    @abstractmethod
    def create_pipeline(self, config: IngestionConfig) -> Any:
        """Create ingestion pipeline from data-product source configuration."""
        pass

    @abstractmethod
    def run(self, pipeline: Any, **kwargs) -> IngestionResult:
        """Execute the ingestion pipeline."""
        pass

    def get_composition_requirements(self) -> Any:
        """Declare storage/catalog requirements for platform composition."""
        return None

    def build_deployment_binding(self, *, storage: Any, catalog: Any) -> Any:
        """Build a secret-free deployment binding from composed platform state."""
        raise NotImplementedError

Reference Implementations

Plugin	Description	External
DLTIngestionPlugin	Embedded Python ingestion (dlt)	No
AirbyteIngestionPlugin	External connector platform	Yes

Related Documents

• ADR-0020: Ingestion Plugins
• Plugin Architecture
• CatalogPlugin - For destination table registration