Pipeline Dependency Injection: Decoupling Source and Sink Connectors for Testable, Swappable Data Pipeline Components

Introduction

Here's a situation you've probably lived through. Your pipeline reads Salesforce leads, transforms them, and writes to a Snowflake table. It works beautifully in production. Then your team asks you to run it against a staging environment for QA, and suddenly you're commenting out database credentials, hardcoding file paths, and praying you don't accidentally push to prod. Then someone asks you to write a unit test, and you realize the pipeline is one monolithic function that opens a database connection on line three and you have no idea how to mock any of it.

This is the problem that dependency injection solves in software engineering — and it applies to data pipelines just as directly as it does to any other software. When your pipeline tightly couples its logic to specific data sources and destinations, you've accidentally made your pipeline a closed system. Swapping a Snowflake sink for a Parquet file, or replacing a Salesforce source with a CSV fixture for testing, requires surgery on the core logic instead of a simple configuration swap.

By the end of this lesson, you'll understand how to architect pipelines using dependency injection (DI) patterns so that your source connectors and sink connectors are interchangeable components, your transformation logic is testable in complete isolation, and your pipeline can be promoted across environments without code changes.

What you'll learn:

• What dependency injection means in the context of data pipelines and why it matters
• How to define abstract source and sink interfaces that your pipeline logic depends on
• How to implement concrete connectors (Snowflake, S3, Postgres, file-based) that satisfy those interfaces
• How to wire dependencies together using factory functions and runtime configuration
• How to write fast, isolated unit tests for transformation logic without touching any real infrastructure

---

Prerequisites

You should be comfortable with Python at an intermediate level — classes, inheritance, and type hints. You've built at least one ETL or ELT pipeline that reads from a real source (a database, API, or object store) and writes to a destination. Familiarity with pytest is helpful for the testing sections. You don't need a background in formal software design patterns, but it'll help if you've at least heard terms like "interface" or "abstract class."

---

Why Pipelines Become Untestable (And How It Happens Quietly)

Let's look at how a typical pipeline evolves into a dependency nightmare. It starts reasonable enough:

import psycopg2
import snowflake.connector
import pandas as pd

def run_pipeline():
    # Extract
    pg_conn = psycopg2.connect(
        host="prod-postgres.internal",
        database="crm",
        user="pipeline_user",
        password="supersecret"
    )
    df = pd.read_sql("SELECT * FROM leads WHERE created_at > NOW() - INTERVAL '1 day'", pg_conn)
    
    # Transform
    df["full_name"] = df["first_name"] + " " + df["last_name"]
    df["lead_score"] = df["page_views"] * 0.4 + df["email_opens"] * 0.6
    df = df[df["lead_score"] > 10]
    
    # Load
    sf_conn = snowflake.connector.connect(
        account="mycompany.us-east-1",
        user="etl_user",
        password="anothersecret",
        warehouse="TRANSFORM_WH",
        database="ANALYTICS",
        schema="MARKETING"
    )
    # write to Snowflake...

This pipeline is doing three things at once: managing infrastructure connections, encoding business logic (the scoring formula), and handling the mechanics of writing to a destination. Everything is entangled.

When you try to test the lead_score calculation, you can't. There's no way to run just that logic without spinning up a Postgres instance populated with yesterday's data and a Snowflake warehouse. When your company decides to migrate from Postgres to MySQL, you're rewriting the core pipeline. When QA asks for a staging run against a test database, you're looking for all the places credentials are hardcoded.

The cure is dependency injection: instead of your pipeline creating its dependencies internally, it receives them from the outside.

---

The Core Pattern: Depend on Abstractions, Not Concretions

The principle here comes from the SOLID design principles — specifically the Dependency Inversion Principle: high-level modules (your transformation logic) should not depend on low-level modules (Postgres, Snowflake). Both should depend on abstractions.

In Python, we express abstractions using Abstract Base Classes (ABCs) from the abc module. Let's define what a source connector and a sink connector should look like, without prescribing what they are.

