Reality Bytes: Building for the Real - BRIDGE design pattern

Does Reality Have to Byte?

Building software isn’t all sunshine and perfectly aligned divs. Those cozy toy problems with "animals" and "vehicles" that made patterns seem so clean? Turns out, they were just training wheels. Sometimes, you need examples rooted, at least slightly more, in the messy reality of actual software. And if you squint hard enough, focus longer than a TikTok loop, and resist the urge to ctrl+c your way through life, you might just discover that real-world design patterns can save you. Or at least, make reality byte a little less.

Introduction to the Bridge Pattern

In software development, design patterns are used to create more flexible and scalable code. One such pattern is the Bridge Pattern, which helps separate an abstraction from its implementation. This separation allows both parts to evolve independently, making it easier to adapt and maintain the system.

What Is the Bridge Pattern?

The Bridge Pattern divides a system into two distinct hierarchies: one for the abstraction (what the system does) and one for the implementation (how it does it). These two parts are connected through a "bridge" that allows the abstraction to communicate with the implementation. This decoupling means that changes in the abstraction or the implementation can be made without affecting the other, offering greater flexibility.

Why Use the Bridge Pattern?

When the abstraction and implementation are tightly coupled, changes to one often require changes to the other. This can make the code hard to maintain, especially in large projects. The Bridge Pattern solves this by separating the two, allowing each to change independently. As a result, the system becomes more flexible, easier to maintain, and better able to adapt to future needs without breaking existing code.

Common Uses

The Bridge Pattern is commonly used in scenarios like UI libraries, where UI elements are abstracted from platform-specific details, or in cross-platform systems that need to work across different platforms. It’s also helpful when integrating with multiple third-party APIs, as it separates the API interaction logic from the main application, making the code more flexible and easier to maintain.

Components

The Bridge Pattern has four main components that work together to provide flexibility and separation.

The Abstraction defines the high-level interface and relies on an implementation to handle the details. For example, a DataService class might define operations like fetching or updating data but delegate the actual work to another layer.

The Refined Abstraction extends the base abstraction by adding more specific functionality or custom behavior.

The Implementor is an interface that specifies the methods concrete implementations must follow, focusing on how tasks are performed. For instance, an APIImplementation interface might define methods for connecting to external APIs.

Finally, the Concrete Implementations are the classes that carry out the actual work, following the implementor's guidelines.

Bringing the Bridge Pattern to Life: Real-World API Integration

Now that we explored the Bridge Pattern concept, let's make this more tangible. Let’s dive into a real-world example of how the Bridge Pattern can be applied in a REST API setting. Specifically, we’ll look at integrating two external services: CoinCap, which provides cryptocurrency data, and AnimeChan, an API that fetches anime-related quotes. These were chosen for simplicity since the have no auth requirements and are free!

The power of the Bridge Pattern in this scenario lies in its ability to separate the high-level abstraction of fetching and processing data from the low-level specifics of how the data is obtained from these external APIs. This allows us to interact with different data sources without tightly coupling the logic for fetching data from the APIs with the service layer that handles the data.

We’ll break down how we implement this pattern step-by-step, starting with defining the core structure and extending it through concrete implementations, each tailored to interact with a specific API.

By the end of this exercise, you'll see how the Bridge Pattern helps create a system that's easy to extend with new APIs and services, offering flexibility and reducing the risk of tightly coupled code. Let's dive into how this all works together!

Defining the Base Data Service and Its Subclasses

To start, we first define a Base Data Service class that serves as the abstraction. This class will define the core functionality and outline the methods we expect to interact with the external APIs. By doing so, we create a flexible interface that any specific service class can implement, regardless of which API it's interacting with.

The BaseDataService Class

This class is designed to serve as a base for all specific services that interact with external APIs. It establishes the core contract that any subclass must adhere to, specifically the ability to fetch data from an API and process that data into a usable format.

Here’s how we define this base class:

from abc import abstractmethod, ABC

from rest_api_bridge.abstractions.data_fetcher import DataFetcher


class BaseDataService(ABC):

    #Base class for services
    
    def __init__(self, api_impl: DataFetcher):
        self._api_impl = api_impl
    
    @abstractmethod
    async def fetch_data(self, endpoint: str, params: dict = None):
        pass

    @abstractmethod
    def process_data(self, data):
        pass

Subclassing the BaseDataService

Now, we can define specific subclasses of the BaseDataService, each tailored to interact with different external APIs. These subclasses will implement the methods defined in the base class, providing specific logic for fetching and processing data from each API.

AnimeService

The AnimeService subclass interacts with the AnimeChan API, which returns anime quotes. The fetch_data method in this class fetches the data from the API, while the process_data method processes the raw data into a usable format.

from rest_api_bridge.services.base_service import BaseDataService


