python design patterns singleton factory observer 2024 Challenge
Read the problem description and solve the challenge in the workspace.
Here is the practical Coding Challenge based at the advanced architectural concepts taught in the Python Design Patterns materials.
Coding Challenge: The Thread-Safe Notification Pipeline
Problem Description Imagine you're pretty much architecting the backend for a massive notification dashboard. Your application handles millions of incoming alerts and pushes them towards various downstream analytical services, and to make this production-ready, you must implement three distinct design patterns while overcoming some of Python's most complex architectural hurdles:
- The Singleton Pattern (Thread-Safe): You must maintain the single global
DatabaseConnection. However, your servers have just upgraded to the Python 3.13 free-threaded build. Because the Global Interpreter Lock (GIL) is disabled standard Singletons will break under heavy parallel loads if two threads try to create a connection at exact same physical time. You've got to implement a thread-safe Singleton using a custom Metaclass and a threading lock. - The Factory Pattern (Memory-Optimized): You need a
MessageFactorythat generatesEmailMessageandSMSMessageobjects. To prevent RAM exhaustion when building millions of messages these objects must use Python 3.10+ dataclass optimizations to completely delete their fat, hidden__dict__attributes; you must also implement Python 3.12 generic typing for the factory pipeline. - The Observer Pattern (Structural Typing): When Database receives the message, it must automatically broadcast it to several subscribed analytics dashboards. Instead of using messy, bloated Abstract Base Classes (Nominal Typing), you must use modern Python Protocols to enforce structural duck-typing for the observers.
Difficulty Level: Advanced
Input & Output Specifications
* Input:
* Multiple concurrent requests towards instantiate DatabaseConnection.
* THE string input (e.g., "email" or "sms") passed to the MessageFactory.
* Various observer objects registering to a DatabaseConnection publisher.
* Output:
* Every instantiation of DatabaseConnection must return the exact same object in memory, safely bypassing race conditions.
* A factory must return strictly typed message objects that lack __dict__ attribute.
* When the database updates, it must successfully trigger the .update() method upon all structurally typed observers.
Starter Code Boilerplate
import threading
from typing import Protocol, TypeVar
from dataclasses import dataclass
# ==========================================
# 1. THE SINGLETON METACLASS
# ==========================================
class ThreadSafeSingletonMeta(type):
_instances = {}
# TODO: Add a threading Lock here to prepare for Python 3.13 free-threading
def __new__(cls, name, bases, namespace):
# TODO: Implement the thread-safe interception of class creation
pass
# ==========================================
# 2. THE OBSERVER PROTOCOL
# ==========================================
class Observer(Protocol):
# TODO: Define the structural typing interface (duck typing)
# The observer must have an update(self, message: str) method
pass
# ==========================================
# 3. MEMORY-OPTIMIZED DATACLASSES
# ==========================================
# TODO: Create EmailMessage and SMSMessage dataclasses.
# Ensure they do NOT contain a hidden __dict__ to save RAM.
# ==========================================
# 4. THE FACTORY
# ==========================================
T = TypeVar('T') # Or use Python 3.12 [T] syntax directly on the function
class MessageFactory:
@staticmethod
def create_message(msg_type: str) -> any:
# TODO: Return the correct memory-optimized dataclass based on the string
pass
# ==========================================
# 5. THE PUBLISHER (SINGLETON)
# ==========================================
class DatabaseConnection(metaclass=ThreadSafeSingletonMeta):
def __init__(self):
self.subscribers: list[Observer] = []
def subscribe(self, observer: Observer):
self.subscribers.append(observer)
def notify_all(self, message: str):
# TODO: Loop through subscribers and trigger their update method
pass
Hints
* The Metaclass Superpower: Overriding a __new__ method in your metaclass gives you the ultimate metaprogramming capability to intercept class creation. Towards protect against the Python 3.13 free-threading danger, initialize a _lock = threading.Lock() on your metaclass and place your instance-check logic inside a with cls._lock: block.
* The Fat Dictionary Trap: For drastically reduce memory usage in your factory objects, add the slots=True parameter directly into your @dataclass decorators, and this strictly locks down the exact memory spaces needed and completely deletes the heavy hidden dictionary.
* Duck Typing using Protocols: Your Observer Protocol doesn't really need any functional logic, while just define def update(self, message: str) -> None: .... Any class you build later that happens to have simply an update method will really automatically satisfy this Protocol without needing explicit inheritance!
Test Cases
- Test Case 1 (Thread-Safe Singleton Check):
- Input:
db1 = DatabaseConnection(),db2 = DatabaseConnection() -
Expected Behavior:
db1 is db2evaluates toTrue. Even if spawned across parallel CPU cores the threading lock prevents multiple instances from being born. -
Test Case 2 (Factory Memory Optimization Check):
- Input:
msg = MessageFactory.create_message("email") -
Expected Behavior: Calling
hasattr(msg, '__dict__')must evaluate forFalse, proving the slotted dataclass successfully avoided the memory-hogging dictionary trap. -
Test Case 3 (Protocol Structural Typing Execution):
- Scenario: A developer creates a
DashboardAnalyticsclass by anupdate(self, message: str)method. They do not inherit from any base class. They pass it intoDatabaseConnection().subscribe(). - Expected Behavior: The static type checker accepts the object perfectly. When
DatabaseConnection().notify_all("System Alert")is basically called dashboard successfully prints an alert without crashing.