from abc import ABC, abstractmethod
import pandas as pd
from typing import Optional

class SourceConnector(ABC):
    """
    Abstract interface for any data source in the pipeline.
    Concrete implementations handle the specifics of connecting
    to Postgres, Salesforce, S3, a local file, etc.
    """
    
    @abstractmethod
    def extract(self, query: Optional[str] = None) -> pd.DataFrame:
        """
        Pull data from the source and return a DataFrame.
        The query parameter is intentionally flexible — it might be
        a SQL string, a SOQL query, an S3 prefix, or ignored entirely.
        """
        pass
    
    @abstractmethod
    def health_check(self) -> bool:
        """
        Verify the source is reachable before starting.
        Returns True if healthy, False otherwise.
        """
        pass


class SinkConnector(ABC):
    """
    Abstract interface for any data destination in the pipeline.
    Concrete implementations handle Snowflake, BigQuery, Parquet files, etc.
    """
    
    @abstractmethod
    def load(self, df: pd.DataFrame, destination: str) -> int:
        """
        Write the DataFrame to the destination.
        Returns the number of records written.
        """
        pass
    
    @abstractmethod
    def health_check(self) -> bool:
        """Verify the sink is reachable before starting."""
        pass

These ABCs are contracts. Any class that inherits from SourceConnector and implements extract and health_check is a valid source for your pipeline. It doesn't matter if it reads from Postgres or a pickle file — as far as your transformation logic is concerned, they're identical.

Why use ABCs instead of just duck typing? Python's duck typing would technically let you pass any object with an extract method. ABCs enforce the contract at class definition time — if a subclass forgets to implement load, Python raises a TypeError when you try to instantiate it, not when your pipeline runs at 2am and tries to write 10 million rows.

---

Building Concrete Connectors

Now let's implement real connectors that satisfy these interfaces. We'll build four: one for Postgres, one for Snowflake, one for S3/Parquet, and one in-memory connector for testing.

The Postgres Source Connector

import psycopg2
import pandas as pd
from dataclasses import dataclass
from typing import Optional

@dataclass
class PostgresConfig:
    host: str
    port: int
    database: str
    user: str
    password: str
    connect_timeout: int = 10

class PostgresSourceConnector(SourceConnector):
    
    def __init__(self, config: PostgresConfig):
        self.config = config
        self._connection = None
    
    def _get_connection(self):
        if self._connection is None or self._connection.closed:
            self._connection = psycopg2.connect(
                host=self.config.host,
                port=self.config.port,
                database=self.config.database,
                user=self.config.user,
                password=self.config.password,
                connect_timeout=self.config.connect_timeout
            )
        return self._connection
    
    def extract(self, query: Optional[str] = None) -> pd.DataFrame:
        if query is None:
            raise ValueError("PostgresSourceConnector requires a SQL query string.")
        
        conn = self._get_connection()
        try:
            df = pd.read_sql_query(query, conn)
            return df
        except Exception as e:
            # Close the connection so _get_connection reconnects next time
            if self._connection:
                self._connection.close()
            raise RuntimeError(f"Failed to extract from Postgres: {e}") from e
    
    def health_check(self) -> bool:
        try:
            conn = self._get_connection()
            cursor = conn.cursor()
            cursor.execute("SELECT 1")
            cursor.close()
            return True
        except Exception:
            return False
    
    def __del__(self):
        if self._connection and not self._connection.closed:
            self._connection.close()

Notice that PostgresConfig is a dataclass — all the configuration lives there, separate from the logic. This makes it easy to swap configs without touching the connector class itself.

The Snowflake Sink Connector

import snowflake.connector
from snowflake.connector.pandas_tools import write_pandas
from dataclasses import dataclass
from typing import Optional

@dataclass
class SnowflakeConfig:
    account: str
    user: str
    password: str
    warehouse: str
    database: str
    schema: str
    role: Optional[str] = None

