Патерн «Замісник» у Python

У Python, як і в будь-якій мові програмування, є патерни та антипатерни проєктування. Хоч патерни і є загальними рішеннями для поширених проблем програмування, у кожній мові є свої особливості реалізації. Наприклад, в динамічних мовах проєктування патерни додають рівень абстракції, який ускладнює розуміння імплементації.

Динамічна сутність Python, а також функції першого класу роблять більшість патернів класичних ООП-мов надлишковими. Тому замість складних інженерних підходів у Python віддають перевагу об'єктам першого класу, неявній типізації, заміні методів і значень атрибутів класу програми під час виконання – тобто роблять все, щоб виконати задачу швидше. Однак не завжди такий підхід виправданий.

У цій статті ми розглянемо патерн, який не складно реалізувати, але з ним код стане набагато якіснішим. Знайомимось із «Замісником».

Патерн «Замісник» (Proxy)

Перш ніж перейдемо до термінів, спробуймо розібратись, в чому суть патерну «Замісник» на прикладі з життя.

Ви колись користувалися ключем-карткою, щоб відчинити двері? Зазвичай це робиться або спеціальною карткою, або кнопкою для зняття системи безпеки. Основна дія дверей в такому випадку — відчинитись, але «Замісник» додає ще деяку функціональність. Розглянемо його реалізацію мовою Python:

class Door:
    def open_method(self) -> None:
        pass


class SecuredDoor:
    def __init__(self) -> None:
        self._klass = Door()

    def open_method(self) -> None:
        print(f"Adding security measure to the method of {self._klass}")


secured_door = SecuredDoor()
secured_door.open_method()

>>> Adding security measure to the method of <__main__.Door object at 0x7f9dab3b6670>

Тут клас Door має один єдиний метод під назвою open_method, який відповідає за те, щоб відчинити об'єкт Door. Цей клас наслідується класом SecuredDoor, який розширює метод виводом в консоль деякої інформації.

Зверніть увагу на те, як клас Door був викликаний класом SecuredDoor за допомогою композиції. З патерном «Замісник» ви можете замінити основний об'єкт об'єктом-замісником без будь-яких додаткових дій. Такий підхід відповідає принципу заміщення Барбари Лісков. Він звучить так:

Об'єкти суперкласу можна замінити об'єктами підкласів, не зламавши застосунок. Для цього необхідно, аби об'єкти підкласів поводились так само, як їхні суперкласи.

Об'єкт Door можна замінити об'єктом SecuredDoor, адже підклас не оголошує додаткових методів, а лише розширює функціональність наявного методу класу Door.

Ще трохи термінів:

З патерном «Замісник» ми можемо одним класом замінити інший, не зламавши застосунок.

Якщо звернутись до Вікіпедії: «Замісник» в ПЗ — це клас, що функціонує як інтерфейс для іншої сутності застосунку. Об'єкт-замісник є обгорткою, з якою взаємодіє клієнт, а за лаштунками розташований об'єкт, який і виконує основну функціональність. Об'єкт-замісник може просто перенаправляти на основний об'єкт, або ж додавати логіку. Наприклад, можлива організація кешування, якщо ресурси основного об'єкта потребують складних та дорогих обчислень, або ж перевірка попередніх умов, перш ніж виконати операцію в основному об'єкті.

Патерн «Замісник» належить до групи структурних патернів.

У чому переваги патерну

Слабке зв'язування

Патерн «Замісник» чудово справляється з розділенням основної логіки та додаткової функціональності. Модульна природа коду робить підтримку та розширення коду основної логіки більш швидким та легким.

Припустимо, що нам треба створити функцію division, яка прийматиме два аргументи типу integer, а повертатиме результат ділення між ними. Функція також обробляє граничні випадки помилками ZeroDivisionError і TypeError.

import logging
from typing import Union

logging.basicConfig(level=logging.INFO)


def division(a: Union[int, float], b: Union[int, float]) -> float:
    try:
        result = a / b
        return result

    except ZeroDivisionError:
        logging.error(f"Argument b cannot be {b}")

    except TypeError:
        logging.error(f"Arguments must be integers/floats")


print(division(1.9, 2))

>>> 0.95

Ця функція тричі порушила принцип єдиної відповідальності. Його суть в тому, що функція або клас повинні мати лише одну причину для зміни. В нашому прикладі функція відповідає за три речі, які призводять до її зміни. Таку функцію складно змінювати та розширювати.

Ми можемо покращити наш код, якщо напишемо два класи. Основний клас Division відповідатиме за логіку ділення, а от ProxyDivision розширюватиме функціональність Division обробниками винятків та логерами.

import logging
from typing import Union

logging.basicConfig(level=logging.INFO)


class Division:
    def div(self, a: Union[int, float], b: Union[int, float]) -> float:
        return a / b


