На этом шаге мы рассмотрим реализацию этой задачи.

Задача

Вы хотите реализовать конечный автомат (объект, который может находиться в определенном количестве различных состояний), но не хотите замусоривать код большим количеством условий.

Решение

В некоторых приложениях вам могут понадобиться объекты, которые работают по-разному в зависимости от некого внутреннего состояния. Например, рассмотрим простой класс, представляющий соединение:

class Connection:
    def __init__(self):
        self.state = 'CLOSED'

def read(self):
    if self.state != 'OPEN':
        raise RuntimeError('Not open')
    print('reading')

def write(self, data):
    if self.state != 'OPEN':
        raise RuntimeError('Not open')
    print('writing')

def open(self):
    if self.state == 'OPEN':
        raise RuntimeError('Already open')
    self.state = 'OPEN'

def close(self):
    if self.state == 'CLOSED':
        raise RuntimeError('Already closed')
    self.state = 'CLOSED'

Эта реализация приводит к появлению нескольких трудных моментов. Во-первых, код переусложнен большим количеством условных проверок состояния. Во-вторых, производительность страдает из-за большого количества операций (например, read() и write() всегда проверяют состояние перед выполнением).

Более элегантный подход - закодировать каждое операционное состояние как отдельный класс, а класс Connection заставить делегировать операции классу состояния. Например:

>>> class Connection:
	def __init__(self):
		self.new_state(ClosedConnectionState)
	def new_state(self, newstate):
		self._state = newstate
	# Делегирует классу состояния
	def read(self):
		return self._state.read(self)
	def write(self, data):
		return self._state.write(self, data)
	def open(self):
		return self._state.open(self)
	def close(self):
		return self._state.close(self)

	
>>> # Базовый класс состояния соединения
>>> class ConnectionState:
	@staticmethod
	def read(conn):
		raise NotImplementedError()
	@staticmethod
	def write(conn, data):
		raise NotImplementedError()
	@staticmethod
	def open(conn):
		raise NotImplementedError()
	@staticmethod
	def close(conn):
		raise NotImplementedError()

	
>>> # Реализация различных состояний
>>> class ClosedConnectionState(ConnectionState):
	@staticmethod
	def read(conn):
		raise RuntimeError('Not open')
	@staticmethod
	def write(conn, data):
		raise RuntimeError('Not open')
	@staticmethod
	def open(conn):
		conn.new_state(OpenConnectionState)
	@staticmethod
	def close(conn):
		raise RuntimeError('Already closed')

	
>>> class OpenConnectionState(ConnectionState):
	@staticmethod
	def read(conn):
		print('reading')
	@staticmethod
	def write(conn, data):
		print('writing')
	@staticmethod
	def open(conn):
		raise RuntimeError('Already open')
	@staticmethod
	def close(conn):
		conn.new_state(ClosedConnectionState)

		
>>>

Вот пример использования этих классов:

>>> c = Connection()
>>> c._state
<class '__main__.ClosedConnectionState'>
>>> c.read()
Traceback (most recent call last):
  File "<pyshell#14>", line 1, in <module>
    c.read()
  File "<pyshell#3>", line 8, in read
    return self._state.read(self)
  File "<pyshell#9>", line 4, in read
    raise RuntimeError('Not open')
RuntimeError: Not open
>>> c.open()
>>> c._state
<class '__main__.OpenConnectionState'>
>>> c.read()
reading
>>> c.write('hello')
writing
>>> c.close()
>>> c._state
<class '__main__.ClosedConnectionState'>
>>>

Обсуждение

Код с большим количеством сложных проверок выполнения условий и связанных вместе состояний трудно поддерживать и понимать. Представленное решение обходит эту проблему путем выделения индивидуальных состояний в отдельные классы.

