Книга «Объектно-ориентированный Python, 4-е изд.»

Привет, Хаброжители!

Глубоко погрузитесь в различные аспекты объектно-ориентированного программирования на Python, паттерны проектирования, приемы манипулирования данными и вопросы тестирования сложных объектно-ориентированных систем. Обсуждение всех понятий подкрепляется примерами, написанными специально для этого издания, и практическими упражнениями в конце каждой главы. Код всех примеров совместим с синтаксисом Python 3.9+ и дополнен аннотациями типов для упрощения изучения.

Стивен и Дасти предлагают вашему вниманию понятный и всесторонний обзор важных концепций ООП, таких как наследование, композиция и полиморфизм, и объясняют их работу на примерах классов и структур данных Python, что заметно облегчает проектирование. В тексте широко используются UML-диаграммы классов, чтобы было проще понять взаимоотношения между классами. Помимо ООП, в книге подробно рассматривается обработка исключений в Python, а также приемы функционального программирования, пересекающиеся с приемами ООП. В издании представлены не одна, а две очень мощные системы автоматического тестирования: unittest и pytest, а в последней главе детально обсуждается экосистема параллельного программирования в Python.

Получите полное представление о том, как применять принципы объектно-ориентированного программирования с использованием синтаксиса Python, и научитесь создавать надежные и устойчивые программы.

Паттерн Декоратор

Паттерн Декоратор позволяет добавлять объектам новое поведение, помещая их в объекты-оболочки. Декоратор оборачивает объекты бесчисленное количество раз благодаря тому, что и обертки, и сами обернутые объекты имеют общий интерфейс. Любой объект, использующий объект-декоратор, будет взаимодействовать с ним точно так же, как если бы он не был декоратором (то есть интерфейс объекта-декоратора идентичен интерфейсу основного объекта).

Существует два способа реализации паттерна Декоратор:

• улучшение отклика компонента при отправке данных второму компоненту;
• поддержка нескольких дополнительных вариантов поведения.

Второй вариант обычно является подходящей альтернативой множественному наследованию. Мы можем создать основной объект, а затем декоратор, обертывающий это ядро. Поскольку объект-декоратор имеет тот же интерфейс, что и основной объект, можно даже обернуть и новый объект в другие декораторы. Рассмотрим, как это выглядит на UML-диаграмме (рис. 11.1).

В данном случае ядро и все декораторы реализуют определенный интерфейс. Пунктирные линии означают «реализацию». Декораторы поддерживают ссылку на основной экземпляр данного интерфейса через композицию. При вызове декоратор выполняет некоторую дополнительную обработку до или после вызова своего обернутого интерфейса. Обернутый объект может быть другим декоратором или основным функционалом. Хотя несколько декораторов могут обертывать друг друга, основную функциональность обеспечивает объект в конце цепочки всех декораторов.

Важно, чтобы каждый из объектов в цепочке обеспечивал реализацию общей функции. Цель состоит в том, чтобы предоставить набор шагов обработки ядра различными декораторами. Как правило, декораторы невелики по размеру, фактически это определение функции без какого-либо состояния.

В Python благодаря утиной типизации нет необходимости формализовать эти отношения с помощью официального определения абстрактного интерфейса. Достаточно убедиться, что классы имеют соответствующие методы. В некоторых случаях можно определить typing.Protocol как подсказку типа, помогающую анализировать отношения.

Пример реализации паттерна Декоратор

Рассмотрим пример из сетевого программирования. Скажем, необходимо создать небольшой сервер и клиент. Сервер предоставляет данные, клиент с ним взаимодействует. Сервер имитирует бросание костей. Клиент отправляет запрос и ожидает ответ, содержащий случайные числа.

В этом примере два процесса взаимодействуют через сокет TCP — способ передачи байтов между компьютерными системами. Сокеты создаются сервером, прослушивающим соединения. При попытке клиента подключиться к сокету сервер должен принять новое соединение, после чего два процесса смогут передавать байты в обе стороны. Для данного примера это будет запрос от клиента к серверу и ответ от сервера к клиенту. Сокет TCP является частью основы протокола HTTP, вокруг которого построена Всемирная паутина.

При клиент-серверном взаимодействии для передачи строки байтов через сокет будет использоваться метод socket.send(), а для получения байтов — метод socket.recv().

Проанализируем этот пример, начиная с интерактивного сервера, который ожидает подключения от клиента, а затем отвечает на запрос. Назовем этот модуль socket_server.py. Например:

import contextlib
import socket

