Asyncio Patterns: Producer-Consumer, Fan-Out, Pipelines

tags: python, async, tutorial, advanced

tags: python, oop, tutorial, advanced

You’ve probably stared at a block of code that feels like a tangled ball of yarn, wondering why you keep rewriting the same object-creation logic or manually notifying every part of your app when something changes. Those frustrations aren’t bugs in your skills—they’re signals that you’re missing a design pattern. Python’s flexibility makes it easy to build quickly, but without patterns like Factory, Singleton, and Observer, your code can become fragile, hard to test, and impossible to scale. Today, you’ll walk away with three battle-tested patterns you can drop into your project today to make object creation cleaner, enforce single-instance guarantees, and build event-driven systems that actually work.

Why Design Patterns Matter in Python

Design patterns aren’t abstract theory—they’re reusable solutions to recurring problems that have been proven across decades of software development [2]. In Python, they help you communicate intent, reduce duplication, and make your code more maintainable. Think of them as shared vocabulary: when you say “I’m using a Factory here,” your teammates instantly understand you’re decoupling object creation from usage.

The three patterns we’ll cover fall into two categories:

Creational patterns (Factory, Singleton): Handle how objects are created.
Behavioral patterns (Observer): Manage how objects communicate and react to changes.

You’ll use these when your code feels “tangled,” you’re repeating logic, or you know you’ll need to add features later [4].

Factory Pattern: Decouple Object Creation

The Factory pattern simplifies complex object creation by letting you decide which class to instantiate at runtime, without hard-coding class names. This is crucial when you have multiple similar classes (e.g., Student, Teacher, Admin) and want to avoid if/else spaghetti.

Real-World Use Case

Imagine a user management system where you need to create different user types based on input. Without a factory, you’d write:

if user_type == "student":
    user = Student()
elif user_type == "teacher":
    user = Teacher()

This violates the Open/Closed Principle—adding a new type requires modifying existing code. A factory solves this.

Working Python Example

from abc import ABC, abstractmethod

class User(ABC):
    @abstractmethod
    def greet(self):
        pass

class Student(User):
    def greet(self):
        return "Hi, I'm a student!"

class Teacher(User):
    def greet(self):
        return "Hello, I'm a teacher!"

class UserFactory:
    def create_user(self, user_type: str) -> User:
        if user_type == "student":
            return Student()
        elif user_type == "teacher":
            return Teacher()
        raise ValueError(f"Unknown user type: {user_type}")

# Usage
factory = UserFactory()
user = factory.create_user("student")
print(user.greet())  # Output: Hi, I'm a student!

Notice how the creation logic is isolated in UserFactory. Adding Admin later only requires updating the factory—not every place that creates users [6]. This is your actionable takeaway: wrap object creation in a factory class to keep your main code clean and extensible.

Singleton Pattern: Ensure One Instance

The Singleton pattern guarantees that a class has only one instance and provides a global point of access to it. This is perfect for shared resources like database connections, configuration managers, or logging systems.

When to Use Singleton

You need a single configuration object across your app.
You’re managing a database connection pool where one instance is sufficient.
You want to avoid redundant initialization of expensive resources.

Python Implementation

Python doesn’t have a built-in Singleton, but you can implement one using a class with a private instance variable:

class ConfigManager:
    _instance = None
    _initialized = False

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self):
        if not self._initialized:
            self.settings = {"debug": False, "timeout": 30}
            self._initialized = True

    def get_setting(self, key):
        return self.settings.get(key)

# Usage
config1 = ConfigManager()
config2 = ConfigManager()

print(config1 is config2)  # True: Same instance!
config1.settings["debug"] = True
print(config2.get_setting("debug"))  # True: Shared state

Here, __new__ ensures only one instance is created, and __init__ runs only once [4]. Use this pattern when you must have a single point of control—but avoid overusing it, as it can hide dependencies and make testing harder.

Observer Pattern: Build Event-Driven Systems

The Observer pattern lets objects (observers) subscribe to events from another object (the subject) and react automatically when changes occur. This is the backbone of event-driven systems, like UI updates, logging, or real-time notifications.

Real-World Scenario

Imagine a dashboard that updates when a sensor value changes. Instead of manually checking the sensor everywhere, you attach observers that get notified automatically.

Python Implementation

from abc import ABC, abstractmethod

class Subject(ABC):
    def __init__(self):
        self._observers = []

    def attach(self, observer):
        self._observers.append(observer)

    def detach(self, observer):
        self._observers.remove(observer)

    def notify(self):
        for observer in self._observers:
            observer.update(self)

class Sensor(Subject):
    def __init__(self):
        super().__init__()
        self._value = 0

    def set_value(self, value):
        self._value = value
        self.notify()

    def get_value(self):
        return self._value

class Dashboard:
    def __init__(self, name):
        self.name = name

    def update(self, subject):
        print(f"{self.name} updated: {subject.get_value()}")

# Usage
sensor = Sensor()
dashboard1 = Dashboard("Dashboard 1")
dashboard2 = Dashboard("Dashboard 2")

sensor.attach(dashboard1)
sensor.attach(dashboard2)

sensor.set_value(42)  # Both dashboards update automatically

Here, Dashboard objects subscribe to Sensor changes. When set_value() is called, all observers are notified [1]. This pattern is ideal for building modular, scalable systems where components react to events without knowing about each other.

Putting It All Together: A Practical Mini-Project

Let’s combine these patterns in a tiny but realistic app: a logging system that:

Uses Singleton for a single Logger instance.
Uses Factory to create different log handlers (file, console, email).
Uses Observer to notify multiple dashboards when a log event occurs.

import logging
from abc import ABC, abstractmethod

# Singleton Logger
class Logger:
    _instance = None
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance.handlers = []
        return cls._instance

    def add_handler(self, handler):
        self.handlers.append(handler)

    def log(self, message):
        for handler in self.handlers:
            handler.handle(message)

# Factory for Handlers
class LogHandler(ABC):
    @abstractmethod
    def handle(self, message):
        pass

class ConsoleHandler(LogHandler):
    def handle(self, message):
        print(f"[Console] {message}")

class FileHandler(LogHandler):
    def handle(self, message):
        with open("logs.txt", "a") as f:
            f.write(f"[File] {message}\n")

class HandlerFactory:
    def create_handler(self, type: str) -> LogHandler:
        if type == "console":
            return ConsoleHandler()
        elif type == "file":
            return FileHandler()
        raise ValueError(f"Unknown handler: {type}")

# Observer for Dashboards
class LogDashboard:
    def __init__(self, name):
        self.name = name

    def update(self, message):
        print(f"{self.name} received: {message}")

logger = Logger()
factory = HandlerFactory()
logger.add_handler(factory.create_handler("console"))
logger.add_handler(factory.create_handler("file"))

dashboard = LogDashboard("Main Dashboard")
logger.handlers.append(dashboard)  # Attach as observer

logger.log("User logged in")