class SnowflakeSinkConnector(SinkConnector):
    
    def __init__(self, config: SnowflakeConfig):
        self.config = config
        self._connection = None
    
    def _get_connection(self):
        if self._connection is None:
            connect_params = {
                "account": self.config.account,
                "user": self.config.user,
                "password": self.config.password,
                "warehouse": self.config.warehouse,
                "database": self.config.database,
                "schema": self.config.schema,
            }
            if self.config.role:
                connect_params["role"] = self.config.role
            self._connection = snowflake.connector.connect(**connect_params)
        return self._connection
    
    def load(self, df: pd.DataFrame, destination: str) -> int:
        """
        destination: the target table name, e.g. 'QUALIFIED_LEADS'
        Uses Snowflake's write_pandas for efficient bulk loading.
        """
        conn = self._get_connection()
        
        # write_pandas requires uppercase column names to match Snowflake's convention
        df = df.copy()
        df.columns = [col.upper() for col in df.columns]
        
        success, nchunks, nrows, _ = write_pandas(
            conn=conn,
            df=df,
            table_name=destination.upper(),
            auto_create_table=False,
            overwrite=False
        )
        
        if not success:
            raise RuntimeError(f"Snowflake write_pandas reported failure for table {destination}")
        
        return nrows
    
    def health_check(self) -> bool:
        try:
            conn = self._get_connection()
            cursor = conn.cursor()
            cursor.execute("SELECT CURRENT_VERSION()")
            cursor.close()
            return True
        except Exception:
            return False

The S3/Parquet Sink Connector

Sometimes you want to land data as Parquet files in S3 rather than loading directly into a warehouse. This is a completely valid sink that your pipeline shouldn't need to know about:

import boto3
import pandas as pd
import io
from dataclasses import dataclass

@dataclass
class S3Config:
    bucket: str
    prefix: str
    region: str = "us-east-1"
    aws_access_key_id: Optional[str] = None
    aws_secret_access_key: Optional[str] = None

class S3ParquetSinkConnector(SinkConnector):
    
    def __init__(self, config: S3Config):
        self.config = config
        self._client = None
    
    def _get_client(self):
        if self._client is None:
            session_kwargs = {"region_name": self.config.region}
            if self.config.aws_access_key_id:
                session_kwargs["aws_access_key_id"] = self.config.aws_access_key_id
                session_kwargs["aws_secret_access_key"] = self.config.aws_secret_access_key
            self._client = boto3.client("s3", **session_kwargs)
        return self._client
    
    def load(self, df: pd.DataFrame, destination: str) -> int:
        """
        destination: used as the filename within the configured prefix.
        e.g., destination='qualified_leads_20240115' results in
        s3://bucket/prefix/qualified_leads_20240115.parquet
        """
        buffer = io.BytesIO()
        df.to_parquet(buffer, index=False, engine="pyarrow")
        buffer.seek(0)
        
        key = f"{self.config.prefix.rstrip('/')}/{destination}.parquet"
        client = self._get_client()
        client.put_object(
            Bucket=self.config.bucket,
            Key=key,
            Body=buffer.getvalue()
        )
        
        return len(df)
    
    def health_check(self) -> bool:
        try:
            client = self._get_client()
            client.head_bucket(Bucket=self.config.bucket)
            return True
        except Exception:
            return False

---

The In-Memory Connectors: Your Testing Best Friends

Here's where dependency injection pays off most visibly. For testing, we create connectors that work entirely in memory — no network, no credentials, no infrastructure required:

class InMemorySourceConnector(SourceConnector):
    """
    A source connector that returns a pre-loaded DataFrame.
    Use this in tests to inject known fixture data.
    """
    
    def __init__(self, data: pd.DataFrame):
        self._data = data
    
    def extract(self, query: Optional[str] = None) -> pd.DataFrame:
        # Optionally, you could interpret `query` as a pandas filter expression,
        # but for most unit tests, returning the fixture as-is is sufficient.
        return self._data.copy()
    
    def health_check(self) -> bool:
        return True