class AnimeService(BaseDataService):
    """
    A specific DataService implementation for fetching anime quotes from AnimeChan API.
    """

    async def fetch_data(self, endpoint: str, params: dict = None):
        print(f"AnimeService: Fetching data from '{endpoint}'...")
        return await self._api_impl.fetch_data(endpoint, params)

    def process_data(self, data):
        """
        Process and return the quote data.
        """
        print("AnimeService: Processing data...")
        if not data or "data" not in data:
            print(f"AnimeService: Invalid data received: {data}")
            return None
        quote_data = data["data"]
        return {
            "quote": quote_data["content"],
            "character": quote_data["character"]["name"],
            "anime": quote_data["anime"]["name"],
        }

CoinCapService

The CoinCapService subclass works with the CoinCap API to fetch cryptocurrency data. It follows the same pattern as AnimeService, but the data is processed differently to handle the structure of the cryptocurrency data.

from rest_api_bridge.services.base_service import BaseDataService


class CoinCapService(BaseDataService):
    """
    A specific DataService implementation for fetching data from CoinCap API.
    """

    async def fetch_data(self, endpoint: str, params: dict = None):
        print(f"CoinCapService: Fetching data from '{endpoint}'...")
        return await self._api_impl.fetch_data(endpoint, params)

    def process_data(self, data):
        """
        Process the fetched data into a simplified format.
        """
        try:
            assets = data.get("data", [])
            return [
                {
                    "id": asset["id"],
                    "rank": asset["rank"],
                    "symbol": asset["symbol"],
                    "priceUsd": asset["priceUsd"]
                }
                for asset in assets
            ]
        except Exception as e:
            print(f"Error processing data: {e}")
            return []

How These Subclasses Work Together

By defining specific service classes like AnimeService and CoinCapService, we achieve loose coupling between the abstraction and the implementation. The BaseDataService class defines the high-level interface for fetching and processing data, but it doesn’t dictate how the data is fetched or how it should be processed. Each subclass implements the details for a particular API, making it easy to introduce new services or change the implementation without affecting other parts of the application.

For instance, if we wanted to add a new API service, like a WeatherService that fetches weather data, we can simply create another subclass of BaseDataService and implement the fetch_data and process_data methods. This modularity ensures that the core logic of our application remains unchanged, and we can scale our service architecture easily.

Introducing the APIImplementation Class and Its Subclasses

In the Bridge Pattern, one of the key concepts is separating the abstraction (how we interact with the external service) from the implementation (the specifics of how we fetch the data). In this section, we’ll explore the APIImplementation class, which defines the interface for interacting with various external APIs, and then look at the subclasses that implement the specifics for different APIs.

The APIImplementation Abstract Base Class

The Bridge Pattern separates how we interact with external services (abstraction) from how the data is fetched (implementation). The APIImplementation class serves as the abstract base class, defining the contract for interacting with external APIs. It includes an abstract method, fetch_data, which all subclasses must implement. This setup provides a flexible structure for data fetching while allowing each subclass to handle the specific details for different APIs. Here's how the DataFetcher class is defined:

from abc import ABC, abstractmethod


class DataFetcher(ABC):
    """
    Abstract base class for fetching data from various sources.

    Subclasses must implement the `fetch_data` method, which specifies 
    how to fetch data from a given source. This can include APIs, databases, files, etc.
    """

    @abstractmethod
    async def fetch_data(self, source: str, params: dict = None):
        """
        Fetch data from the specified source.

        Args:
            source (str): The data source (e.g., URL, file path, database query).
            params (dict, optional): Additional parameters for data fetching.

        Returns:
            Data from the specified source, structured as needed by the implementation.
        """
        pass

This abstract class establishes the general structure for how any service will interact with an API. The fetch_data method is asynchronous, ensuring that we can make non-blocking calls to external APIs, which is crucial for performance in real-time applications.

Subclasses of APIImplementation

Now, let’s look at the concrete implementations of this interface, specifically for the AnimeChan API (for fetching anime quotes) and the CoinCap API (for cryptocurrency data). These subclasses implement the fetch_data method, tailored to the specific requirements of each API.

The AnimeApi Subclass

The AnimeApi class is an implementation of the DataFetcher abstract class that interacts with the AnimeChan API. It fetches anime-related quotes by making an HTTP GET request to the API’s endpoint.

Here’s the implementation:

import httpx
from rest_api_bridge.abstractions.data_fetcher import DataFetcher

class AnimeApi(DataFetcher):
    """
    Implementation of DataFetcher for the AnimeChan API.
    """

    async def fetch_data(self, source: str, params: dict = None):
        """
        Fetch data from the AnimeChan API.

        Args:
            source (str): The endpoint URL.
            params (dict, optional): Additional parameters for the API call.

        Returns:
            dict: JSON response from the AnimeChan API.
        """
        print(f"AnimeApi: Fetching data from {source}...")
        async with httpx.AsyncClient() as client:
            response = await client.get(source, params=params)
            response.raise_for_status()
            return response.json()

In this implementation:

