Python

Python デザインパターンを学ぶ Commandパターン

  /

概要

Commandパターンは、一連のコマンドをオブジェクト（Invoker）にカプセル化したもの。これにより、ユーザーは事前に設定した順序で、後からコマンドを実行できる。また、 Invoker には、コマンドの失敗を処理する機能など、追加の動作をカプセル化できる。

以下の実装例では、データベースの移行システムをシミュレートする。一連の移行コマンドを、特定の順序で アトミックに実行する必要がある（すべて通過するか、何も通過しないか）。この場合、各コマンドはMigrationCommand（またはその子）のインスタンスとしてカプセル化され、マイグレーションを実行したり、ロールバックしたりすることができる。 Invoker は、失敗した場合に完了したすべてのマイグレーションをロールバックすることで、マイグレーションの順序とそのアトミック性を維持する責任がある。このようにして、 Invoker は、マイグレーション・コマンドのゆるい結合を作り、処理する。

実装例

Invokerクラスの_versionは、アトミックなロールバックのための変数。

class MigrationCommand:
    def __init__(self, title, instructions):
        self.title = title
        self.instructions = instructions

    def run(self):
        print("Run : ", self.title)

    def rollback(self):
        print("Roll Back : ", self.title)


class BadMigrationCommand(MigrationCommand):
    """失敗を試すための実装"""

    def run(self): raise RuntimeError("Something went wrong.")


class Invoker:

    def __init__(self, migrations):

        self.migrations = migrations
        self._version = 0

    def run_all(self):
        for migration in self.migrations:
            try:
                migration.run()
                self._version += 1
            except Exception as e:
                print("Migrations failed ", e.__class__.__name__)
                self.rollback_all()
                return False
                # raise e

        print("complete. Current Version: ", self._version)

    def rollback_all(self):

        if self._version == 0:
            print("Nothing rollback")
            return None

        for migration in reversed(self.migrations[:self._version]):
            migration.rollback()
            self._version -= 1

        print("Rollbacks complete. Current Version: ", self._version)


if __name__ == "__main__":

    migrations = [
        MigrationCommand(title="Command A", instructions={}),
        MigrationCommand(title="Command AB", instructions={}),
        BadMigrationCommand(title="Bad Command", instructions={}),
    ]

    migrations_invoker = Invoker(migrations)
    migrations_invoker.run_all()

    print("\n")

    del migrations_invoker.migrations[-1]
    migrations_invoker.run_all()
RELATED POST