28. Design Patterns

28.1. Examples

28.1.1. Singleton

Code Listing 28.1. Singleton Design Pattern
class DB:
    connection = None

    def __init__(self):

    def connect():
        if not DB.connection:
            print('Nawiazujemy nowe polaczenie')
            DB.connection = ...

        return DB.connection

# Bedzie sie laczyl do bazy danych
conn = DB().connect()

# uzyje juz istniejacego polaczenia
conn = DB().connect()

28.1.2. Gateway

Code Listing 28.2. Gateway Design Pattern
import datetime
import json
import logging
import os
import requests

log = logging.getLogger(__name__)

class HTTPGateway:
    def __init__(self, username=None, password=None, cache_expiry_days=30, cache_directory='.'):
        self.username = username
        self.password = password
        self.cache_expiry_days = cache_expiry_days
        self.cache_directory = cache_directory

    def _get_cache_name_from_url(url):
        return url.replace('/', '-').replace(':', '-')

    def _fetch_from_url(self, url):
        connection = requests.get(url, auth=(self.username, self.password))

        if connection.status_code != 200:
            log.error(f'Cannot fetch from URL: {url}')
            raise ConnectionError
            log.debug(f'Fetched from {url}')
            return connection.text

    def _fetch_from_cache(self, url):
        cache_name = self._get_cache_name_from_url(url)
        path = os.path.join(self.cache_directory, cache_name)

        with open(path) as file:
            log.debug(f'Reading from cache file {path}')
            return file.read()

    def _set_cache(self, url, data):
        cache_name = self._get_cache_name_from_url(url)
        path = os.path.join(self.cache_directory, cache_name)

        with open(path, 'w') as file:
            log.debug(f'Writing to cache file {path}')

    def _cache_invalid(self, url):
        def last_modification(filepath, since):
            modification_timestamp = os.path.getmtime(filepath)
            modification_datetime = datetime.datetime.fromtimestamp(modification_timestamp)
            now = datetime.datetime.now()
            return (now - modification_datetime).days < since

        cache_name = self._get_cache_name_from_url(url)
        path = os.path.join(self.cache_directory, cache_name)

        if os.path.isfile(path) and last_modification(path, since=self.cache_expiry_days):
            return False
            return True

    def get(self, url):
        if self._cache_invalid(url):
            log.info(f'Will read from URL {url}')
            data = self._fetch_from_url(url)
            self._set_cache(url, data)
            log.info(f'Will read from cache')
            data = self._fetch_from_cache(url)

        return json.loads(data)