Это может показаться немного странным, однако каждое состояние реализовано как класс со статическими методами, каждый из которых принимает экземпляр Connection первым аргументом. Это проектировочное решение основано на отказе от хранения любых данных экземпляра в состояниях других классов. Вместо этого все данные экземпляра должны храниться в экземпляре Connection. Группирование состояний в общем базовом классе часто помогает упорядочить код и убедиться, что нужные методы реализованы. Исключение NotImplementedError возбуждается в методах базового класса и помогает убедиться, что подклассы предоставляют реализацию требуемых методов. В качестве альтернативы вы можете рассмотреть использование абстрактного базового класса, как то описано на 141 шаге. Альтернативная реализация затрагивает прямое управление атрибутом __class__ в экземплярах. Рассмотрите такой пример:

>>> class Connection:
    def __init__(self):
        self.new_state(ClosedConnection)

    def new_state(self, state):
        self.__class__ = state

    def read(self):
        raise NotImplementedError()

    def write(self, data):
        raise NotImplementedError()

    def open(self):
        raise NotImplementedError()

    def close(self):
        raise NotImplementedError()

>>> class ClosedConnection(Connection):
    def read(self):
        raise RuntimeError('Not open')

    def write(self, data):
        raise RuntimeError('Not open')

    def open(self):
        self.new_state(OpenConnection)

    def close(self):
        raise RuntimeError('Already closed')

>>> class OpenConnection(Connection):
    def read(self):
        print('reading')

    def write(self, data):
        print('writing')

    def open(self):
        raise RuntimeError('Already open')

    def close(self):
        self.new_state(ClosedConnection)

>>>

Основная особенность этой реализации в том, что она устраняет дополнительный слой "косвенности" (indirection). Вместо создания отдельных классов Connection и ConnectionState вы сливаете эти классы вместе. Когда меняется состояние, экземпляр изменяет свой тип, как показано тут:

>>> c = Connection()
>>> c
<__main__.ClosedConnection object at 0x000001E5E75181C0>
>>> c.read()
Traceback (most recent call last):
  File "<pyshell#30>", line 1, in <module>
    c.read()
  File "<pyshell#24>", line 3, in read
    raise RuntimeError('Not open')
RuntimeError: Not open
>>> c.open()
>>> c
<__main__.OpenConnection object at 0x000001E5E75181C0>
>>> c.read()
reading
>>> c.close()
>>> c
<__main__.ClosedConnection object at 0x000001E5E75181C0>
>>>

Пуристы от объектно-ориентированного программирования могут быть оскорблены идеей простого изменения атрибута экземпляра __class__(). Однако это технически возможно. Также это может ускорить выполнение программы, поскольку методы не используют дополнительный шаг делегирования.

И наконец, каждый из этих приемов полезен для реализации более сложных конечных автоматов, особенно в коде, который может послужить альтернативой огромным блокам if-elif-else. Например:

# Изначальная реализация
class State:
    def __init__(self):
        self.state = 'A'

    def action(self, x):
        if state == 'A':
            # Action for A
            . . .
            state = 'B'
        elif state == 'B':
            # Action for B
            . . .
            state = 'C'
        elif state == 'C':
            # Action for C
            . . .
            state = 'A'

# Альтернативная реализация
class State:
    def __init__(self):
        self.new_state(State_A)

    def new_state(self, state):
        self.__class__ = state

    def action(self, x):
        raise NotImplementedError()

class State_A(State):
    def action(self, x):
        # Действие для A
        . . .
        self.new_state(State_B)

class State_B(State):
    def action(self, x):
        # Действие для B
        . . .
        self.new_state(State_C)

class State_C(State):
    def action(self, x):
        # Действие для C
        . . .
        self.new_state(State_A)

Этот рецепт основан на паттерне (шаблоне) проектирования "Состояние" (State) из книги "Приемы объектно-ориентированного проектирования. Паттерны проектирования" Эриха Гаммы, Ричарда Хелма, Ральфа Джонсона и Джона Влиссидеса.

https://ru.wikipedia.org/wiki/Design_Patterns.

На следующем шаге мы рассмотрим вызов метода объекта с передачей имени метода в строке.

Предыдущий шаг Содержание Следующий шаг