The fetch_data method sends a GET request to the AnimeChan API using the httpx library. If the response is successful, it returns the JSON data; otherwise, it raises an exception for invalid HTTP status codes.

The CoinCapAPI Subclass

The CoinCapAPI class is another implementation of the DataFetcher class, this time for interacting with the CoinCap API. This API provides real-time cryptocurrency data, and our implementation fetches that data by sending an HTTP GET request to the CoinCap API’s endpoint.

Here’s the implementation:

import httpx
from rest_api_bridge.abstractions.data_fetcher import DataFetcher

class CoinCapAPI(DataFetcher):
    """
    Implementation of DataFetcher for the CoinCap API.
    """

    BASE_URL = "https://api.coincap.io/v2"

    async def fetch_data(self, source: str, params: dict = None):
        """
        Fetch data from the CoinCap API.

        Args:
            source (str): The endpoint path (relative to BASE_URL).
            params (dict, optional): Additional parameters for the API call.

        Returns:
            dict: JSON response from the CoinCap API.
        """
        url = f"{self.BASE_URL}{source}"
        print(f"CoinCapAPI: Fetching data from {url}...")
        async with httpx.AsyncClient() as client:
            response = await client.get(url, params=params)
            response.raise_for_status()
            return response.json())

How These Implementations Work Together

The key takeaway from these implementations is that they encapsulate the logic for interacting with different APIs behind a common interface, the APIImplementation class. Each API implementation knows how to communicate with its respective external service, but the rest of the application doesn’t need to worry about the details of these interactions.

For example, the CoinCapAPI class knows how to interact with the CoinCap API, while the AnimeApi knows how to fetch quotes from the AnimeChan API. The rest of the application, including the BaseDataService subclasses like CoinCapService and AnimeService, only need to rely on the abstract fetch_data method to fetch data, without worrying about the specifics of each API.

This allows us to easily add new API integrations in the future. If we wanted to integrate another API, say a Weather API, we would simply create a new subclass of DataFetcher and implement the fetch_data method for that specific API.

Using the Bridge Pattern in FastAPI Routes

Now let’s look at how these components are integrated into FastAPI routes.

Setting Up the Service Instances

We need to instantiate these classes and use them in the AnimeService and CoinCapService classes. These service classes are responsible for using the appropriate fetch_data method to retrieve raw data, then processing it before returning it to the API route.

from apis.api_implementation import AnimeApi, CoinCapAPI
from services.anime_service import AnimeService
from services.coincap_service import CoinCapService
from fastapi import APIRouter

# Instantiate the API implementations
anime_api_impl = AnimeApi()
coincap_api_impl = CoinCapAPI()

# Create instances of the service classes
anime_service = AnimeService(anime_api_impl)
coincap_service = CoinCapService(coincap_api_impl)

# Initialize the router
router = APIRouter()

We can now define two FastAPI routes that will use these services to fetch data from external APIs:

Route for Fetching a Random Anime Quote

@router.get("/anime/quote/")
async def get_anime_quote():
    
    #Fetch a random anime quote from the AnimeChan API.
    
    endpoint = "https://animechan.io/api/v1/quotes/random"
    # Use the AnimeService to fetch and process data
    raw_data = await anime_service.fetch_data(endpoint)
    processed_data = anime_service.process_data(raw_data)
    
    # Return the processed data if available
    if processed_data:
        return {
            "message": "Random Anime Quote",
            "data": processed_data,
        }
    
    
    return {"message": "Failed to fetch Anime Quote"}

The get_anime_quote route defines the endpoint /anime/quote/.

@router.get("/coincap/data/")
async def get_coincap_data():
    
    #Fetch data from the CoinCap API.
   
    endpoint = "/assets"  
    
    raw_data = await coincap_service.fetch_data(endpoint)
    processed_data = coincap_service.process_data(raw_data)
    

    if processed_data:
        return {
            "message": "CoinCap Data",
            "data": processed_data,
        }
    

    return {"message": "Failed to fetch CoinCap data"}

The get_coincap_data route defines the endpoint /coincap/data/. It calls the CoinCapService to fetch data from the CoinCap API and processes the raw data accordingly. The processed data is returned if successful, otherwise, an error message is sent back.

Integrating the Routes into the FastAPI Application

Finally, we need to include the routes in the main FastAPI application. We simply import the router and add it to the application instance:

from fastapi import FastAPI
from routes import router  

app = FastAPI()

# Include the router in the FastAPI app
app.include_router(router)

This ensures that both the /anime/quote/ and /coincap/data/ endpoints are accessible via the FastAPI application.

Conclusion By using the Bridge Pattern in this example, we've created a flexible architecture where the logic of fetching data from external APIs is separated from the way the data is processed and exposed through routes. This structure makes it easy to add more APIs in the future—simply create new implementations of the APIImplementation class and corresponding services, without affecting the existing code.

To view the complete code, checkout the repository here: https://github.com/gizmoGremlin/fast_api_python_bridge