All 23 GoF design patterns as reusable, Pythonic, type-safe implementations.
Built with Python 3.12+ features: PEP 695 generics, type statements, Self return types, and full async support. Zero dependencies.
pip install pydesignpatternsNote: The package is installed as
pydesignpatternsbut imported aspypatterns:from pypatterns import singleton, Subject, Builder
from pypatterns import singleton, Subject, Builder, chain, Handler
# Singleton via decorator
@singleton
class Config:
def __init__(self):
self.debug = False
assert Config() is Config() # always the same instanceThree decorator variants — no metaclass magic needed.
from pypatterns import singleton, thread_safe_singleton, resettable_singleton
@singleton
class AppConfig:
def __init__(self):
self.db_url = "sqlite:///app.db"
@thread_safe_singleton
class ConnectionPool:
"""Safe for concurrent access."""
pass
@resettable_singleton
class Cache:
"""Call Cache.reset() in tests to get a fresh instance."""
passAbstract creator + a ready-made typed registry.
from pypatterns import Creator, Registry, registry_factory
# Classic approach
class Dialog(Creator["Button"]):
def factory_method(self) -> "Button":
return Button()
# Registry approach — no subclassing needed
animals: Registry[str, Animal] = registry_factory()
animals.register("dog", Dog)
animals.register("cat", lambda: Cat("whiskers"))
dog = animals.create("dog")
assert "dog" in animals
print(animals.keys) # ["dog", "cat"]Produce families of related objects.
from pypatterns import AbstractFactory, AbstractProduct
class Button(AbstractProduct):
pass
class GUIFactory(AbstractFactory):
def create_button(self) -> Button: ...
def create_checkbox(self) -> "Checkbox": ...
class MaterialFactory(GUIFactory):
def create_button(self) -> Button:
return MaterialButton()
def create_checkbox(self) -> "Checkbox":
return MaterialCheckbox()Step-by-step construction with fluent method chaining.
from pypatterns import Builder, Director
from typing import Self
class QueryBuilder(Builder[str]):
def __init__(self):
self._parts: list[str] = []
def select(self, fields: str) -> Self:
self._parts.append(f"SELECT {fields}")
return self
def from_table(self, table: str) -> Self:
self._parts.append(f"FROM {table}")
return self
def where(self, condition: str) -> Self:
self._parts.append(f"WHERE {condition}")
return self
def build(self) -> str:
return " ".join(self._parts)
query = QueryBuilder().select("*").from_table("users").where("active = 1").build()
# "SELECT * FROM users WHERE active = 1"Clone objects without depending on their classes.
from pypatterns import prototype, PrototypeRegistry
@prototype
class Shape:
def __init__(self, x: int, y: int):
self.x = x
self.y = y
self.tags: list[str] = []
original = Shape(10, 20)
original.tags.append("red")
cloned = original.clone() # deep copy
cloned.tags.append("blue")
assert original.tags == ["red"] # original unchanged
assert cloned.tags == ["red", "blue"]
# Named prototype registry
registry = PrototypeRegistry[str]()
registry.register("origin", Shape(0, 0))
point = registry.clone("origin")Bridge incompatible interfaces with attribute mapping.
from pypatterns import Adapter, adapt
class LegacyPrinter:
def print_old(self, text: str) -> str:
return f"[OLD] {text}"
# Quick functional adapter
printer = adapt(LegacyPrinter(), print_text="print_old")
printer.print_text("hello") # "[OLD] hello"
# Class-based adapter
adapter = Adapter(adaptee=LegacyPrinter(), mapping={"print_text": "print_old"})Separate abstraction from implementation.
from pypatterns import Abstraction, RefinedAbstraction
class TV:
def operation_impl(self) -> str:
return "TV output"
class Radio:
def operation_impl(self) -> str:
return "Radio output"
remote = Abstraction(TV())
remote.operation() # "TV output"
remote.impl = Radio() # swap at runtime
remote.operation() # "Radio output"Build tree structures and treat them uniformly.
from pypatterns import Composite, Leaf
class File(Leaf[int]):
def __init__(self, size: int):
self.size = size
def operation(self) -> int:
return self.size
class Directory(Composite[int]):
def operation(self) -> int:
return sum(child.operation() for child in self)
root = Directory()
root.add(File(100)).add(File(200))
sub = Directory()
sub.add(File(50))
root += sub # operator support
assert root.operation() == 350
assert len(root) == 3Stack behaviors on objects at runtime. This is the GoF structural pattern, not Python's @decorator syntax.
from pypatterns import DecoratorBase, ComponentProto, stackable
class TextComponent(ComponentProto):
def operation(self) -> str:
return "Hello"
class BracketDecorator(DecoratorBase):
def operation(self) -> str:
return f"[{self._wrapped.operation()}]"
class StarDecorator(DecoratorBase):
def operation(self) -> str:
return f"*{self._wrapped.operation()}*"
# Stack multiple decorators
result = stackable(BracketDecorator, StarDecorator)(TextComponent())
result.operation() # "*[Hello]*"Simplify complex subsystems behind a clean interface.
from pypatterns import Facade
class VideoConverter(Facade):
def __init__(self):
self.codec = CodecFactory()
self.mixer = AudioMixer()
self.encoder = BitrateEncoder()
def convert(self, filename: str, fmt: str) -> str:
codec = self.codec.extract(filename)
audio = self.mixer.fix(codec)
return self.encoder.encode(audio, fmt)Share common state between objects to save memory.
from pypatterns import FlyweightFactory, flyweight
# Decorator approach — instances cached by args
@flyweight
class TreeType:
def __init__(self, name: str, color: str):
self.name = name
self.color = color
a = TreeType("Oak", "green")
b = TreeType("Oak", "green")
assert a is b # same instance reused
# Factory approach
factory = FlyweightFactory[str, dict]()
fw = factory.get_or_create("oak", {"name": "Oak", "color": "green"})
assert len(factory) == 1Control access to objects with lazy-loading, protection, and logging variants.
from pypatterns import LazyProxy, ProtectiveProxy, LoggingProxy, lazy_proxy
# Lazy — defers creation until first use
proxy = lazy_proxy(lambda: ExpensiveService())
# ExpensiveService not created yet
proxy.do_work() # now created
# Protective — guard attribute access
proxy = ProtectiveProxy(service, guard=lambda name: name != "secret")
proxy.public_method() # OK
proxy.secret # raises PermissionError
# Logging — track all access
logs: list[str] = []
proxy = LoggingProxy(service, logger=logs.append)
proxy.method() # logs "Accessing 'method'"Pass requests along a chain of handlers. Supports >> operator.
from pypatterns import Handler, chain
class AuthHandler(Handler[dict]):
def handle(self, request: dict) -> dict | None:
if not request.get("authenticated"):
return None # reject
return super().handle(request)
class LogHandler(Handler[dict]):
def handle(self, request: dict) -> dict | None:
print(f"Processing: {request}")
return super().handle(request)
# Chain with >> operator
pipeline = AuthHandler() >> LogHandler()
pipeline.handle({"authenticated": True, "data": "hello"})
# Or use the chain() helper
pipeline = chain(AuthHandler(), LogHandler())Encapsulate requests as objects with undo/redo support.
from pypatterns import SimpleCommand, CommandInvoker
class AddTextCommand(SimpleCommand):
def __init__(self, doc: list[str], text: str):
self.doc = doc
self.text = text
def execute(self) -> None:
self.doc.append(self.text)
def undo(self) -> None:
self.doc.remove(self.text)
doc: list[str] = []
invoker = CommandInvoker()
invoker.submit(AddTextCommand(doc, "Hello"))
invoker.submit(AddTextCommand(doc, " World"))
# doc == ["Hello", " World"]
invoker.undo() # doc == ["Hello"]
invoker.redo() # doc == ["Hello", " World"]
assert invoker.can_undo
assert len(invoker.history) == 2Enhanced iterators beyond Python's built-in __iter__.
from pypatterns import PeekableIterator, FilterIterator, BatchIterator, MappingIterator
# Peek without consuming
it = PeekableIterator([1, 2, 3])
assert it.peek() == 1 # still there
assert next(it) == 1 # now consumed
# Filter
evens = FilterIterator([1, 2, 3, 4, 5], lambda x: x % 2 == 0)
assert list(evens) == [2, 4]
# Batch processing
batches = BatchIterator(range(7), batch_size=3)
assert list(batches) == [[0, 1, 2], [3, 4, 5], [6]]
# Transform
doubled = MappingIterator([1, 2, 3], lambda x: x * 2)
assert list(doubled) == [2, 4, 6]Decouple components through a central event hub.
from pypatterns import EventMediator, Colleague
mediator = EventMediator()
class Button(Colleague):
def click(self) -> None:
self.mediator.notify(self, "click", button="submit")
class Dialog(Colleague):
def __init__(self, mediator):
super().__init__(mediator)
self.submitted = False
dialog = Dialog(mediator)
button = Button(mediator)
mediator.on("click", lambda sender, **kw: setattr(dialog, "submitted", True))
button.click()
assert dialog.submittedSave and restore object state with undo support.
from pypatterns import Originator, Memento, Caretaker
import time
class Editor(Originator[str]):
def __init__(self):
self.content = ""
def save(self) -> Memento[str]:
return Memento(state=self.content, timestamp=time.time())
def restore(self, memento: Memento[str]) -> None:
self.content = memento.state
editor = Editor()
caretaker = Caretaker[str]()
editor.content = "v1"
caretaker.backup(editor)
editor.content = "v2"
caretaker.backup(editor)
editor.content = "v3"
caretaker.undo(editor) # back to "v2"
caretaker.undo(editor) # back to "v1"Pub/sub event system. Accepts both protocol objects and plain callables.
from pypatterns import Subject, Observer
subject = Subject[str]()
# With callables
messages: list[str] = []
subject.subscribe(messages.append)
# With Observer objects
class Logger:
def __init__(self):
self.logs: list[str] = []
def update(self, event: str) -> None:
self.logs.append(event)
logger = Logger()
subject.subscribe(logger)
subject.notify("user_login")
assert messages == ["user_login"]
assert logger.logs == ["user_login"]
subject.unsubscribe(logger)
assert subject.observer_count == 1Objects change behavior when their internal state changes.
from pypatterns import State, Context
class Draft(State["Document"]):
def handle(self) -> str:
return "Document is a draft"
class Published(State["Document"]):
def handle(self) -> str:
return "Document is published"
class Document(Context):
pass
doc = Document(Draft())
assert doc.request() == "Document is a draft"
doc.transition_to(Published())
assert doc.request() == "Document is published"Swap algorithms at runtime. Works with both objects and plain callables.
from pypatterns import StrategyContext
# With callables — the most Pythonic way
ctx = StrategyContext[list[int], list[int]](sorted)
ctx.execute([3, 1, 2]) # [1, 2, 3]
ctx.set_strategy(lambda x: sorted(x, reverse=True))
ctx.execute([3, 1, 2]) # [3, 2, 1]
# With Strategy objects
class BubbleSort:
def execute(self, data: list[int]) -> list[int]:
return sorted(data)
ctx.set_strategy(BubbleSort())Define algorithm skeleton, let subclasses fill in the steps.
from pypatterns import TemplateMethod
class DataPipeline(TemplateMethod):
def __init__(self):
self.data: list[str] = []
def step_one(self) -> None: # Extract
self.data = ["raw1", "raw2"]
def step_two(self) -> None: # Transform
self.data = [d.upper() for d in self.data]
def step_three(self) -> None: # Load
print(f"Loaded {len(self.data)} records")
def hook_before(self) -> None: # Optional hook
print("Starting pipeline...")
pipeline = DataPipeline()
pipeline.template_method()
# Starting pipeline...
# Loaded 2 recordsSeparate algorithms from the objects they operate on using automatic type dispatch.
from pypatterns import visitor
@visitor
class Serializer:
def visit_int(self, element: int) -> str:
return f"<int>{element}</int>"
def visit_str(self, element: str) -> str:
return f"<str>{element}</str>"
def visit_float(self, element: float) -> str:
return f"<float>{element}</float>"
s = Serializer()
s.visit(42) # "<int>42</int>"
s.visit("hello") # "<str>hello</str>"
s.visit(3.14) # "<float>3.14</float>"Observer, Chain of Responsibility, Command, and Mediator all have async variants:
from pypatterns import AsyncSubject, AsyncHandler, AsyncCommandInvoker, AsyncEventMediator
# Async Observer — notifies concurrently via asyncio.gather
subject = AsyncSubject[str]()
async def log_event(event: str) -> None:
print(f"Got: {event}")
subject.subscribe(log_event)
await subject.notify("hello")
# Async Chain of Responsibility
class AsyncAuth(AsyncHandler[dict]):
async def handle(self, request: dict) -> dict | None:
if not request.get("token"):
return None
return await super().handle(request)
# Async Command with undo/redo
invoker = AsyncCommandInvoker()
await invoker.submit(async_command)
await invoker.undo()
# Async Event Mediator — handlers run concurrently
mediator = AsyncEventMediator()
mediator.on("save", async_save_handler)
await mediator.notify(self, "save", data=payload)# Import everything from the top level
from pypatterns import singleton, Subject, Builder, Handler, chain
# Or import by category
from pypatterns.creational import singleton, Builder, registry_factory
from pypatterns.structural import Proxy, lazy_proxy, Composite
from pypatterns.behavioral import Subject, AsyncSubject, Handler, CommandInvoker| Category | Exports |
|---|---|
| Creational | singleton, thread_safe_singleton, resettable_singleton, Creator, Registry, registry_factory, AbstractFactory, AbstractProduct, Builder, Director, Prototype, prototype, PrototypeRegistry |
| Structural | Adapter, adapt, Abstraction, Implementor, RefinedAbstraction, Component, Composite, Leaf, ComponentProto, DecoratorBase, stackable, Facade, Flyweight, FlyweightFactory, flyweight, Proxy, LazyProxy, ProtectiveProxy, LoggingProxy, lazy_proxy, protective_proxy, logging_proxy |
| Behavioral | Handler, AsyncHandler, chain, async_chain, Command, SimpleCommand, CommandInvoker, AsyncCommand, AsyncCommandInvoker, IterableCollection, PeekableIterator, FilterIterator, MappingIterator, BatchIterator, Mediator, Colleague, EventMediator, AsyncEventMediator, Memento, Originator, Caretaker, Observer, Subject, AsyncObserver, AsyncSubject, State, Context, Strategy, StrategyContext, TemplateMethod, Visitable, BaseVisitor, visitor |
- Protocol over ABC when only structural conformance is needed
- PEP 695 generics (
class Foo[T]) throughout Selfreturn types for fluent APIs (Builder, Composite, Chain)- Operator overloads where natural:
>>for chaining,+=for composites - Decorators as the Pythonic entry point for Singleton, Prototype, Flyweight, Visitor
- Async is additive — async variants live alongside sync in the same module
All patterns follow the catalog at refactoring.guru/design-patterns.
MIT