Command

Command

Command patterns encapsulate requests as objects, and thereby allow for parameterization of clients with different requests. It turns the request into a stand-alone object that contains all information about that requests; this allows requests to be passed as method arguments, delay, or queue a request’s execution, and to support undoable operations.

TL;DR

The Command pattern allows actions to be defined as objects; this is primarily done to decouple clients from recievers. Gives the ability to perform actions or trigger events at a later time.

Problem

When implementing a database, for example, you might need the ability to create and execute migrations. At the same time, tightly coupling the database logic itself with the command interface makes the design of the application very difficult to maintain:

# Tightly coupled - bad
class MigrationRunner:
    def run_migration(self, operation_type, table_name):
        if operation_type == "create":
            self.database.add_table(table_name)
        elif operation_type == "drop":
            self.database.remove_table(table_name)
        # What if you need to add 20 more operation types?

Additionally, whenever a command is executed we might need the ability to easily rollback or undo the command, or to delay its execution (for example, to batch requests), or to maintain a log of the steps executed.

Solution

The Command pattern is made up of three major parts:

1. The command interface
2. The Invoker
3. The Receiver

The Command interface defines a common method to execute a method, and often includes the reverse of that same method (do/undo, for example). The receiver is the part of the program the does the actual work, such as a Database class that interacts with the database itself - this includes methods to add, remove, and modify tables, for example.

Finally, there is the invoker, which manages and executes commands, often maintaining a history of what commands were executed.

With this, we can create objects that encapsulate specific database operations as a command, such as:

drop_posts = DropTableCommand(db, "posts")

This drop_posts object can then be fed into the invoker class which can execute the command, or undo it as necessary:

# Simple Command Pattern Example - Database Migrations
 
from abc import ABC, abstractmethod
 
# Command interface
class Command(ABC):
    @abstractmethod
    def execute(self):
        pass
    
    @abstractmethod
    def undo(self):
        pass
 
# Receiver - Database that performs the actual work
class Database:
    def __init__(self):
        self.tables = []
    
    def add_table(self, table_name):
        self.tables.append(table_name)
        print(f"Added table: {table_name}")
    
    def remove_table(self, table_name):
        if table_name in self.tables:
            self.tables.remove(table_name)
            print(f"Removed table: {table_name}")
 
# Concrete Commands
class CreateTableCommand(Command):
    def __init__(self, database, table_name):
        self.database = database
        self.table_name = table_name
    
    def execute(self):
        self.database.add_table(self.table_name)
    
    def undo(self):
        self.database.remove_table(self.table_name)
 
class DropTableCommand(Command):
    def __init__(self, database, table_name):
        self.database = database
        self.table_name = table_name
    
    def execute(self):
        self.database.remove_table(self.table_name)
    
    def undo(self):
        self.database.add_table(self.table_name)
 
# Invoker - Manages and executes commands
class MigrationRunner:
    def __init__(self):
        self.history = []
    
    def run_command(self, command):
        command.execute()
        self.history.append(command)
    
    def undo_last(self):
        if self.history:
            last_command = self.history.pop()
            last_command.undo()
            print("Undid last command")
 
# Usage example
if __name__ == "__main__":
    # Create database and migration runner
    db = Database()
    runner = MigrationRunner()
    
    # Create commands
    create_users = CreateTableCommand(db, "users")
    create_posts = CreateTableCommand(db, "posts")
    drop_posts = DropTableCommand(db, "posts")
    
    # Run migrations
    print("Running migrations:")
    runner.run_command(create_users)
    runner.run_command(create_posts)
    
    print(f"\nCurrent tables: {db.tables}")
    
    # Undo last migration
    print("\nUndoing last migration:")
    runner.undo_last()
    
    print(f"Tables after undo: {db.tables}")
    
    # Run another command
    print("\nDropping posts table:")
    runner.run_command(drop_posts)
    
    print(f"Final tables: {db.tables}")