def main_1() -> None:
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.bind(("localhost", 2401))
    server.listen(1)
    with contextlib.closing(server):
        while True:
            client, addr = server.accept()
            dice_response(client)
            client.close()

Сервер привязан к общедоступному сокету через произвольный порт с номером 2401. Именно здесь сервер прослушивает запросы на подключение. При попытке клиента подключиться к этому сокету создается дочерний сокет, чтобы клиент и сервер могли общаться, оставляя общедоступный сокет готовым для дополнительных подключений. Чтобы обеспечить большое количество одновременных сессий веб-сервер часто применяет несколько потоков. Мы не используем здесь потоки, и второй клиент должен ждать, пока сервер не закончит работу с первым клиентом. Это напоминает очередь в кофейню, где только один бариста готовит эспрессо.

Обратите внимание, что сокеты TCP/IP имеют как адрес хоста, так и номер порта. Номер порта должен быть выше чем 1023. Номера портов 1023 и ниже зарезервированы и требуют специальных привилегий ОС. В примере выбран порт 2401, так как он не используется для других задач.

Функция dice_response() выполняет всю реальную работу описанного сервиса. Для ответа клиенту функция принимает параметр сокета. Считываются байты с клиентским запросом, создается ответ, который затем отправляется. Функция dice_response() для корректной обработки исключений выглядит следующим образом:

def dice_response(client: socket.socket) -> None:
    request = client.recv(1024)
    try:
        response = dice.dice_roller(request)
    except (ValueError, KeyError) as ex:
        response = repr(ex).encode("utf-8")
    client.send(response)

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

import random

def dice_roller(request: bytes) -> bytes:
    request_text = request.decode("utf-8")
    numbers = [random.randint(1, 6) for _ in range(6)]
    response = f"{request_text} = {numbers}"
    return response.encode("utf-8")

Это не слишком сложно. Данный пример более подробно рассмотрим в разделе, касающемся паттерна Команда, но уже здесь в качестве ответа предоставляется последовательность случайных чисел.

Обратите внимание, что фактически мы ничего не делаем с объектом request, пришедшим от клиента. В примерах мы будем считывать эти байты и игнорировать их. Request является плейсхолдером для более сложного запроса, описывающего необходимое количество игральных костей и количество бросков.

Паттерн проектирования Декоратор иногда используется для добавления функций. Он обертывает основную функцию dice_response(), которой предоставляется объект socket, а его уже можно и читать, и записывать. Чтобы использовать паттерн проектирования, важно учитывать, как при добавлении функционала эта функция опирается на методы socket.send() и socket.recv(): необходимо сохранить определение интерфейса.

Чтобы протестировать сервер, напишем код простого клиента, который подключается к тому же порту и перед выходом выводит ответ:

import socket

def main() -> None:
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.connect(("localhost", 2401))
    count = input("How many rolls: ") or "1"
    pattern = input("Dice pattern nd6[dk+-]a: ") or "d6"
    command = f"Dice {count} {pattern}"
    server.send(command.encode("utf8"))
    response = server.recv(1024)
    print(response.decode("utf-8"))
    server.close()

if __name__ == "__main__":
    main()

Клиент задает два вопроса и создает довольно сложную строку, command, содержащую подсчет и паттерн броска костей. На данный момент сервер еще не использует эту команду, и код представляет собой задачу для более сложной игры в кости.

Чтобы использовать два приведенных отдельных приложения, надо выполнить следующие действия.

1. Откройте рядом два окна терминала. Это поможет изменить заголовки окон на client (клиент) и server (сервер). Пользователи терминала macOS могут использовать элемент change title (изменение заголовка) в меню shell (оболочка). Пользователи Windows — команду title.
2. В окне сервера запустите серверное приложение:
   
   

   python src/socket_server.py

3. В окне клиента запустите клиентское приложение:
   
   

   python src/socket_client.py

4. В окне клиента введите свои ответы. Например:
   
   

   How many rolls: 2 ow many rolls: 2
   Dice pattern nd6[dk+-]a: d6

5. Клиент отправит команду, прочитает ответ, выведет его на консоль и выйдет. Запускайте клиент сколько угодно раз, чтобы получить последовательность бросков костей.

Результат будет выглядеть так, как показано на рис. 11.2.


На схеме слева изображен сервер. Приложение запущено, и оно начало прослушивать порт 2401 для клиентов. Справа изображен клиент. Каждый раз, когда запускается клиент, он подключается к общедоступному сокету. Операция подключения создает дочерний сокет, который можно использовать для остальной части взаимодействия. Клиент отправляет команду, сервер отвечает на эту команду, и клиент выводит ее.