http = HTTPGateway(

# pobranie i ustawienie cache
html1 = http.get('http://python.astrotech.io')

# wczytanie z cache, bez komunikacji z internetem
html2 = http.get('http://python.astrotech.io')

28.1.3. Factory

Code Listing 28.3. Factory Design Pattern
import os

class HttpClientInterface:
    def GET(self):
        raise NotImplementedError

    def POST(self):
        raise NotImplementedError

class GatewayLive(HttpClientInterface):
    def GET(self):
        """execute GET request over network"""

    def POST(self):
        """execute POST request over network"""

class GatewayStub(HttpClientInterface):
    def GET(self):
        return {'first_name': 'Jose', 'last_name': 'Jimenez'}

    def POST(self):
        return {'status': 200, 'reason': 'OK'}

class HttpClientFactory:
    instance = None

    def __new__(cls, *args, **kwargs):
        if not cls.instance:
            if os.getenv('ENVIRONMENT') == 'production':
                cls.instance = GatewayLive()
                cls.instance = GatewayStub()

        return cls.instance

client = HttpClientFactory()
out = client.GET()

client2 = HttpClientFactory()
out1 = client2.GET()
out2 = client2.POST()

Code Listing 28.4. Factory Design Pattern
class ConfigParserInterface:
    extension = None

    def __init__(self, filename):
        self.filename = filename

    def read(self):
        with open(self.filename) as file:
            content = file.read()
            return self.parse(content)

    def parse(self, content):
        return NotImplementedError

class ConfigParserINI(ConfigParserInterface):
    extension = '.ini'

    def parse(self, content):
        print('Parsing INI file')
        return dict(...)

class ConfigParserCSV(ConfigParserInterface):
    extension = '.csv'

    def parse(self, content):
       print('Parsing CSV file')
       return dict()

class ConfigParserYAML(ConfigParserInterface):
    extension = '.yaml'

    def parse(self, content):
       print('Parsing YAML file')
       return dict()

class ConfigFileJSON(ConfigParserInterface):
    extension = '.json'

    def parse(self, content):
       print('Parsing JSON file')
       return dict()

class ConfigFileXML(ConfigParserInterface):
    extension = '.xml'

    def parse(self, content):
       print('Parsing XML file')
       return dict()

def config_parser_factory(filename):
    import os
    parsers = {p.extension: p for p in ConfigParserInterface.__subclasses__()}
    extension = os.path.splitext(filename)[1]

        return parsers[extension](filename)
    except KeyError:
        raise NotImplementedError

 # iris.csv or *.csv, *.json *.yaml...
filename = input('Type filename: ')

config_parser = config_parser_factory(filename)

28.1.4. Dependency Injection

Code Listing 28.5. Dependency Injection Design Pattern
from datetime import timedelta

class Cache:
    def __init__(self, expiration=timedelta(days=30), location=None):
        self.expiration = expiration
        self.location = location

    def get(self):
        raise NotImplementedError

    def set(self):
        raise NotImplementedError

    def is_valid(self):
        raise NotImplementedError

class CacheFilesystem(Cache):
    """Cache using files"""

class CacheMemory(Cache):
    """Cache using memory"""

class CacheDatabase(Cache):
    """Cache using database"""

class HTTP:
    def __init__(self, cache):
        # Inject Cache object
        self._cache = cache

    def _fetch(self, url):
        return ...

    def get(self, url):
        if self._cache.is_valid():
            # Use cached data
            data = self._fetch(url)
            self._cache.set(url, data)

if __name__ == '__main__':
    database = CacheDatabase(location='sqlite3://http-cache.sqlite3')
    filesystem = CacheFilesystem(location='/tmp/http-cache.txt')
    memory = CacheMemory(expiration=timedelta(hours=2))

    http1 = HTTP(cache=database)

    http2 = HTTP(cache=filesystem)

    http3 = HTTP(cache=memory)

28.1.5. Callback

Code Listing 28.6. Callback Design Pattern
from http import HTTPStatus
import requests

def noop(*arg, **kwargs):

def http_request(url, on_success=noop, on_error=noop):
    result = requests.get(url)
    if result.status_code == HTTPStatus.OK:

def success(result):

def error(result):


28.1.6. State Machine

  • StateMachine imposes a structure to automatically change the implementation from one object to the next
  • The current implementation represents the state that a system is in
  • System behaves differently from one state to the next
  • The code that moves the system from one state to the next
  • Each state can be run() to perform its behavior
  • You can pass it an input object so it can tell you what new state to move to based on that input
  • Each State object decides what other states it can move to, based on the input
  • Each State object has its own little State table
  • There is a single master state transition table for the whole system
statemachine TrafficLight:
    Red -> Green
    Green -> Amber
    Amber -> Red

Red.wait = sleep(2)
Amber.wait = sleep(1)
Green.wait = sleep(2)
Code Listing 28.7. State Machine
from time import sleep

class Light:
    def __init__(self, previous=None):
        self.previous = previous

    def run(self):
        raise NotImplementedError

    def __next__(self):
        raise NotImplementedError

class Red(Light):
    color = 'Red'
    wait = 2

    def run(self):

    def __next__(self):
        return Amber(previous=self)

class Amber(Light):
    color = 'Amber'
    wait = 1

    def run(self):

    def __next__(self):
        if isinstance(self.previous, Red):
            return Green(previous=self)
            return Red(previous=self)

class Green(Light):
    color = 'Green'
    wait = 2

    def run(self):

    def __next__(self):
        return Amber(previous=self)

class TrafficLights:
    def __init__(self, initial_state=Green(), max_changes=10):
        self.state = initial_state
        self.max_changes = max_changes

    def __iter__(self):
        self.changes = 0
        return self

    def __next__(self):
        if self.changes >= self.max_changes:
            raise StopIteration

        self.changes += 1
        self.state = next(self.state)
        return self

for light in TrafficLights(max_changes=10):

28.2. Structural Design Patterns

  • Adapter (klasowy i obiektowy)
  • Most (ang. Bridge) (obiektowy)
  • Kompozyt (ang. Composite) (obiektowy)
  • Dekorator (ang. Decorator) (obiektowy)
  • Fasada (ang. Façade) (obiektowy)
  • Pyłek (ang. Flyweight) (obiektowy)
  • Pełnomocnik (ang. Proxy) (obiektowy)

28.3. Creational Design Patterns

  • Metoda wytwórcza (ang. Factory Method) (klasowy)
  • Fabryka Abstrakcyjna (ang. Abstract Factory) (obiektowy)
  • Budowniczy (ang. Builder) (obiektowy)
  • Prototyp (ang. Prototype) (obiektowy)
  • Singleton (obiektowy)

28.4. Behavioral Design Patterns

  • Łańcuch zobowiązań (ang. Chain of Responsibility) (obiektowy)
  • Polecenie (ang. Command) (obiektowy)
  • Interpreter (ang. Interpreter) (klasowy)
  • Interator (obiektowy)
  • Mediator (ang. Mediator) (obiektowy)
  • Pamiątka (ang. Memento) (obiektowy)
  • Obserwator (ang. Observer) (obiektowy)
  • Stan (ang. State) (obiektowy)
  • Strategia (ang. Strategy) (obiektowy)
  • Metoda szablonowa (ang. Template Method) (klasowy)
  • Odwiedzający (ang. Visitor) (obiektowy)

28.5. Idiomy języka programowania

  • Wzorzec EFAP (ang. It’s easier to ask for forgiveness than permission)
  • Wzorzec Metaklasy
  • Borg
  • Klasa domieszkowa w języku Python (ang. Mixin)