class ProxyDivision:
    def __init__(self) -> None:
        self._klass = Division()

    def div(self, a: Union[int, float], b: Union[int, float]) -> float:
        try:
            result = self._klass.div(a, b)
            return result

        except ZeroDivisionError:
            logging.error(f"Argument b cannot be {b}")

        except TypeError:
            logging.error(f"Arguments must be integers/floats")


klass = ProxyDivision()
print(klass.div(2, 0))

>>> ERROR:root:Argument b cannot be 0
    None

Оскільки класи Division та ProxyDivision реалізують однаковий інтерфейс, вони взаємозамінні. Другий клас не наслідується напряму від першого і не додає додаткових методів. Тобто ви можете з легкістю написати інший клас для розширення функціональності Division та ProxyDivision, не зачіпаючи їхню внутрішню логіку напряму.

Покращене тестування

«Замісник» допомагає покращити код так, що його стає простіше тестувати. Оскільки ваша основна логіка слабко зв'язана з додатковою функціональністю, тестувати все можна в ізоляції. Такі тести більш лаконічні та модульні.

Використовуючи класи Division та ProxyDivision, можемо продемонструвати всі переваги слабкого зв'язування в тестуванні. Так логіку основного класу легко відстежити. Оскільки він містить важливу функціональність, то саме для нього і варто написати модульні тести в першу чергу, а вже потім тестувати додаткові фічі. З розділенням логік клас Division стає значно простішим для тестування, аніж попередня функція division, що відповідає за декілька дій. Одразу як протестуєте основний клас, можна переходити до іншої функціональності. Зазвичай, розділення ключової логіки та інкапсуляція додаткової функціональності допомагає писати більш надійні та точні модульні тести.

Патерн «Замісник» з інтерфейсом

В реальних проєктах ваші класи навряд чи будуть такими ж простими, як Division, що складається з одного методу. Зазвичай основний клас містить декілька методів, які займаються досить складними завданнями.

Сподіваємось, ви вже зрозуміли, що класи-замісники повинні реалізовувати всі методи основного класу. Часто це правило забувають при створенні класу-замісника, якщо основний клас досить складний. А це вже відхилення від правильної архітектури.

Розв'язати проблему можна за допомогою інтерфейсу, який буде сповіщати автора класу замісника про методи, які необхідно реалізувати. Інтерфейс — це повністю абстрактний клас, який визначає методи, котрі необхідно імплементувати конкретному класу. Однак інтерфейси неможливо ініціалізувати незалежно. Вам потрібен клас, який буде реалізовувати інтерфейс, тобто всі його методи. Якщо ви цього не зробите, виникне помилка. Розглянемо простий приклад, як можна створити інтерфейс в Python за допомогою abc.ABC та abc.abstractmethod і використати його для створення «Замісника».

from abc import ABC, abstractmethod


class Interface(ABC):
    """Interfaces of Interface, Concrete & Proxy should
    be the same, because the client should be able to use
    Concrete or Proxy without any change in their internals.
    """

    @abstractmethod
    def job_a(self, user: str) -> None:
        pass

    @abstractmethod
    def job_b(self, user: str) -> None:
        pass


class Concrete(Interface):
    """This is the main job doer. External services like
    payment gateways can be a good example.
    """

    def job_a(self, user: str) -> None:
        print(f"I am doing the job_a for {user}")

    def job_b(self, user: str) -> None:
        print(f"I am doing the job_b for {user}")


class Proxy(Interface):
    def __init__(self) -> None:
        self._concrete = Concrete()

    def job_a(self, user: str) -> None:
        print(f"I'm extending job_a for user {user}")

    def job_b(self, user: str) -> None:
        print(f"I'm extending job_b for user {user}")


if __name__ == "__main__":
    klass = Proxy()
    print(klass.job_a("red"))
    print(klass.job_b("nafi"))

>>> I'm extending job_a for user red
    None
    I'm extending job_b for user nafi
    None

З прикладу очевидно, що спочатку треба визначити Interface. Python пропонує базові абстрактні класи як ABC у модулі abc. Абстрактний клас Interface наслідується від ABC та оголошує всі методи, які пізніше треба буде реалізувати конкретному класу Concrete. Зверніть увагу, що кожен метод класу Interface позначений декоратором @abstractmethod. Якщо вам треба підтягнути власні знання з декораторів, то автор рекомендує подивитися цей матеріал.

Декоратор @abstractmethod перетворює звичайний метод в абстрактний, тобто в зразок методу, який необхідно реалізувати конкретному класу. Ви не можете напряму створити екземпляр Interface або використати будь-який з його абстрактних методів без його реалізації.

Клас Concrete реалізує створений нами інтерфейс, тобто всі його методи позначені як абстрактні. Це конкретний клас, екземпляр якого можна створити, а методи використовувати напряму. Якщо ж ви забудете реалізувати будь-який з методів інтерфейсу, виникне помилка TypeError.

