Scribd
Upload
3 views6 pages

Python Patterns Guide

The document is a guide on various Python design patterns, including Strategy, Factory, Adapter, Repository, Observer, Producer-Consumer, Dependency Injection, Ports and Adapters, and Domain-Driven Design (DDD). Each pattern is explained with its objective, a code example, and a main.py call to demonstrate its usage. The guide also provides tips on how to study these patterns and how to articulate their use in interviews.

Uploaded by

Bruno Brito
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd
3 views6 pages

Python Patterns Guide

The document is a guide on various Python design patterns, including Strategy, Factory, Adapter, Repository, Observer, Producer-Consumer, Dependency Injection, Ports and Adapters, and Domain-Driven Design (DDD). Each pattern is explained with its objective, a code example, and a main.py call to demonstrate its usage. The guide also provides tips on how to study these patterns and how to articulate their use in interviews.

Uploaded by

Bruno Brito
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd

Python Patterns Guide

Objective explanations, simple examples, and a [Link] call for each pattern.

Quick rule of thumb


Use Strategy when behavior changes, Factory when creation changes, Adapter when an external API does not fit
your interface, Repository for persistence abstraction, Producer-Consumer for queues and async workloads,
Dependency Injection for testability, Ports and Adapters for clean architecture, and DDD when the business domain
should drive the model.

Strategy
Objective: Same contract, different behavior. Use it when you want to switch algorithms or execution logic without
changing the caller.

Example
from abc import ABC, abstractmethod

class PaymentStrategy(ABC):
@abstractmethod
def pay(self, amount: float) -> str:
...

class PixPayment(PaymentStrategy):
def pay(self, amount: float) -> str:
return f"Paid {amount} with PIX"

class CreditCardPayment(PaymentStrategy):
def pay(self, amount: float) -> str:
return f"Paid {amount} with Credit Card"

class Checkout:
def __init__(self, strategy: PaymentStrategy) -> None:
[Link] = strategy

def finish(self, amount: float) -> str:


return [Link](amount)

[Link] call
from strategy_example import Checkout, PixPayment, CreditCardPayment

def main():
checkout = Checkout(PixPayment())
print([Link](150.0))

checkout = Checkout(CreditCardPayment())
print([Link](150.0))

if __name__ == "__main__":
main()

Factory
Objective: Centralizes object creation. Use it when the caller should not know which concrete class is being
instantiated.

Example
class EmailNotifier:
def send(self, message: str) -> str:
return f"Email sent: {message}"

class SmsNotifier:
def send(self, message: str) -> str:
return f"SMS sent: {message}"

class NotifierFactory:
@staticmethod
def create(channel: str):
if channel == "email":
return EmailNotifier()
if channel == "sms":
return SmsNotifier()
raise ValueError("Unsupported channel")

[Link] call
from factory_example import NotifierFactory

def main():
notifier = [Link]("email")
print([Link]("Hello"))

if __name__ == "__main__":
main()

Adapter
Objective: Wraps an incompatible interface and exposes the interface your app expects.

Example
class ExternalRedisClient:
def rpush(self, queue_name: str, value: str) -> None:
print(f"Redis RPUSH {queue_name}: {value}")

class QueueAdapter:
def __init__(self, redis_client: ExternalRedisClient, queue_name: str) -> None:
self.redis_client = redis_client
self.queue_name = queue_name

def enqueue(self, message: str) -> None:


self.redis_client.rpush(self.queue_name, message)

[Link] call
from adapter_example import ExternalRedisClient, QueueAdapter

def main():
redis_client = ExternalRedisClient()
queue = QueueAdapter(redis_client, "events")
[Link]("device-online")

if __name__ == "__main__":
main()

Repository
Objective: Separates persistence from business logic. Your domain code talks to a repository, not directly to SQL or
an ORM session.

Example
class Device:
def __init__(self, device_id: str, status: str) -> None:
self.device_id = device_id
[Link] = status

class DeviceRepository:
def __init__(self) -> None:
self._storage: dict[str, Device] = {}

def save(self, device: Device) -> None:


self._storage[device.device_id] = device

def get_by_id(self, device_id: str) -> Device | None:


return self._storage.get(device_id)
[Link] call
from repository_example import Device, DeviceRepository

def main():
repo = DeviceRepository()
[Link](Device("dev-1", "online"))

device = repo.get_by_id("dev-1")
print([Link] if device else "not found")

if __name__ == "__main__":
main()

Observer / Pub-Sub
Objective: One event notifies multiple listeners. Good for alerts, logs, dashboards, and loosely coupled reactions.

Example
class EventBus:
def __init__(self) -> None:
self._subscribers: dict[str, list] = {}

def subscribe(self, event_name: str, callback) -> None:


