The Data Source Abstraction Pattern: One Interface, Multiple Backends

Introduction

Your pipeline reads from MongoDB today. Tomorrow it’s Redshift. Next quarter, someone adds Salesforce. Without abstraction, every new source means rewriting orchestration logic. Here’s a pattern that keeps your pipeline code clean regardless of where data lives.

💥 Master Any Skills in 3 Months 📚 Up to 50% OFF Premium Courses ⏳ Limited-Time Offer *👉 **Enroll Now & Start Learning***

The Problem

In data engineering, sources multiply fast. A typical metadata pipeline might pull from:

→ MongoDB (operational data)

→ Redshift (analytics warehouse)

→ Salesforce (business context)

→ REST APIs (vendor systems)

Without abstraction, your main pipeline becomes a tangle of if/else blocks and source-specific logic

The Pattern: Abstract Base + Concrete Implementations

from abc import ABC, abstractmethod
from typing import Any

class BaseDataSource(ABC):
 """Common interface for all data sources."""

 @abstractmethod
 def connect(self) -> None:
   """Establish connection to the data source."""
   …

 @abstractmethod
 def fetch(self, query: dict) -> list[dict[str, Any]]:
   """Fetch data based on query parameters."""
   …

 @abstractmethod
 def close(self) -> None:
   """Clean up resources."""
   …

 def __enter__(self):
   self.connect()
   return self

 def __exit__(self, *args):
   self.close()

Concrete Implementation Example

class MongoDataSource(BaseDataSource):

  def __init__(self, uri: str, database: str, collection: str):
   self.uri = uri
   self.database = database
   self.collection = collection

  def connect(self):
   self.client = MongoClient(self.uri)
   self.db = self.client[self.database]

  def fetch(self, query: dict) -> list[dict]:
   return list(self.db[self.collection].find(query))

  def close(self):
   self.client.close()


 class RedshiftDataSource(BaseDataSource):

    def __init__(self, cluster: str, database: str):
     self.cluster = cluster
     self.database = database

    def connect(self):
     self.conn = redshift_connector.connect(…)

    def fetch(self, query: dict) -> list[dict]:
     sql = self._build_query(query)
     cursor = self.conn.cursor()
     cursor.execute(sql)
     return cursor.fetchall()

    def close(self):
     self.conn.close()

The Pipeline Stays Clean

def collect_metadata(sources: list[BaseDataSource], query: dict):
  """Pipeline logic is source-agnostic."""
  results = []
  for source in sources:
    with source:
      data = source.fetch(query)
      results.extend(data)
  return results

Benefits

1. Adding a new source = one new class, zero changes to pipeline logic
2. Testing is trivial — mock the interface, not the implementation
3. Config-driven source selection via factory pattern
4. Each source handles its own connection lifecycle

When This Pattern Shines

This pattern pays off when you have 3+ data sources, when sources change over time, or when different environments use different backends (e.g., SQLite in tests, Redshift in prod).

🎁 Bonus: Handling “Shape-Shifting” Business APIs (The Salesforce Case)

In the architecture diagram above, you might have noticed Salesforce listed as a backend. Unlike production databases (MongoDB) or data warehouses (Redshift), third-party SaaS APIs like Salesforce are heavily managed by business operations teams, not engineers.

An admin can add custom fields, change picklist options, or alter objects directly in a UI on any given Tuesday. This means its API “shape” changes constantly and unexpectedly.

Here is how our BaseDataSource pattern earns its keep by isolating that chaos inside a concrete implementation:

from simple_salesforce import Salesforce

class SalesforceDataSource(BaseDataSource):
    def __init__(self, username, password, security_token):
        self.username = username
        self.password = password
        self.token = security_token
        self.sf = None

    def connect(self):
        # Establish connection using the simple-salesforce client
        self.sf = Salesforce(
            username=self.username, 
            password=self.password, 
            security_token=self.token
        )

    def fetch(self, query: dict) -> list[dict]:
        """
        Dynamically handles the shifting shape of business data
        without breaking the core pipeline code.
        """
        obj_name = query.get("object")       # e.g., "Contact"
        fields = query.get("fields", ["Id"]) # e.g., ["Id", "Email", "Custom_Field__c"]

        # Build the Salesforce Object Query Language (SOQL) string on the fly
        soql = f"SELECT {', '.join(fields)} FROM {obj_name}"

        # Execute and return raw records
        response = self.sf.query(soql)
        return response.get("records", [])

    def close(self):
        # Salesforce REST APIs are stateless, so no active connection to close!
        pass

Conclusion

Abstraction isn’t about over-engineering. It’s about making the next source addition a 30-minute task instead of a 3-day refactor. Define the interface once, implement it per source, and let your pipeline stay blissfully ignorant of where data comes from.

Thank you for being a part of the community

Before you go:

👉 Be sure to clap and follow the writer ️👏️️

👉 Follow us: **Linkedin| [Medium](https://medium.com/codetodeploy)**

👉 CodeToDeploy Tech Community is live on Discord — **Join now!**

Disclosure: This post includes affiliate and partnership links.