Клас Proxy розширює функціональність базового класу Concrete. Він посилається на Concrete за допомогою композиції та реалізує всі його методи. З таким підходом ми використовуємо результати конкретних методів, але до того ж розширюємо їхню функціональність без повторювання коду.

Ще один приклад для закріплення

Аби зрозуміти концепцію ще краще, розглянемо реальний приклад. Припустимо, вам треба зберегти дані, отримані зі стороннього API ендпоінта. Для цього ви надсилаєте GET-запит з вашого http-клієнта і зберігаєте відповідь у форматі json. Можливо, далі ви захочете перевірити заголовки відповіді (headers) та аргументи (arguments), з якими надсилався запит.

За звичайних умов публічне API вводить обмеження на кількість запитів. Якщо ви його перевищите, то, найімовірніше, отримаєте повідомлення про те, що час очікування http-запиту сплив. Припустимо, ви хотіли б обробити такий тип помилки поза основною логікою, що відповідає за надсилання запитів GET.

Мабуть, ви також хотіли б закешувати відповіді, які вже надходили на такі ж аргументи. Якщо ви вже відправляли запит з такими параметрами декілька разів, , клієнт повертає відповіді з кешу — замість того щоб робити черговий запит на API. Кешування значно покращує швидкість отримання відповіді з API.

Для демонстрації візьмемо публічний API Postman.

https://postman-echo.com/get?foo1=bar_1&foo2=bar_2

Цей API ідеально підходить для демонстрації, оскільки має обмеження на кількість запитів, яке діє довільно, а клієнт повертає помилки ConnectTimeOut та ReadTimeOutError. Всі дії необхідно виконувати в такому порядку:

1. Оголосіть інтерфейс IFetchUrl з трьома абстрактними методами. Перший метод get_data отримуватиме дані з URL та серіалізуватиме їх у формат JSON. Наступний метод get_headers візьме зразок даних та поверне заголовки у вигляді словника. І останній метод, get_args, подібно до попереднього методу, візьме дані, але цього разу поверне аргументи запита у вигляді словника. Однак для реалізації всі цих методів потрібен конкретний клас.
2. Створимо конкретний клас FetchUrl і реалізуємо усі три методи інтерфейсу IFetchUrl. Тут не варто турбуватись про граничні випадки. В методах повинна бути лише основна логіка, без додаткової функціональності.
3. Створіть клас-замісник ExcFetchUrl. Він також реалізовуватиме інтерфейс, однак додатково буде відповідати за логіку обробки помилок та логування. В ньому можна викликати методи класу FetchUrl за допомогою композиції. Ми уникаємо повторювання коду, адже методи вже визначені в основному класі. Оскільки ExcFetchUrl реалізовує інтерфейс IFetchUrl, то йому необхідно реалізувати також всі його методи.
4. Наостанок створюємо клас, який розширює ExcFetchUrl і додає функцію кешування для get_data. Цей клас необхідно створити за тим самим шаблоном, що й ExcFetchUrl.

Нарешті ми ознайомились з послідовністю реалізації патерну «Замісник» в реальному проєкті. Наше рішення зайняло 110 рядків коду:

import logging
import sys
from abc import ABC, abstractmethod
from datetime import datetime
from pprint import pprint

import httpx
from httpx._exceptions import ConnectTimeout, ReadTimeout
from functools import lru_cache


logging.basicConfig(level=logging.INFO)


class IFetchUrl(ABC):
    """Abstract base class. You can't instantiate this independently."""

    @abstractmethod
    def get_data(self, url: str) -> dict:
        pass

    @abstractmethod
    def get_headers(self, data: dict) -> dict:
        pass

    @abstractmethod
    def get_args(self, data: dict) -> dict:
        pass


class FetchUrl(IFetchUrl):
    """Concrete class that doesn't handle exceptions and loggings."""

    def get_data(self, url: str) -> dict:
        with httpx.Client() as client:
            response = client.get(url)
            data = response.json()
            return data

    def get_headers(self, data: dict) -> dict:
        return data["headers"]

    def get_args(self, data: dict) -> dict:
        return data["args"]


class ExcFetchUrl(IFetchUrl):
    """This class can be swapped out with the FetchUrl class.
    It provides additional exception handling and logging."""

    def __init__(self) -> None:
        self._fetch_url = FetchUrl()

    def get_data(self, url: str) -> dict:
        try:
            data = self._fetch_url.get_data(url)
            return data

        except ConnectTimeout:
            logging.error("Connection time out. Try again later.")
            sys.exit(1)

        except ReadTimeout:
            logging.error("Read timed out. Try again later.")
            sys.exit(1)

    def get_headers(self, data: dict) -> dict:
        headers = self._fetch_url.get_headers(data)
        logging.info(f"Getting the headers at {datetime.now()}")
        return headers

    def get_args(self, data: dict) -> dict:
        args = self._fetch_url.get_args(data)
        logging.info(f"Getting the args at {datetime.now()}")
        return args