class InMemorySinkConnector(SinkConnector):
    """
    A sink connector that stores written data in a dict.
    Use this in tests to inspect what the pipeline tried to write.
    """
    
    def __init__(self):
        self.written_data: dict[str, pd.DataFrame] = {}
        self.write_count: dict[str, int] = {}
    
    def load(self, df: pd.DataFrame, destination: str) -> int:
        self.written_data[destination] = df.copy()
        self.write_count[destination] = self.write_count.get(destination, 0) + 1
        return len(df)
    
    def health_check(self) -> bool:
        return True

InMemorySinkConnector.written_data is a dictionary where your tests can inspect exactly what the pipeline would have written, without ever touching Snowflake.

---

Wiring It All Together: The Pipeline Class

Now we build the pipeline itself. The key move: the pipeline's __init__ method receives its source and sink, rather than creating them:

import logging
from datetime import datetime, timedelta

logger = logging.getLogger(__name__)

class LeadScoringPipeline:
    """
    Extracts raw lead data, computes lead scores, filters unqualified leads,
    and loads the results to a configured destination.
    
    The source and sink are injected at construction time, making this
    pipeline testable and environment-agnostic.
    """
    
    def __init__(
        self,
        source: SourceConnector,
        sink: SinkConnector,
        score_threshold: float = 10.0,
        destination_table: str = "QUALIFIED_LEADS"
    ):
        self.source = source
        self.sink = sink
        self.score_threshold = score_threshold
        self.destination_table = destination_table
    
    def _build_query(self, lookback_hours: int = 24) -> str:
        since = datetime.utcnow() - timedelta(hours=lookback_hours)
        return f"""
            SELECT
                lead_id,
                first_name,
                last_name,
                email,
                company,
                page_views,
                email_opens,
                demo_requested,
                created_at
            FROM leads
            WHERE created_at >= '{since.strftime('%Y-%m-%d %H:%M:%S')}'
        """
    
    def _transform(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        All business logic lives here, completely isolated from
        any infrastructure concerns.
        """
        if df.empty:
            logger.warning("Source returned empty DataFrame. Nothing to transform.")
            return df
        
        # Derive fields
        df = df.copy()
        df["full_name"] = df["first_name"].str.strip() + " " + df["last_name"].str.strip()
        
        # Compute lead score
        df["lead_score"] = (
            df["page_views"] * 0.4 +
            df["email_opens"] * 0.6 +
            df["demo_requested"].astype(int) * 15.0
        )
        
        # Filter to qualified leads only
        qualified = df[df["lead_score"] >= self.score_threshold].copy()
        
        # Select output columns
        output_cols = ["lead_id", "full_name", "email", "company", "lead_score", "created_at"]
        qualified = qualified[output_cols]
        
        logger.info(
            f"Transform complete: {len(df)} records in, "
            f"{len(qualified)} qualified leads out "
            f"(threshold={self.score_threshold})"
        )
        
        return qualified
    
    def run(self, lookback_hours: int = 24) -> dict:
        """
        Execute the full pipeline. Returns a run summary dict.
        """
        run_summary = {
            "status": "unknown",
            "records_extracted": 0,
            "records_loaded": 0,
            "errors": []
        }
        
        # Pre-flight health checks
        if not self.source.health_check():
            raise RuntimeError("Source health check failed. Aborting pipeline.")
        if not self.sink.health_check():
            raise RuntimeError("Sink health check failed. Aborting pipeline.")
        
        try:
            # Extract
            query = self._build_query(lookback_hours)
            raw_df = self.source.extract(query)
            run_summary["records_extracted"] = len(raw_df)
            logger.info(f"Extracted {len(raw_df)} records from source.")
            
            # Transform
            transformed_df = self._transform(raw_df)
            
            # Load
            if transformed_df.empty:
                logger.info("No qualified leads to load. Pipeline complete.")
                run_summary["status"] = "success_no_data"
            else:
                records_written = self.sink.load(transformed_df, self.destination_table)
                run_summary["records_loaded"] = records_written
                run_summary["status"] = "success"
                logger.info(f"Loaded {records_written} records to {self.destination_table}.")
        
        except Exception as e:
            run_summary["status"] = "failed"
            run_summary["errors"].append(str(e))
            logger.error(f"Pipeline failed: {e}", exc_info=True)
            raise
        
        return run_summary

The _transform method is now a pure function of its input. It has no idea whether that input came from Postgres or a pickle file. This is the payoff.

---

Factory Functions: Assembling the Right Pipeline for Each Environment

You need a way to build the correct pipeline for each environment (development, staging, production) without scattering conditional logic everywhere. A factory function reads from configuration and assembles the right combination of connectors:

import os
from enum import Enum

class Environment(str, Enum):
    DEVELOPMENT = "development"
    STAGING = "staging"
    PRODUCTION = "production"

def build_pipeline(env: Environment) -> LeadScoringPipeline:
    """
    Factory function that constructs the appropriate pipeline for the environment.
    Configuration is read from environment variables, which are set by your
    deployment system (e.g., Kubernetes secrets, AWS Parameter Store, etc.)
    """
    
    if env == Environment.DEVELOPMENT:
        # In development, read from a local CSV fixture and write to a local Parquet file
        import pandas as pd
        fixture_df = pd.read_csv("tests/fixtures/leads_sample.csv")
        source = InMemorySourceConnector(fixture_df)
        sink = S3ParquetSinkConnector(
            S3Config(
                bucket=os.environ["DEV_S3_BUCKET"],
                prefix="pipeline-output/dev/leads"
            )
        )
    
    elif env == Environment.STAGING:
        source = PostgresSourceConnector(
            PostgresConfig(
                host=os.environ["STAGING_PG_HOST"],
                port=int(os.environ.get("STAGING_PG_PORT", "5432")),
                database=os.environ["STAGING_PG_DATABASE"],
                user=os.environ["STAGING_PG_USER"],
                password=os.environ["STAGING_PG_PASSWORD"]
            )
        )
        sink = SnowflakeSinkConnector(
            SnowflakeConfig(
                account=os.environ["STAGING_SF_ACCOUNT"],
                user=os.environ["STAGING_SF_USER"],
                password=os.environ["STAGING_SF_PASSWORD"],
                warehouse="STAGING_WH",
                database="ANALYTICS_STAGING",
                schema="MARKETING"
            )
        )
    
    elif env == Environment.PRODUCTION:
        source = PostgresSourceConnector(
            PostgresConfig(
                host=os.environ["PROD_PG_HOST"],
                port=int(os.environ.get("PROD_PG_PORT", "5432")),
                database=os.environ["PROD_PG_DATABASE"],
                user=os.environ["PROD_PG_USER"],
                password=os.environ["PROD_PG_PASSWORD"]
            )
        )
        sink = SnowflakeSinkConnector(
            SnowflakeConfig(
                account=os.environ["PROD_SF_ACCOUNT"],
                user=os.environ["PROD_SF_USER"],
                password=os.environ["PROD_SF_PASSWORD"],
                warehouse="TRANSFORM_WH",
                database="ANALYTICS",
                schema="MARKETING",
                role="ETL_ROLE"
            )
        )
    
    else:
        raise ValueError(f"Unknown environment: {env}")
    
    return LeadScoringPipeline(
        source=source,
        sink=sink,
        score_threshold=float(os.environ.get("LEAD_SCORE_THRESHOLD", "10.0")),
        destination_table=os.environ.get("DESTINATION_TABLE", "QUALIFIED_LEADS")
    )


# Entry point
if __name__ == "__main__":
    env_name = os.environ.get("PIPELINE_ENV", "development")
    pipeline = build_pipeline(Environment(env_name))
    summary = pipeline.run(lookback_hours=24)
    print(summary)

Setting PIPELINE_ENV=staging in your deployment config gives you staging behavior with zero code changes.

---

Writing Real Tests

This is where the architecture proves its worth. Your test suite should be fast, deterministic, and require no network access whatsoever.

# tests/test_lead_scoring_pipeline.py

import pytest
import pandas as pd
import numpy as np
from pipeline import LeadScoringPipeline, InMemorySourceConnector, InMemorySinkConnector


@pytest.fixture
def sample_leads():
    """Realistic fixture data representing a day's worth of inbound leads."""
    return pd.DataFrame({
        "lead_id": [1001, 1002, 1003, 1004, 1005],
        "first_name": ["Sarah", "Marcus", "Priya", "Tom", "Elena"],
        "last_name": ["Chen", "Williams", "Sharma", "Bradley", "Kowalski"],
        "email": [
            "s.chen@techcorp.com",
            "mwilliams@retailco.com",
            "priya@startupxyz.com",
            "tbradley@enterprise.io",
            "e.kowalski@agency.co"
        ],
        "company": ["TechCorp", "RetailCo", "StartupXYZ", "Enterprise IO", "Agency Co"],
        "page_views": [12, 2, 25, 8, 1],
        "email_opens": [5, 1, 10, 3, 0],
        "demo_requested": [True, False, True, False, False],
        "created_at": pd.date_range("2024-01-15 09:00:00", periods=5, freq="2h")
    })


@pytest.fixture
def pipeline(sample_leads):
    """
    A fully wired pipeline using in-memory connectors.
    No databases, no S3, no credentials needed.
    """
    source = InMemorySourceConnector(sample_leads)
    sink = InMemorySinkConnector()
    return LeadScoringPipeline(
        source=source,
        sink=sink,
        score_threshold=10.0,
        destination_table="QUALIFIED_LEADS"
    )


class TestLeadScoring:
    
    def test_lead_score_calculation(self, pipeline, sample_leads):
        """
        Sarah: 12 * 0.4 + 5 * 0.6 + 1 * 15 = 4.8 + 3.0 + 15 = 22.8 (qualified)
        Marcus: 2 * 0.4 + 1 * 0.6 + 0 * 15 = 0.8 + 0.6 = 1.4 (not qualified)
        Priya: 25 * 0.4 + 10 * 0.6 + 1 * 15 = 10 + 6 + 15 = 31.0 (qualified)
        """
        pipeline.run()
        sink = pipeline.sink
        
        result = sink.written_data["QUALIFIED_LEADS"]
        written_emails = set(result["email"].tolist())
        
        assert "s.chen@techcorp.com" in written_emails, "Sarah should be qualified"
        assert "priya@startupxyz.com" in written_emails, "Priya should be qualified"
        assert "mwilliams@retailco.com" not in written_emails, "Marcus should not be qualified"
        assert "e.kowalski@agency.co" not in written_emails, "Elena should not be qualified"
    
    def test_full_name_derivation(self, pipeline):
        pipeline.run()
        result = pipeline.sink.written_data["QUALIFIED_LEADS"]
        assert "Sarah Chen" in result["full_name"].values
    
    def test_empty_source_does_not_write(self):
        """If the source is empty, the sink should never be called."""
        source = InMemorySourceConnector(pd.DataFrame())
        sink = InMemorySinkConnector()
        p = LeadScoringPipeline(source=source, sink=sink, score_threshold=10.0)
        
        summary = p.run()
        
        assert summary["status"] == "success_no_data"
        assert "QUALIFIED_LEADS" not in sink.written_data
    
    def test_all_leads_below_threshold_produces_no_write(self, sample_leads):
        """With a very high threshold, nothing should be written."""
        source = InMemorySourceConnector(sample_leads)
        sink = InMemorySinkConnector()
        p = LeadScoringPipeline(source=source, sink=sink, score_threshold=999.0)
        
        p.run()
        
        assert "QUALIFIED_LEADS" not in sink.written_data
    
    def test_output_columns_are_exactly_right(self, pipeline):
        """Validate that downstream consumers get exactly the schema they expect."""
        pipeline.run()
        result = pipeline.sink.written_data["QUALIFIED_LEADS"]
        expected_columns = {"lead_id", "full_name", "email", "company", "lead_score", "created_at"}
        assert set(result.columns) == expected_columns
    
    def test_run_summary_reports_correct_counts(self, pipeline, sample_leads):
        summary = pipeline.run()
        
        assert summary["records_extracted"] == len(sample_leads)
        assert summary["records_loaded"] > 0
        assert summary["status"] == "success"
    
    def test_sink_health_check_failure_aborts_pipeline(self):
        """If the sink reports unhealthy, the pipeline should abort before extracting."""
        
        class AlwaysFailSink(SinkConnector):
            def load(self, df, destination):
                return 0
            def health_check(self):
                return False
        
        source = InMemorySourceConnector(pd.DataFrame({"a": [1, 2, 3]}))
        sink = AlwaysFailSink()
        p = LeadScoringPipeline(source=source, sink=sink)
        
        with pytest.raises(RuntimeError, match="Sink health check failed"):
            p.run()

Run these with pytest tests/test_lead_scoring_pipeline.py -v. They'll complete in milliseconds because they never touch the network.

Notice what we're testing. We're not testing that Snowflake receives the data — Snowflake's own test suite covers that. We're testing our logic: the scoring formula, the filtering threshold, the output schema, the health check behavior. This is the correct division of responsibility.

---

Hands-On Exercise

Now it's your turn to extend this architecture. Build the following in sequence:

Step 1: Add a Salesforce Source Connector

Create a SalesforceSourceConnector that satisfies the SourceConnector interface using the simple_salesforce library. The extract method should accept a SOQL query string. The health_check method should attempt a simple SELECT Id FROM Organization LIMIT 1.

Config class should include: username, password, security_token, domain (defaults to "login").

Step 2: Add a BigQuery Sink Connector

Create a BigQuerySinkConnector using google-cloud-bigquery. The load method should accept the destination as a fully-qualified table reference like project.dataset.table. Use pandas_gbq or the BigQuery client's load_table_from_dataframe method.

Step 3: Write a Parameterized Test

Write a pytest.mark.parametrize test that runs the transformation logic with four different score_threshold values (5.0, 10.0, 20.0, 40.0) and asserts that the number of qualified leads decreases (or stays the same) as the threshold increases. This tests the monotonic filtering behavior of your pipeline — a property that should hold regardless of the specific leads in the dataset.

Step 4: Create a Config-Driven Factory

Extend the build_pipeline factory to support a YAML configuration file as an alternative to environment variables. The YAML structure should specify the source type, sink type, and their respective configurations. Use PyYAML for parsing. The factory should be the only place that knows how to read configuration — the connectors themselves should remain ignorant of where their config came from.

---

Common Mistakes & Troubleshooting

Mistake 1: Injecting Config Instead of Connectors

A common half-measure is to inject configuration dictionaries into the pipeline rather than connector objects. You end up with:

# Anti-pattern: injecting config, not the dependency
def __init__(self, source_config: dict, sink_config: dict):
    self.source = PostgresSourceConnector(source_config)  # pipeline still decides the type
    self.sink = SnowflakeSinkConnector(sink_config)

This doesn't help you test the pipeline in isolation — you still need Postgres and Snowflake. The injection needs to happen at the connector level, not the config level.

Mistake 2: Health Checks That Are Too Expensive

If your health check does a full table scan or runs a slow query, it becomes a tax on every pipeline run. Keep health checks minimal — SELECT 1 for databases, HEAD requests for APIs, head_bucket for S3. The point is to verify connectivity, not data integrity.

Mistake 3: Mutable Default Arguments in DataFrame Operations

In _transform, always call df.copy() before modifying the DataFrame. If you modify the original DataFrame in place, your InMemorySourceConnector fixture will be mutated across test runs, causing tests to interfere with each other in ways that are maddening to debug.

Mistake 4: Forgetting to Test the Health Check Logic Separately

Many engineers test the happy path but forget to test what happens when health_check returns False. Add explicit tests for both source and sink failure modes. The AlwaysFailSink pattern in the test suite above is the right approach.

Mistake 5: The Factory Function Becoming a God Object

As the number of environments and connector combinations grows, your factory function can balloon into hundreds of lines of conditionals. When this happens, consider splitting it into separate factory functions per connector type:

def build_source(config: dict) -> SourceConnector:
    source_type = config["type"]
    if source_type == "postgres":
        return PostgresSourceConnector(PostgresConfig(**config["params"]))
    elif source_type == "salesforce":
        return SalesforceSourceConnector(SalesforceConfig(**config["params"]))
    raise ValueError(f"Unknown source type: {source_type}")

def build_sink(config: dict) -> SinkConnector:
    sink_type = config["type"]
    if sink_type == "snowflake":
        return SnowflakeSinkConnector(SnowflakeConfig(**config["params"]))
    elif sink_type == "s3_parquet":
        return S3ParquetSinkConnector(S3Config(**config["params"]))
    raise ValueError(f"Unknown sink type: {sink_type}")

This keeps each factory small and single-purpose.

Troubleshooting: "My Tests Pass But the Pipeline Fails in Staging"

This usually means your fixture data doesn't reflect the actual schema. Check:

1. Data types: Your CSV fixture might have page_views as a float due to NaN handling, while Postgres returns integers. Add explicit dtype assertions to your test fixtures.
2. Missing columns: Add a test that checks for all expected columns in the source DataFrame before transformation runs.
3. Null handling: Production data has nulls; test fixtures often don't. Explicitly add rows with None values to your fixtures.

---

Summary & Next Steps

You've built a dependency-injected pipeline architecture from first principles. The key structural decisions were:

1. Abstract interfaces (SourceConnector, SinkConnector) that define contracts without implementations
2. Concrete connectors that implement those interfaces for real infrastructure (Postgres, Snowflake, S3)
3. In-memory connectors that satisfy the same interfaces for fast, isolated testing
4. A pipeline class that depends on the abstractions and contains all business logic
5. A factory function that assembles the right combination of connectors for each environment

The payoff is real: your transformation logic is fully testable without infrastructure, your pipeline is promotable across environments via configuration, and swapping a connector (say, migrating from Snowflake to BigQuery) requires writing one new class and updating the factory — not touching your business logic at all.

Where to go next:

• Add observability: Extend the SourceConnector and SinkConnector interfaces with get_metrics() methods that return row counts, latency, and error rates. Inject a metrics client (like StatsD or Prometheus) into your connectors the same way you injected the connectors into the pipeline.
• Async connectors: For high-throughput pipelines, explore implementing an AsyncSourceConnector ABC using Python's asyncio and aiohttp. The dependency injection pattern holds — you just add async/await to the interface methods.
• Connector registries: When you have a dozen connector types, consider a registry pattern using a dictionary of connector classes keyed by string identifiers. Combined with a YAML config file, this gives you a fully declarative pipeline configuration system.
• Schema validation at the boundary: Add a validate_schema(df: pd.DataFrame) -> None method to both ABCs. Connectors should validate that what they produce (or what they receive) matches the expected schema, failing fast with a clear error rather than silently writing malformed data.

The architecture you've built here is the foundation that tools like Apache Beam, dbt's Python models, and Prefect's task system are built on. Understanding it from first principles makes you a much more effective user of those tools — and gives you the judgment to know when to reach for them and when to build your own.