Теперь, проанализировав код сервера, мы явно различаем два его раздела. Функция dice_response() считывает данные и отправляет их обратно клиенту через объект socket. Оставшийся сценарий отвечает за создание объекта socket. Создадим пару декораторов, настраивающих поведение сокета, не расширяя и не изменяя самого сокета.

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

class LogSocket:
    def __init__(self, socket: socket.socket) -> None:
        self.socket = socket
    
    def recv(self, count: int = 0) -> bytes:
        data = self.socket.recv(count)
        print(
            f"Receiving {data!r} from {self.socket.getpeername()[0]}"
        )
        return data

    def send(self, data: bytes) -> None:
        print(f"Sending {data!r} to {self.socket.getpeername()[0]}")
        self.socket.send(data)

    def close(self) -> None:
        self.socket.close()

В коде класс декорирует объект socket и предоставляет обращающимся к нему клиентам интерфейсы send(), recv() и close(). Подходящий декоратор мог бы правильно реализовать все аргументы для отправки (которые на самом деле принимают необязательный аргумент), но лучше не будем усложнять. Всякий раз, когда для экземпляра класса LogSocket вызывается функция send(), перед отправкой данных клиенту, использующему исходный сокет, на экран записывается вывод. Аналогично для recv() считываются и регистрируются полученные данные.

Для применения этого декоратора необходимо изменить только одну строку в исходном коде. Вместо того чтобы вызывать функцию dice_response() с исходным клиентским сокетом, вызовем ее с декорированным сокетом:

def main_2() -> None:
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.bind(("localhost", 2401))
    server.listen(1)
    with contextlib.closing(server):
        while True:
            client, addr = server.accept()
            logging_socket = cast(socket.socket, LogSocket(client))
            dice_response(logging_socket)
            client.close()

Итак, мы декорировали основной сокет LogSocket. Сокет LogSocket будет выводить результат на консоль и вызывать метод, который он декорирует. Основная обработка в функции dice_response() не меняется, так как экземпляр LogSocket ведет себя как базовый объект socket.

Обратите внимание, что необходимо было использовать метод cast(), чтобы сообщить mypy, что экземпляр LogSocket предоставит интерфейс, аналогичный обычному объекту socket. Здесь надо проанализировать, почему нельзя просто расширить класс socket и переопределить метод send. Для фактической отправки после регистрации данных подкласс может вызвать методы super().send() и super().recv(). Декорирование имеет преимущество перед наследованием: декорирование можно повторно использовать в различных классах и в различных иерархиях классов. В данном конкретном примере объектов, подобных сокетам, не слишком много, поэтому возможности повторного использования ограниченны.

Если переключить внимание на что-то более общее, чем socket, можно, например, создать потенциально повторно используемые декораторы. Обработка строк или байтов кажется более распространенной, чем обработка socket. Изменение структуры дает желаемую гибкость в дополнение к потенциалу повторного использования. Первоначально был создан процесс из функции dice_response(), которая обрабатывала чтение и запись сокета, и отдельно из функции dice_roller(), которая работает с байтами. Поскольку функция dice_roller() использует байты запроса и создает байты ответа, ее расширение и добавление свойств оказывается немного проще.

Декораторы могут быть связанными и декорировать уже декорированные элементы. Идея состоит в том, чтобы действовать через композицию. Переработаем декоратор логирования, чтобы сосредоточиться на запросе и ответе в виде байтов, а не на объекте socket. Следующий пример должен выглядеть аналогично предыдущему, но с некоторым изменением кода, касающимся размещения в одном методе __call__():

Address = Tuple[str, int]

class LogRoller:
    def __init__(
            self,
            dice: Callable[[bytes], bytes],
            remote_addr: Address
    ) -> None:
        self.dice_roller = dice
        self.remote_addr = remote_addr

    def __call__(self, request: bytes) -> bytes:
        print(f"Receiving {request!r} from {self.remote_addr}")
        dice_roller = self.dice_roller
        response = dice_roller(request)
        print(f"Sending {response!r} to {self.remote_addr}")
        return response

Ниже представлен пример второго декоратора, который сжимает данные, применяя сжатие gzip для полученных байтов:

import gzip
import io

class ZipRoller:
    def __init__(self, dice: Callable[[bytes], bytes]) -> None:
        self.dice_roller = dice
    
    def __call__(self, request: bytes) -> bytes:
        dice_roller = self.dice_roller
        response = dice_roller(request)
        buffer = io.BytesIO()
        with gzip.GzipFile(fileobj=buffer, mode="w") as zipfile:
            zipfile.write(response)
        return buffer.getvalue()