class CacheFetchUrl(IFetchUrl):
    def __init__(self) -> None:
        self._fetch_url = ExcFetchUrl()

    @lru_cache(maxsize=32)
    def get_data(self, url: str) -> dict:
        data = self._fetch_url.get_data(url)
        return data

    def get_headers(self, data: dict) -> dict:
        headers = self._fetch_url.get_headers(data)
        return headers

    def get_args(self, data: dict) -> dict:
        args = self._fetch_url.get_args(data)
        return args


if __name__ == "__main__":

    # url = "https://postman-echo.com/get?foo1=bar_1&foo2=bar_2"

    fetch = CacheFetchUrl()
    for arg1, arg2 in zip([1, 2, 3, 1, 2, 3], [1, 2, 3, 1, 2, 3]):
        url = f"https://postman-echo.com/get?foo1=bar_{arg1}&foo2=bar_{arg2}"
        print(f"\
 {'-'*75}\
")
        data = fetch.get_data(url)
        print(f"Cache Info: {fetch.get_data.cache_info()}")
        pprint(fetch.get_headers(data))
        pprint(fetch.get_args(data))

---------------------------------------------------------------------------

INFO:root:Getting the headers at 2020-06-16 16:54:36.214562
INFO:root:Getting the args at 2020-06-16 16:54:36.220221
Cache Info: CacheInfo(hits=0, misses=1, maxsize=32, currsize=1)
{'accept': '*/*',
    'accept-encoding': 'gzip, deflate',
    'content-length': '0',
    'host': 'postman-echo.com',
    'user-agent': 'python-httpx/0.13.1',
    'x-amzn-trace-id': 'Root=1-5ee8a4eb-4341ae58365e4090660dfaa4',
    'x-b3-parentspanid': '044bd10726921994',
    'x-b3-sampled': '0',
    'x-b3-spanid': '503e6ceaa2a4f493',
    'x-b3-traceid': '77d5b03fe98fcc1a044bd10726921994',
    'x-envoy-external-address': '10.100.91.201',
    'x-forwarded-client-cert': 'By=spiffe://cluster.local/ns/pm-echo-istio/sa/default;Hash=2ed845a68a0968c80e6e0d0f49dec5ce15ee3c1f87408e56c938306f2129528b;Subject="";URI=spiffe://cluster.local/ns/istio-system/sa/istio-ingressgateway-service-account',
    'x-forwarded-port': '443',
    'x-forwarded-proto': 'http',
    'x-request-id': '295d0b6c-7aa0-4481-aa4d-f47f5eac7d57'}
{'foo1': 'bar_1', 'foo2': 'bar_1'}

....

В методі get_data класу FetchUrl автор використав http-клієнт httpx для отримання даних за URL. Зверніть увагу, що заради спрощення не врахована додаткова логіка обробки помилок та логування. За реалізацію цієї логіки відповідає клас-замісник ExcFetchUrl. Ще один клас CacheFetchUrl розширює клас-замісник і додає функціональність кешування для функції get_data.

В основній частині ви можете використовувати будь-який з перелічених класів без додаткових змін до їх логіки. Клас FetchUrl повідомить, коли виникне помилка, а CacheFetchUrl та ExcFetchUrl реалізовують додаткову функціональність, при цьому імплементуючи однаковий інтерфейс.

Як результат виконання коду ми отримуємо заголовки та аргументи запиту, які повертаються методами get_headers та get_args. Також звернуть увагу, що автор зімітував кешування за допомогою аргументів ендпоінта. Заголовок Cache Info: в третьому рядку показує, що дані повернені з кешу. Значення hits=0 означає, що дані отримані напряму зі стороннього API. Однак якщо дослідити шари виводу, ви побачите, що аргументи запиту повторюються. Заголовок Cache Info: також повідомить про найбільшу кількість потраплянь. Це означатиме, що дані отримані з кешу.

Чи варто використовувати патерн

Так, звичайно. Але варто аналізувати ситуацію. Перш ніж братися до реалізації «Замісника», трохи сплануйте вашу архітектуру. Якщо ви пишете невеликий скрипт, який не плануєте підтримувати довго, не обов'язково все ускладнювати додатковими шарами абстракції з ООП-світу. Вона робить ваш код складним для розуміння.

З іншого боку, патерн «Замісник» допоможе, коли вам треба реалізувати додаткову функціональність для деякого класу, оскільки це ідеальне рішення для слабкого зв'язування. Тож користуйтеся патерном розсудливо.

Примітки

Усі приклади коду в матеріалі написані й тестовані мовою Python 3.8 та запущені на Ubuntu 18.04.