Behavourial
Observer / PubSub
It's common for different components of an app to respond to events or state changes, but how can we communicate these events?
The Observer pattern is a popular solution. We have a Subject (aka Publisher) which will be the source of events. And we could have multiple Observers (aka Subscribers) which will recieve events from the Subject in realtime.
class YoutubeChannel:
def __init__(self, name):
self.name = name
self.subscribers = []
def subscribe(self, sub):
self.subscribers.append(sub)
def notify(self, event):
for sub in self.subscribers:
sub.sendNotification(self.name, event)
from abc import ABC, abstractmethod
class YoutubeSubscriber(ABC):
@abstractmethod
def sendNotification(self, event):
pass
class YoutubeUser(YoutubeSubscriber):
def __init__(self, name):
self.name = name
def sendNotification(self, channel, event):
print(f"User {self.name} received notification from {channel}: {event}")
channel = YoutubeChannel("neetcode")
channel.subscribe(YoutubeUser("sub1"))
channel.subscribe(YoutubeUser("sub2"))
channel.subscribe(YoutubeUser("sub3"))
channel.notify("A new video released")
In this case we have multiple Subscribers listening to a single published. But users could also be subscribed to multiple channels.
Since the Publishers & Subscribers don't have to worry about each others' implementations, they are loosely coupled.
User sub1 received notification from neetcode: A new video released
User sub2 received notification from neetcode: A new video released
User sub3 received notification from neetcode: A new video released
Iterator
Many objects in python have built-in iterators. That's why we can conveniently iterate through an array using the key word in.
myList = [1, 2, 3]
for n in myList:
print(n)
Output
1
2
3
For more complex objects, like Linked Lists or Binary Search Trees, we can define our own iterators.
class ListNode:
def __init__(self, val):
self.val = val
self.next = None
class LinkedList:
def __init__(self, head):
self.head = head
self.cur = None
# Define Iterator
def __iter__(self):
self.cur = self.head
return self
# Iterate
def __next__(self):
if self.cur:
val = self.cur.val
self.cur = self.cur.next
return val
else:
raise StopIteration
# Initialize LinkedList
head = ListNode(1)
head.next = ListNode(2)
head.next.next = ListNode(3)
myList = LinkedList(head)
# Iterate through LinkedList
for n in myList:
print(n)
Output
1
2
3
Strategy
A Class may have different behaviour, or invoke a different method based on something we define (i.e. a Strategy). For example, we can filter an array by removing positive values; or we could filter it by removing all odd values. These are two filtering strategies we could implement, but we could add many more.
from abc import ABC, abstractmethod
class FilterStrategy(ABC):
@abstractmethod
def removeValue(self, val):
pass
class RemoveNegativeStrategy(FilterStrategy):
def removeValue(self, val):
return val < 0
class RemoveOddStrategy(FilterStrategy):
def removeValue(self, val):
return abs(val) % 2
class Values:
def __init__(self, vals):
self.vals = vals
def filter(self, strategy):
res = []
for n in self.vals:
if not strategy.removeValue(n):
res.append(n)
return res
values = Values([-7, -4, -1, 0, 2, 6, 9])
print(values.filter(RemoveNegativeStrategy()))
print(values.filter(RemoveOddStrategy()))
Output
[0, 2, 6, 9]
[-4, 0, 2, 6]
A common alternative to this pattern is to simply pass in an inline / lambda function, which allows us to extend the behaviour of a method or class.
State Machines
I
from discovereddataclasses Stateimport Machinesdataclass
aboutfrom 2typing yearsimport ago,Protocol
itclass wasDocumentState(Protocol):
useddef inedit(self):
a solution to a problem where we were mapping the possible states of a VoIP phone call (incoming, ringing, answered, etc.) to something that we could monitor. I was amazed at how simple this was and decided to adopt state machines in my own projects.
I’m sure everyone knows what happens when you discover some new software principle; you decide that literally everything you’ve ever built needs it. However, as responsible developers, we must ask ourselves whether what we're trying to do is the best solution. We must ask what our use case is and, in this case, whether we even need a state machine. Perhaps the example below will help provide insight into what state machines can be used for.
STATE MACHINES
Before we get into implementing a simple state machine in Python lets quickly go over what a state machine is and what it looks like. In short, it’s basically a set of states and a set of actions/events, we start at one state and like a graph data structure, we can traverse through the nodes based on the condition described by the corresponding edge. We use that condition to get to the node (aka state) we’d like to. Since there is only a single state active at a time, we have a lot of control in terms of where we are within the lifecycle of the state machine. There’s a more thorough CS theory backed explanation that can be found by means of a video here...
Idef highlyreview(self):
recommend...
checkingdef itfinalize(self):
out...
ifclass you’dDocumentContext(Protocol):
likecontent: tolist[str]
knowdef more!
set_state(self, USEstate: CASES
There are many use cases for state machines, some of which include — managing states (like call states, WiFi connectivity, and even the Android activity life cycle) or reporting metrics — duration of time for a user to complete a login (loginDocumentState) -> pendingNone:
->...
success)def foredit(self):
example.
...
Statedef machinesreview(self):
are...
especiallydef interestingfinalize(self):
because,...
amongdef othershow_content(self):
things,...
they@dataclass
provideclass well-definedDraft:
scenariosdocument: andDocumentContext
listdef outedit(self):
print("Editing the conditionsdocument...")
self.document.content.append("Edited content.")
def review(self):
print("The document is now under review.")
self.document.set_state(Reviewed(self.document))
def finalize(self):
print("You need to get to them. This makes it very easy to scope out edge cases and how to handle them, as we are forced to consider every possible scenario our code must fall within.
Personally,review the bestdocument waybefore tofinalizing.")
understand@dataclass
stateclass machinesReviewed:
document: DocumentContext
def edit(self):
print("The document is throughunder anreview, everydaycannot example.
edit Imaginenow.")
youdef arereview(self):
looking at your password protected phone, at a high-level it has two states of operation. print("The firstdocument beingis locked,already wherereviewed.")
youdef havefinalize(self):
limited functionality andprint("Finalizing the seconddocument...")
beingself.document.set_state(Finalized(self.document))
unlocked,@dataclass
whereclass youFinalized:
candocument: nowDocumentContext
usedef theedit(self):
deviceprint("The in a greater capacity.
Thisdocument is whatfinalized. theEditing aboveis statenot machineallowed.")
looksdef likereview(self):
whenprint("The visualized.
document 
is USINGfinalized. AReview STATEis MACHINE
not Wepossible.")
begindef byfinalize(self):
definingprint("The thedocument states,is thesealready arefinalized.")
defined as the nodes within the state machine. In our case, we have two states; locked & unlocked. In the example below, I've also defined a State object which will handle some utility functions for our states (which extend from this object).
# state.py
class State(object):Document:
"""
We define a state object which provides some utility functions for the
individual states within the state machine.
"""
def __init__(self):
printself.state: 'ProcessingDocumentState current= Draft(self)
self.content: list[str] = []
def set_state(self, state:', str(self)DocumentState):
def on_event(self, event):
"""
Handle events that are delegated to this State.
"""
pass
def __repr__(self):
"""
Leverages the __str__ method to describe the State.
"""
return self.__str__()
def __str__(self):
"""
Returns the name of the State.
"""
return self.__class__.__name__
The states can then be defined as follows.
# my_states.py
from state import State
# Start of our states
class LockedState(State):
"""
The state which indicates that there are limited device capabilities.
"""
def on_event(self, event):
if event == 'pin_entered':
return UnlockedState()
return self
class UnlockedState(State):
"""
The state which indicates that there are no limitations on device
capabilities.
"""
def on_event(self, event):
if event == 'device_locked':
return LockedState()
return self
# End of our states.
Then we define the actual state machine. It's fairly simple and looks like this:
# simple_device.py
from my_states import LockedState
class SimpleDevice(object):
"""
A simple state machine that mimics the functionality of a device from a
high level.
"""
def __init__(self):
""" Initialize the components. """
# Start with a default state.
self.state = LockedState()state
def on_event(self,edit(self):
event)self.state.edit()
def review(self):
self.state.review()
def finalize(self):
self.state.finalize()
def show_content(self):
print("Document content:", " This".join(self.content))
def main() -> None:
document = Document()
document.edit() # Expected: "Editing the document..."
document.show_content() # Expected: "Document content: Edited content."
document.finalize() # Expected: "You need to review the document before finalizing."
document.review() # Expected: "The document is thenow breadunder andreview."
butterdocument.edit() of# Expected: "The document is under review, cannot edit now."
document.finalize() # Expected: "Finalizing the statedocument..."
machine.document.edit() Incoming# eventsExpected: are
delegated to the given states which then handle the event. "The result is
then assigned as the new state.
"""
# The next state will be the result of the on_event function.
self.state = self.state.on_event(event)
Fairly simple right? What this state machine doesdocument is definesfinalized. a starting state LockedState and exposes a function to handle events. This function basically assigns the current state to the result of that same state when it handles the event.
Finally, we can test that implementation of the state machine using the python shell.
$ python
Python 2.7.13 (default, Apr 4 2017, 08:47:57)
[GCC 4.2.1 Compatible Apple LLVM 8.1.0 (clang-802.0.38)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>>
>>> from simple_device import SimpleDevice
>>> device = SimpleDevice()
Processing current state: LockedState
>>>
>>> device.on_event('device_locked')
>>> device.on_event('pin_entered')
Processing current state: UnlockedState
>>>
>>> device.state
UnlockedState
>>>
>>> device.on_event('device_locked')
Processing current state: LockedState
>>>
>>> device.state
LockedState
>>> device.on_event('device_locked')
You’ll notice that duplicate events are ignored and only the events that provide transitions are made use of. This becomes a very powerful tool when we want to ignore events or reduce a series of verbose events into a simpler set of events. My favorite thingEditing is thatnot allowed."
if in__name__ the== future"__main__":
we'd like to add more states and transitions, it's very simple to do so without rewriting a huge chunk of our codebase.
CONCLUSION
main()State machines are awesome, from cases that require simple state management, to metric reporting, they have proven to be very useful and extensible. The above technique was a product of implementing a state machine to handle SIP signaling events (for VoIP) and measure deltas between incoming events (to gain a better understanding of our pain points). It can definitely scale to a few dozen states and makes for a simple and easy state measurement.
If you are looking for a different solution, check out the Python transitions library which is a state machine library that looks fairly promising.
REFERENCES
While researching state machine libraries for Python I came across a page that documented a simple implementation, the above solution based on the example provided. Check out the original implementationhere.
Let me know if you spot any errors, or if you just want to say hi!