Этот декоратор сжимает входящие данные перед их отправкой клиенту. Он декорирует базовый объект dice_roller, вычисляющий ответ на запрос.

Теперь, когда имеется два декоратора, напишем код, который накладывает один декоратор на другой:

def dice_response(client: socket.socket) -> None:
    request = client.recv(1024)
    try:
        remote_addr = client.getpeername()
        roller_1 = ZipRoller(dice.dice_roller)
        roller_2 = LogRoller(roller_1, remote_addr=remote_addr)
        response = roller_2(request)
    except (ValueError, KeyError) as ex:
        response = repr(ex).encode("utf-8")
    client.send(response)

Цель состоит в том, чтобы разделить три аспекта этого приложения.

• Архивирование полученного документа.
• Ведение журнала или запись в лог-файл (логирование).
• Выполнение базовых вычислений.

Можно архивировать или логировать любое аналогичное приложение, которое работает с приемом и отправкой байтов. При необходимости, используя динамический выбор, можно выполнить операцию архивирования. У нас может быть отдельный файл конфигурации для включения или отключения функции GZip. Например:

if config.zip_feature:
    roller_1 = ZipRoller(dice.dice_roller)
else:
    roller_1 = dice.dice_roller

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

Декораторы в Python

Паттерн Декоратор полезен в Python, но существуют и другие, дополнительные, варианты реализации аналогичных приемов и трюков. Например, Monkey patching (изменение определения класса во время выполнения). Или, скажем, socket.socket.send = log_send — он изменит работу встроенного сокета. Одним из вариантов может быть одиночное наследование, когда необязательные вычисления выполняются в одном большом методе с набором операторов if. И множественное наследование также не следует сбрасывать со счетов только потому, что оно не подходит для конкретного рассмотренного ранее примера.

В Python паттерн Декоратор очень часто используется для функций. Как вы видели в предыдущей главе, функции также являются объектами. На самом деле декорирование функций настолько распространено, что Python предоставляет специальный синтаксис, облегчающий применение таких декораторов к функциям.

Например, проанализируем запись в лог-файл в более общем виде. Вместо того чтобы регистрировать только вызовы отправки на сокеты, иногда полезно регистрировать все вызовы определенных функций или методов. Подобно тому как это происходит в коде ниже:

from functools import wraps

def log_args(function: Callable[..., Any]) -> Callable[..., Any]:
    @wraps(function)
    def wrapped_function(*args: Any, **kwargs: Any) -> Any:
        print(f"Calling {function.__name__}(*{args}, **{kwargs})")
        result = function(*args, **kwargs)
        return result

    return wrapped_function

Эта функция-декоратор очень похожа на предыдущий пример. Но там декоратор получал объект, похожий на сокет, и создавал такой же объект, а на этот раз декоратор получает объект функции и возвращает новый объект функции. Мы предоставили подсказку типа Callable[..., Any], чтобы указать, что в данном случае будет работать любая функция. Приведенный код состоит из трех отдельных задач.

• Функция log_args(), которая принимает другую функцию function в качестве значения параметра.
• Эта функция определяет (внутри) новую функцию wrap_function, которая выполняет дополнительную работу перед вызовом исходной функции и возвратом результатов исходной функции.
• Новая внутренняя функция wrap_function() возвращается функцией-декоратором.

Поскольку используется @wraps(function), новая функция будет иметь копию имени исходной функции и строки документации исходной функции. Иначе все функции, которые мы декорируем, заканчивались бы именем wrapped_function.

Рассмотрим пример функции:

def test1(a: int, b: int, c: int) -> float:
    return sum(range(a, b + 1)) / c
test1 = log_args(test1)

Эту функцию можно декорировать и использовать следующим образом:

>>> test1(1, 9, 2)
Calling test1(*(1, 9, 2), **{})
22.5

Такой синтаксис позволяет динамически создавать декорированные объекты-функции, как это происходило в примере с сокетом. Если не присваивать новому объекту старого имени, можно даже для разных ситуаций сохранить декорированную и недекорированную версии. Например, такой оператор, как test1_log = log_args(test1), создаст декорированную версию функции test1() с именем test1_log().

Как правило, эти декораторы являются общими модификациями, они постоянно применяются к различным функциям. Здесь Python поддерживает специальный синтаксис для применения декоратора во время определения функции. В нескольких местах нашего кода мы уже встречали такой синтаксис. Пришло время разобраться, как это работает.

Вместо применения функции-декоратора после определения метода, чтобы сделать все сразу, будем использовать синтаксис decorator:

