Bridge

Bridges are used to decouple an OOP Abstraction from its implementation, allowing them to be developed independently of each other.

TL;DR

The Bridge pattern is about using composition over inheritance when designing objects, allowing certain details to be placed in a separate object rather than in a large hierarchy.

It is an implementation of the Liskov Substitution Principle (LSP) using Dependency Injection, and optionally of the Interface Segregation Principle (ISP) if the abstraction is extended.

Use when a large classs needs to be broken up into smaller variants, when a class might grow in several orthogonal dimensions, or if we need to switch implementations at runtime.

Problem

Suppose you have a set of Devices, such as TVs, Radios, etc. Each of them may have a Remote, although each Remote itself might be different. As a result, if Remote operations are defined as part of the Device interface, the complexity of the code will grow exponentially given every combination of Remote and Device possible.

Solution

The solution is to use the Bridge pattern and breakup monolithic classes into separate objects, where one of them acts as an OOP interfaces for the other.

In the case of the Remote and Device example above, we can define the Remote as the “abstraction”, which itself holds a Device as part of its definition. Note that this composition design allows use to extend the Remote into AdvancedRemote without modifying the class itself, and ensures that only actually used methods are inherited, satisfying all the SOLID principles.

This type of pattern is often combined with Factory to create modular code.

# Bridge Pattern Example: Remote Control and TV Device
from abc import ABC, abstractmethod
 
# Implementation interface (the "bridge")
class Device(ABC):
    @abstractmethod
    def is_enabled(self) -> bool:
        pass
    
    @abstractmethod
    def enable(self) -> None:
        pass
    
    @abstractmethod
    def disable(self) -> None:
        pass
    
    @abstractmethod
    def get_volume(self) -> int:
        pass
    
    @abstractmethod
    def set_volume(self, percent: int) -> None:
        pass
    
    @abstractmethod
    def get_channel(self) -> int:
        pass
    
    @abstractmethod
    def set_channel(self, channel: int) -> None:
        pass
 
# Concrete implementations
class TV(Device):
    def __init__(self):
        self._on = False
        self._volume = 30
        self._channel = 1
    
    def is_enabled(self) -> bool:
        return self._on
    
    def enable(self) -> None:
        self._on = True
        print("TV is now ON")
    
    def disable(self) -> None:
        self._on = False
        print("TV is now OFF")
    
    def get_volume(self) -> int:
        return self._volume
    
    def set_volume(self, percent: int) -> None:
        self._volume = max(0, min(100, percent))
        print(f"TV volume set to {self._volume}%")
    
    def get_channel(self) -> int:
        return self._channel
    
    def set_channel(self, channel: int) -> None:
        self._channel = channel
        print(f"TV channel set to {self._channel}")
 
class Radio(Device):
    def __init__(self):
        self._on = False
        self._volume = 50
        self._channel = 101  # FM frequency
    
    def is_enabled(self) -> bool:
        return self._on
    
    def enable(self) -> None:
        self._on = True
        print("Radio is now ON")
    
    def disable(self) -> None:
        self._on = False
        print("Radio is now OFF")
    
    def get_volume(self) -> int:
        return self._volume
    
    def set_volume(self, percent: int) -> None:
        self._volume = max(0, min(100, percent))
        print(f"Radio volume set to {self._volume}%")
    
    def get_channel(self) -> int:
        return self._channel
    
    def set_channel(self, channel: int) -> None:
        self._channel = channel
        print(f"Radio tuned to {self._channel} FM")
 
# Abstraction (uses the bridge to implementation)
class RemoteControl:
    def __init__(self, device: Device):
        self._device = device
    
    def toggle_power(self) -> None:
        if self._device.is_enabled():
            self._device.disable()
        else:
            self._device.enable()
    
    def volume_down(self) -> None:
        current_volume = self._device.get_volume()
        self._device.set_volume(current_volume - 10)
    
    def volume_up(self) -> None:
        current_volume = self._device.get_volume()
        self._device.set_volume(current_volume + 10)
    
    def channel_down(self) -> None:
        current_channel = self._device.get_channel()
        self._device.set_channel(current_channel - 1)
    
    def channel_up(self) -> None:
        current_channel = self._device.get_channel()
        self._device.set_channel(current_channel + 1)
 
# Extended abstraction
class AdvancedRemoteControl(RemoteControl):
    def mute(self) -> None:
        self._device.set_volume(0)
        print("Device muted")
 
# Example usage
if __name__ == "__main__":
    # Create devices
    tv = TV()
    radio = Radio()
    
    # Create remote controls
    tv_remote = RemoteControl(tv)
    radio_remote = AdvancedRemoteControl(radio)
    
    print("=== Controlling TV ===")
    tv_remote.toggle_power()      # Turn on TV
    tv_remote.volume_up()         # Increase volume
    tv_remote.channel_up()        # Change channel
    
    print("\n=== Controlling Radio ===")
    radio_remote.toggle_power()   # Turn on radio
    radio_remote.volume_up()      # Increase volume
    radio_remote.mute()          # Mute radio
    
    print("\n=== Same remote, different device ===")
    # The same remote can work with different devices
    universal_remote = AdvancedRemoteControl(tv)
    universal_remote.toggle_power()  # Turn off TV
    universal_remote.mute()         # Mute TV

More detailed example

# Bridge vs Factory Pattern: Database Connections
from abc import ABC, abstractmethod
from typing import List, Dict, Any
 
# =============================================================================
# BRIDGE PATTERN APPROACH
# =============================================================================
 