self._subscribers.setdefault(event_name, []).append(callback)

def publish(self, event_name: str, payload: dict) -> None:


for callback in self._subscribers.get(event_name, []):
callback(payload)

def log_listener(payload: dict) -> None:


print("LOG:", payload)

def alert_listener(payload: dict) -> None:


print("ALERT:", payload)

[Link] call
from observer_example import EventBus, log_listener, alert_listener

def main():
bus = EventBus()
[Link]("[Link]", log_listener)
[Link]("[Link]", alert_listener)

[Link]("[Link]", {"device_id": "dev-1", "temp": 87})

if __name__ == "__main__":
main()

Producer-Consumer
Objective: Producers create work quickly; consumers process it later. Great for decoupling MQTT/API ingestion
from slower processing.

Example
import asyncio

queue: [Link][dict] = [Link]()

async def producer() -> None:


await [Link]({"device_id": "dev-1", "temp": 80})

async def worker() -> None:


while True:
message = await [Link]()
print("Processing", message)
queue.task_done()
[Link] call
import asyncio
from producer_consumer_example import producer, worker, queue

async def main():


worker_task = asyncio.create_task(worker())
await producer()
await [Link]()
worker_task.cancel()

if __name__ == "__main__":
[Link](main())

Dependency Injection
Objective: Pass dependencies from outside instead of creating them inside the class. This improves testing and
flexibility.

Example
class EmailService:
def send(self, message: str) -> None:
print("EMAIL:", message)

class NotificationManager:
def __init__(self, email_service: EmailService) -> None:
self.email_service = email_service

def notify(self, message: str) -> None:


self.email_service.send(message)

[Link] call
from di_example import EmailService, NotificationManager

def main():
email_service = EmailService()
manager = NotificationManager(email_service)
[Link]("Welcome")

if __name__ == "__main__":
main()

Ports and Adapters (Hexagonal)


Objective: Core code depends on abstractions (ports). Infrastructure implements them (adapters).

Example
from abc import ABC, abstractmethod

class MessageQueuePort(ABC):
@abstractmethod
async def enqueue(self, message: dict) -> None:
...

class InMemoryQueueAdapter(MessageQueuePort):
def __init__(self) -> None:
[Link]: list[dict] = []

async def enqueue(self, message: dict) -> None:


[Link](message)

class PublishTelemetryUseCase:
def __init__(self, queue: MessageQueuePort) -> None:
[Link] = queue

async def execute(self, payload: dict) -> None:


await [Link](payload)
[Link] call
import asyncio
from ports_adapters_example import InMemoryQueueAdapter, PublishTelemetryUseCase

async def main():


queue = InMemoryQueueAdapter()
use_case = PublishTelemetryUseCase(queue)
await use_case.execute({"device_id": "dev-1", "temp": 42})
print([Link])

if __name__ == "__main__":
[Link](main())

DDD (Domain-Driven Design) basics


Objective: Model the software around business concepts. Keep rules in the domain, orchestration in use cases,
persistence in repositories.

Example
class Device:
def __init__(self, device_id: str, max_temp: float) -> None:
self.device_id = device_id
self.max_temp = max_temp

def is_overheating(self, current_temp: float) -> bool:


return current_temp > self.max_temp

class DeviceRepository:
def __init__(self) -> None:
self._devices: dict[str, Device] = {
"dev-1": Device("dev-1", 75.0)
}

def get_by_id(self, device_id: str) -> Device | None:


return self._devices.get(device_id)

class AlertService:
def send(self, device_id: str, current_temp: float) -> None:
print(f"ALERT: {device_id} at {current_temp}")

class ProcessTelemetryUseCase:
def __init__(self, repo: DeviceRepository, alert_service: AlertService) -> None:
[Link] = repo
self.alert_service = alert_service

def execute(self, device_id: str, current_temp: float) -> None:


device = [Link].get_by_id(device_id)
if device and device.is_overheating(current_temp):
self.alert_service.send(device_id, current_temp)

[Link] call
from ddd_example import DeviceRepository, AlertService, ProcessTelemetryUseCase

def main():
repo = DeviceRepository()
alert_service = AlertService()
use_case = ProcessTelemetryUseCase(repo, alert_service)
use_case.execute("dev-1", 81.0)

if __name__ == "__main__":
main()

How to study these patterns


1. Learn the problem each pattern solves. 2. Recognize the tradeoff. 3. Practice small examples. 4. Map patterns to
real projects. 5. In interviews, explain why you chose the pattern, not just the name.

Interview tip
A strong answer is: 'I used Strategy because I had the same contract with different execution behaviors' or 'I used
Repository to keep SQL out of my domain logic.' That is better than naming patterns without context.

You might also like