The Python dictionary dispatch pattern

↑ Top

Introduction

Recently, I’ve been doing a lot of mentoring at work. Particularly around standards and patterns. Quite a bit is, obviously, proprietary but one super simple pattern a lot of my DevOps peers seem to not know is the dispatch pattern. In particular, Python’s flavour of this in the “dictionary dispatch” pattern.

So I’ve decided to do a little primer on said pattern, with some quite silly examples.

The goal

Within this blog post, I aim to:

• Briefly describe the “dispatch” pattern

• Describe the “dictionary dispatch” pattern

• Show an end-to-end example of such a pattern using a super simple Python script and mock RESTAPI backend

The actual setup

Prerequisites

I’m assuming:

• You know what a Python function is

• You what what a map is in Computer Science, and by extension what a dictionary in Python is

If not, I’ve hyperlinked various useful articles below:

• What is a Python function?

• What is a map?

• What is a dictionary?

In addition, you’ll need the following software on your laptop to follow along:

• Python 3.10 to write the code

The dispatch pattern

In simple terms, the problems this pattern tries to solve are as follows:

• Loose coupling

• Encapsulation

If you’re familiar with this concepts, or indeed the dispatch pattern as a whole, feel free to skip ahead to the dictionary dispatch pattern section and its examples.

If not, hopefully the below may be a useful primer.

Loose coupling

If we link together our chunky business logic with intermediate glue code, rather than having such logic directly call each other, we’re free to refactor entire subsections of code without updating numerous dependencies.

For example, if I’m making an automated robot to bake cookies and - at the moment - am using a Bosch oven to do so I can use the dispatcher pattern:

---title: Dispatcher example---flowchart LR    boschOven    Dispatcher    robot    Dispatcher -- directs to --> boschOven    robot -- consumes --> Dispatcher

But if Siemens suddenly come along and offer me a bulk discount in ovens, I might want to swap my bakeCookie logic to instead use their ovens:

---title: Dispatcher example---flowchart LR    siemensOven    Dispatcher    robot    Dispatcher -- directs to --> siemensOven    robot -- consumes --> Dispatcher

With this loose coupling - and the dispatcher sitting between my robot and oven logic - I don’t have to really change much at all to use my robot with a different oven.

This could even be extended to work both ways as the code gets more complicated, but exploring that isn’t needed at current for our basic example.

Encapsulation

Put simply, we organise related functionality together and hide the implementation behind an abstraction.

The examples given for loose coupling show this neatly. Our dispatcher may be relatively easily updated to work with multiple different oven types. The differences in implementation detail are hidden behind the dispatcher code. This is especially easily if we implement this pattern from the beginning, rather than refactoring later on, but we hardly have the luxury for shiny new codebases when working in enterprise environments.

The dictionary dispatch pattern

In Python, functions are first-class citizens. This means we can use a dictionary object to quickly and easily implement the dispatch pattern.

Cookie baking example

For example, continuing our cookie example from earlier we could implement our cookie baking skynet as follows:

import sys
from dataclasses import dataclass
from collections import defaultdict

@dataclass
class Cookie:
    """
    I'm a cookie
    """
    type: str

def bosh_oven(cookie: Cookie) -> None:
    """
    A function to bake cookies with bosch
    :param cookie: The cookie we wish to bake
    :return:
    """
    print(f"{cookie.type} cookie baked with bosch")

def siemens_oven(cookie: Cookie) -> None:
    """
    A function to bake cookies with siemens
    :param cookie: The cookie we wish to bake
    :return:
    """
    print(f"{cookie.type} cookie baked with siemens")

def default(cookie: Cookie) -> None:
    """
    A function to act as our default return for the dispatcher
    :param cookie: The cookie we wish to do something with
    :return:
    """
    print(f"Unable to interact with {cookie.type} cookie")

dispatcher = defaultdict(lambda: default)
dispatcher["bake"] = bosh_oven

if __name__ == "__main__":
    operation = sys.argv[1]
    yummy_cookie = Cookie(sys.argv[2])
    dispatcher[operation](yummy_cookie)

Thus:

% python3 -m robot eat chocolate
Unable to interact with chocolate cookie

% python3 -m robot bake chocolate
chocolate cookie baked with bosch

Here you can see the strength of the pattern. We may add new operations for our cookies by simply updating known keys in our dispatcher dictionary, or change the underlying implementation of business logic by swapping out the function returned by our dispatcher dictionary. Equally so, we may define useful defaults in a manner similar to a switch/case statement in a readable fashion.

Mock RESTAPI example

Where this pattern really shines for those in DevOps is when implementing RESTAPI backends. We may use it to greatly simplify adding new endpoints and methods within our RESTAPI backends.

Showcasing an entire RESTAPI backend made in this fashion is outside the scope of this post (but not in subsequent posts!), but we can easily see a mock example below. Using some simple I/O wrappers to persist data, sets for uniqueness, and two distinct endpoints we may see the flexibility of this pattern for more complex tasks.

Example RESTAPI code

import sys
from collections import defaultdict
import json
from abc import ABC, abstractmethod
from typing import Set

dogs = [
    "Cabbage",
    "Gravy",
    "Muffin"
]

class IDispatch(ABC):
    @abstractmethod
    def dispatch(self, method: str, payload: str):
        pass