# Implementation interface (the "bridge")
class DatabaseDriver(ABC):
    @abstractmethod
    def connect(self, connection_string: str) -> None:
        pass
    
    @abstractmethod
    def execute_query(self, query: str) -> List[Dict[str, Any]]:
        pass
    
    @abstractmethod
    def close(self) -> None:
        pass
 
# Concrete implementations
class PostgreSQLDriver(DatabaseDriver):
    def connect(self, connection_string: str) -> None:
        print(f"Connecting to PostgreSQL: {connection_string}")
    
    def execute_query(self, query: str) -> List[Dict[str, Any]]:
        print(f"PostgreSQL executing: {query}")
        return [{"id": 1, "name": "John"}]  # Mock result
    
    def close(self) -> None:
        print("PostgreSQL connection closed")
 
class SQLiteDriver(DatabaseDriver):
    def connect(self, connection_string: str) -> None:
        print(f"Connecting to SQLite: {connection_string}")
    
    def execute_query(self, query: str) -> List[Dict[str, Any]]:
        print(f"SQLite executing: {query}")
        return [{"id": 1, "name": "Jane"}]  # Mock result
    
    def close(self) -> None:
        print("SQLite connection closed")
 
# Abstraction (uses dependency injection + LSP)
class Database:
    def __init__(self, driver: DatabaseDriver, connection_string: str):
        # DEPENDENCY INJECTION: Database depends on DatabaseDriver abstraction
        self._driver = driver  # Any DatabaseDriver implementation works (LSP)
        self._connection_string = connection_string
        self._driver.connect(connection_string)
    
    def find_user(self, user_id: int) -> Dict[str, Any]:
        query = f"SELECT * FROM users WHERE id = {user_id}"
        results = self._driver.execute_query(query)
        return results[0] if results else {}
    
    def close(self) -> None:
        self._driver.close()
 
# Refined abstraction with additional features
class AdvancedDatabase(Database):
    def find_users_with_caching(self, user_id: int) -> Dict[str, Any]:
        print("Checking cache first...")
        return self.find_user(user_id)  # Simplified - would add caching logic
 
# =============================================================================
# FACTORY PATTERN APPROACH
# =============================================================================
 
class DatabaseConnection(ABC):
    @abstractmethod
    def connect(self) -> None:
        pass
    
    @abstractmethod
    def execute_query(self, query: str) -> List[Dict[str, Any]]:
        pass
    
    @abstractmethod
    def close(self) -> None:
        pass
 
class PostgreSQLConnection(DatabaseConnection):
    def __init__(self, connection_string: str):
        self.connection_string = connection_string
    
    def connect(self) -> None:
        print(f"Connecting to PostgreSQL: {self.connection_string}")
    
    def execute_query(self, query: str) -> List[Dict[str, Any]]:
        print(f"PostgreSQL executing: {query}")
        return [{"id": 1, "name": "John"}]
    
    def close(self) -> None:
        print("PostgreSQL connection closed")
 
class SQLiteConnection(DatabaseConnection):
    def __init__(self, connection_string: str):
        self.connection_string = connection_string
    
    def connect(self) -> None:
        print(f"Connecting to SQLite: {self.connection_string}")
    
    def execute_query(self, query: str) -> List[Dict[str, Any]]:
        print(f"SQLite executing: {query}")
        return [{"id": 1, "name": "Jane"}]
    
    def close(self) -> None:
        print("SQLite connection closed")
 
# Factory
class DatabaseFactory:
    @staticmethod
    def create_connection(db_type: str, connection_string: str) -> DatabaseConnection:
        if db_type.lower() == "postgresql":
            return PostgreSQLConnection(connection_string)
        elif db_type.lower() == "sqlite":
            return SQLiteConnection(connection_string)
        else:
            raise ValueError(f"Unsupported database type: {db_type}")
 
# =============================================================================
# USAGE EXAMPLES
# =============================================================================
 
def demonstrate_bridge_pattern():
    print("=== BRIDGE PATTERN ===")
    # Key difference: You can switch drivers at runtime!
    pg_driver = PostgreSQLDriver()
    sqlite_driver = SQLiteDriver()
    
    # Same Database instance can use different drivers
    db = Database(pg_driver, "postgresql://localhost:5432/mydb")
    user1 = db.find_user(1)
    print(f"Found user with PostgreSQL: {user1}")
    
    # THIS IS THE KEY: Switch the driver at runtime
    print("\n--- Switching driver at runtime ---")
    db._driver = sqlite_driver  # Switch implementation
    db._driver.connect("sqlite:///mydb.db")
    user2 = db.find_user(1)
    print(f"Found user with SQLite: {user2}")
    db.close()
    
    # AdvancedDatabase shows you can extend abstraction independently
    advanced_db = AdvancedDatabase(pg_driver, "postgresql://localhost:5432/mydb")
    user3 = advanced_db.find_users_with_caching(1)
    print(f"Found user with caching: {user3}")
    advanced_db.close()
 
def demonstrate_factory_pattern():
    print("\n=== FACTORY PATTERN ===")
    # Factory creates the right connection type
    pg_conn = DatabaseFactory.create_connection("postgresql", "postgresql://localhost:5432/mydb")
    pg_conn.connect()
    user1 = pg_conn.execute_query("SELECT * FROM users WHERE id = 1")
    print(f"Found user: {user1}")
    pg_conn.close()
    
    sqlite_conn = DatabaseFactory.create_connection("sqlite", "sqlite:///mydb.db")
    sqlite_conn.connect()
    user2 = sqlite_conn.execute_query("SELECT * FROM users WHERE id = 1")
    print(f"Found user: {user2}")
    sqlite_conn.close()
 
if __name__ == "__main__":
    demonstrate_bridge_pattern()
    demonstrate_factory_pattern()