@log_args
def test1(a: int, b: int, c: int) -> float:
    return sum(range(a, b + 1)) / c

Основное преимущество такого синтаксиса заключается в том, что можно проанализировать, что функция была декорирована всякий раз, когда мы читаем определение функции. Если декоратор применяется позже, то специалист, работающий с кодом, может не заметить, что функция вообще была изменена.

Теперь может стать намного сложнее ответить на вопрос «Почему моя функция ведения журнала программы вызывает консоль?». Однако такой синтаксис можно применять только к определяемым нами функциям, поскольку у нас нет доступа к исходному коду других модулей. При необходимости декорировать функции, которые являются частью сторонней библиотеки, надо использовать более ранний синтаксис.

Декораторы Python также допускают параметры. functools.lru_cache — один из самых полезных декораторов в стандартной библиотеке. Идея кэширования состоит в том, чтобы сохранять вычисленные результаты функции, тем самым избегая их повторного вычисления. Вместо того чтобы сохранять все параметры и результаты, мы можем уменьшить размер кэша, отбрасывая наименее использовавшиеся (LRU) значения. Например, ниже представлена функция, требующая потенциально дорогостоящих вычислений:

>>> from math import factorial
>>> def binom(n: int, k: int) -> int:
...     return factorial(n) // (factorial(k) * factorial(n-k))

>>> f"6-card deals: {binom(52, 6):,d}"
'6-card deals: 20,358,520'

Применим декоратор lru_cache, чтобы при известном ответе избежать выполнения этих вычислений. Например, так:

>>> from math import factorial
>>> from functools import lru_cache

>>> @lru_cache(64)
... def binom(n: int, k: int) -> int:
...     return factorial(n) // (factorial(k) * factorial(n-k))

Параметризованный декоратор @lru_cache(64), применяемый для создания второй версии функции binom(), сохраняет самые последние 64 результата, чтобы избежать повторного вычисления значений, если они уже были вычислены один раз. Теперь в другом месте приложения никакие изменения больше не требуются. Иногда улучшение от этого небольшого нюанса может быть значительным. Конечно, можно точно настроить размер кэша на основе данных и количества выполняемых вычислений.

Использование подобных параметризованных декораторов похоже на топтание на одном месте. Сначала мы настраиваем декоратор с параметром, затем применяем этот настроенный декоратор к определению функции. Эти два отдельных шага аналогичны действиям, когда вызываемые объекты инициализируются с помощью метода __init__() и могут быть вызваны в качестве функции с помощью метода __call__().

Рассмотрим пример настраиваемого декоратора логирования NamedLogger:

class NamedLogger:
    def __init__(self, logger_name: str) -> None:
        self.logger = logging.getLogger(logger_name)

    def __call__(
           self,
           function: Callable[..., Any]
    ) -> Callable[..., Any]:
        @wraps(function)
        def wrapped_function(*args: Any, **kwargs: Any) -> Any:
            start = time.perf_counter()
            try:
                result = function(*args, **kwargs)
                μs = (time.perf_counter() - start) * 1_000_000
                self.logger.info(
                    f"{function.__name__}, { μs:.1f}μs")
                return result
             except Exception as ex:
                 μs = (time.perf_counter() - start) * 1_000_000
                 self.logger.error(
                     f"{ex}, {function.__name__}, { μs:.1f}μs")
                 raise

         return wrapped_function

Метод __init__() гарантирует, что для создания декоратора мы можем написать код вроде NamedLogger(«log4»). Этот декоратор позаботится о том, чтобы следующая функция использовала определенный регистратор.

Метод __call__() следует уже известному паттерну. Определяем новую функцию wrap_function(), которая делает всю работу, и возвращаем новую функцию:

>>> @NamedLogger("log4")
... def test4(median: float, sample: float) -> float:
...     return abs(sample-median)

Здесь мы создали экземпляр класса NamedLogger. Затем применили этот экземпляр к определению функции test4(). При этом вызывался метод __call__(), который создал новую функцию, декорированную версию функции test4().

Существует еще несколько вариантов использования синтаксиса декоратора. Например, когда декоратор является методом класса, он также сохраняет информацию о декорированной функции, создавая реестр декорированных функций. Кроме того, классы также могут быть декорированы. В этом случае декоратор возвращает новый класс вместо новой функции.

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


Более подробно с книгой можно ознакомиться на сайте издательства:

» Оглавление
» Отрывок

По факту оплаты бумажной версии книги на e-mail высылается электронная книга.
Для Хаброжителей скидка 25% по купону — Python