class Dogs(IDispatch):
    """
    Provide functionality for /dogs endpoint
    """

    @staticmethod
    def dispatch(method: str, payload):
        """
        Simple method to dispatch requests to appropriate methods

        :param method: Which method we are requesting to dispatch to
        :param kwargs: Capture any un-used arguments
        :return: status_code and response for API query as int and str respectively
        """
        endpoints = {
            "get": Dogs.get
        }
        try:
            return endpoints[method.lower()]()
        except KeyError:
            return 405, f"'{method}' method not supported for endpoint"

    @staticmethod
    def get(**kwargs) -> (int, str):
        """
        Function for GET method

        :param kwargs: Capture any un-used arguments
        :return: Returns status_code and json string of all dogs known to API
        """
        return 200, json.dumps(dogs)

class Plants(IDispatch):
    """
    Provide functionality for /plants endpoint
    """

    def __init__(self):
        self.plants = Plants.read_plants_from_disk()

    def dispatch(self, method: str, payload: str):
        """
        Simple method to dispatch requests to appropriate methods

        :param method: Which method we are requesting to dispatch to
        :param payload: Payload of API request we shall mutate for method interaction
        :return: status_code and response for API query as int and str respectively
        """
        endpoints = {
            "get": Plants.get,
            "post": Plants.post
        }
        try:
            return endpoints[method.lower()](self, plant=payload)
        except KeyError:
            return 405, f"'{method}' method not supported for endpoint"

    def get(self, **kwargs) -> (int, str):
        """
        Function for GET method

        :param kwargs: Capture any un-used arguments
        :return: Returns status_code and json string of all plants known to API
        """
        # Sets are not serializable, but lists are, so we convert first.
        return 200, json.dumps(list(self.plants))

    def post(self, plant: str) -> (int, str):
        """
        Functionality for POST method

        :param plant: Which plant we wish to add
        :return: Returns status_code and string signifying operation status
        """
        if payload in self.plants:
            return 400, f"{payload} already managed by API."
        try:
            self.plants.update({payload})
            self.persist_plants_to_disk()
            return 200, f"Successfully added {payload} to managed plants"
        except Exception as e:
            return 400, str(e)

    def persist_plants_to_disk(self):
        """
        Method to persist current plants set to disk

        :return:
        """
        with open("plants.txt", "w") as plant_file:
            for plant in self.plants:
                plant_file.write(f"{plant}\n")

    @staticmethod
    def read_plants_from_disk() -> Set[str]:
        """
        Method to read plants from disk

        :return: Set containing all known plants
        """
        returned_plants = set()
        try:
            with open("plants.txt", "r") as plant_file:
                for line in plant_file:
                    returned_plants.update({line.strip()})
        except IOError:
            # Probably means we haven't persisted any plants to disk yet
            # Or, otherwise, outside scope of demo to remediate
            pass
        return returned_plants

dispatcher = defaultdict(lambda: {"status_code": 404, "response": "Endpoint not found"})
dispatcher["/plant"] = Plants()
dispatcher["/dog"] = Dogs()

if __name__ == "__main__":
    endpoint = sys.argv[1] if len(sys.argv) >= 2 else ""
    method = sys.argv[2] if len(sys.argv) >= 3 else ""
    payload = sys.argv[3] if len(sys.argv) >= 4 else ""

    endpoint_dispatcher = dispatcher[endpoint]
    returned_data = {"status_code": 400, "response": "Endpoint not found"}

    if issubclass(type(endpoint_dispatcher), IDispatch):
        status_code, response = endpoint_dispatcher.dispatch(method, payload)
        returned_data = {"status_code": status_code, "response": response}
    else:
        returned_data = endpoint_dispatcher

    print(json.dumps(returned_data))

With results as follows:

Non-supported endpoint

% python3 -m api /potato   
{"status_code": 404, "response": "Endpoint not found"}

Plant endpoint examples

% python3 -m api /plant GET
{"status_code": 200, "response": "[]"}

 % python3 -m api /plant POST cactus
{"status_code": 200, "response": "Successfully added cactus to managed plants"}

% python3 -m api /plant POST cactus
{"status_code": 400, "response": "cactus already managed by API."}

% python3 -m api /plant GET        
{"status_code": 200, "response": "[\"cactus\"]"}

% python3 -m api /plant POST lilly
{"status_code": 200, "response": "Successfully added lilly to managed plants"}

% python3 -m api /plant GET       
{"status_code": 200, "response": "[\"lilly\", \"cactus\"]"}

% python3 -m api /plant DELETE lilly
{"status_code": 405, "response": "'DELETE' method not supported for endpoint"}

Dog endpoint examples

% python3 -m api /dog GET           
{"status_code": 200, "response": "[\"Cabbage\", \"Gravy\", \"Muffin\"]"}

% python3 -m api /dog POST
{"status_code": 405, "response": "'POST' method not supported for endpoint"}

Conclusion

So that’s it, we’ve managed to successfully:

• Briefly describe the dispatcher pattern

• Describe the dictionary dispatcher pattern in Python

• Show some basic examples of this pattern in action for both:

  • A simple terminal script

  • A mock RESTAPI backend

The pattern here is nothing new, and is probably quite dull for those with a software engineering background. The examples hopefully show its utility for usual DevOps programming tasks.

But a simple pattern like this can go a long way. In my next blog post I’ll use a good use-case for this; implement actual backed API logic within a lambda, DynamoDB and API Gateway in an extensible, decoupled, clean fashion.

I hope you’ve enjoyed this little read, and found it somewhat useful.