Фасад для python библиотеки

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

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

Но что, если даже в этом случае код со временем становится все более запутанным и сложным для читаемости? Как правило, такое происходит, когда проект разрастается, появляется необходимость использовать разные стратегии управления данными или масштабироваться каким-то иным способом.

На мой взгляд, неплохим выходом из ситуации служит использование объектно-ориентированного подхода, а именно написание некого класса «обвязки» с более простыми методами, инкапсулирующими в себе сложную логику обращения к оригинальной библиотеке.

Фасад (Facade) — структурный паттерн проектирования, реализующий простой интерфейс для работы со сложным модулем, библиотекой, фреймворком.

В этой статье для иллюстрации данного паттерна я бы хотел показать реализацию небольшого синтетического проекта.

Основная задача — создание модуля для работы с файловыми хранилищами. И в качестве примера сторонней библиотеки — boto3 — официальную python библиотеку для работы с AWS API. Нас, в частности, интересует работа с s3.

Для начала определим функциональные требования.

Нам необходима возможность получать данные, хранящиеся в файле, записывать данные в файл или удалять его. Файл может находиться в разных типах хранилищ.

И если конкретнее, то реализовать класс, являющийся моделью файла в хранилище данных и имеющий три метода: read(), write() и delete()

Также опишем требования к архитектуре.

• Отсутствие повторяемости низкоуровневого кода (низкоуровнего по отношению к проекту)

  Преимущество: Изменения реализации необходимо производить только в одном месте

• Одна точка инициализации доступа к хранилищу

  Преимущество: Доступ определяется на уровне конфигурации приложения, а не где-то в коде

• Одна точка для запросов к хранилищу

  Преимущество: Появляется возможность единого декорирования для всех методов. Допустим, для добавления логирования или обработки ошибок.

• Конкретный тип хранилища не привязан жестко к модели файла

  Преимущество: Возможность использовать разные типы хранилищ

Данные требования немного выходят за рамки реализации «фасада», но полагаю, что так будет немного интереснее.

И прежде чем перейти к реализации давайте разберемся, как вообще такая функциональность реализуется «в лоб». В boto3 есть много способов ее имплементировать, но для примера возьмем один. Также будем считать, что ключи доступа к AWS хранятся в переменных окружения как AWS_ACCESS_KEY_ID и AWS_SECRET_ACCESS_KEY.

В первую очередь получим s3 как ресурс:

import boto3 import os  AWS_ACCESS_KEY_ID = os.environ['AWS_ACCESS_KEY_ID'] AWS_SECRET_ACCESS_KEY = os.environ['AWS_SECRET_ACCESS_KEY']  resource = boto3.resource(     's3',     aws_access_key_id=AWS_ACCESS_KEY_ID,     aws_secret_access_key=AWS_SECRET_ACCESS_KEY )

Теперь определим объект хранилища, на котором будет тестировать работу программы. Для этого нам необходим тестовый bucket и имя используемого объекта.

bucket_name = 'test_bucket' path_to_file = 'test_folder/test_object.txt'

Сохранить файл путем передачи контента:

content = b'test_object_data'  bucket = resource.Bucket(bucket_name) file_object = bucket.Object(path_to_file)  file_object.put(Body=content)

Получить контент:

content = file_object.get()['Body'].read()

Удалить файл:

file_object.delete()

Выглядит не сложно, а теперь представим, что нам необходимо постоянно работать с данными, хранящимися в s3 в разных модулях, в разных методах. И если надо будет как-то изменить реализацию, сменить библиотеку или подключить другое хранилище, то все становится совсем уж плохо.

Приступим к реализации архитектуры.

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

class StorageObject:      def __init__(self, path: str, base_path: str, resource: Any = None) -> None:         """         path: путь к файлу или же какой-либо другой идентификатор         base_path: базовый путь         resource: некий исходный объект хранилища, реализуемый сторонней библиотекой,                   необходим для дальнейшего вызова методов         """         raise NotImplementedError      def read(self) -> bytes:         raise NotImplementedError      def write(self, content: bytes) -> None:         raise NotImplementedError      def delete(self) -> None:         raise NotImplementedError

Далее перейдем уже непосредственно к реализации объекта файла с s3.

class S3StorageObject(StorageObject):     _object: BotoS3Object      def __init__(self, path: str, base_path: str, resource: BotoS3Resource) -> None:         """         в данном случае base_path - это название бакета         """         self._object = resource.Bucket(base_path).Object(path)      def read(self) -> bytes:         return self._object.get()['Body'].read()      def write(self, content: bytes) -> None:         self._object.put(Body=content)      def delete(self) -> None:         self._object.delete()

Небольшое замечание насчет типа ресурса: BotoS3Resource . Это результат выполнения функции boto3.resource('s3', *args, **kwargs)

Но так как она явно никакой конкретный тип не возвращает, для лучшего понимания мы определили наследника typing.Protocol. И там же нам нужен нативный тип объекта s3: BotoS3Object . Опять же явно в boto3 такого типа нет, потому что он формируется динамически при помощи фабрики, поэтому пишем свой.

class BotoS3Resource(Protocol):     """Результат вызова boto3.resource('s3', ...)"""     def Bucket(self, bucket_name: str): ...  class BotoS3Object(Protocol):     """Результат boto3.resource('s3', ...).Bucket(...).Object(...)"""     def get(self) -> dict: ...     def put(self, Body: bytes) -> dict: ...     def delete(self) -> dict: ...

Таким образом мы инкапсулировали в S3StorageObject все вызовы к более низкоуровневой библиотеке и закрыли первое архитектурное требование. С таким классом уже можно работать, то есть частично функциональные требования выполнены:

resource = boto3.resource(     's3',     aws_access_key_id=AWS_ACCESS_KEY_ID,     aws_secret_access_key=AWS_SECRET_ACCESS_KEY ) file_object = S3StorageObject(path_to_file, bucket_name, resource)  content = file_object.read() file_object.write(content) file_object.delete()

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

class OSStorageObject(StorageObject):     _path: str      def __init__(self, path: str, base_path: str = '', resource: Any = None) -> None:         self._path = os.path.join(base_path, path)      def read(self) -> bytes:         with open(self._path, 'rb') as file:             return file.read()      def write(self, content: bytes) -> None:         os.makedirs(os.path.dirname(self._path), exist_ok=True)          with open(self._path, 'wb') as file:             file.write(content)      def delete(self) -> None:         os.remove(self._path)

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

• во-первых, proxy будет сам решать, объект какого типа ему создавать, а так же выступать для пользователя одной точкой входа для работы с файлами

• во-вторых, добавлять необходимую дополнительную логику в работу с файлами

Выглядеть он будет примерно так:

class File:     storage: Storage      def __init__(         self,         path: str,         base_path: str | None = None,         storage: Storage | None = None     ) -> None:         # хранилище можно задать при инициализации, либо заранее добавить в класс         if storage: self.storage = storage          self._object = self.storage.build_object(path, base_path)      def read(self) -> bytes:         return self._action('read')      def write(self, content: bytes) -> None:         self._action('write', content)      def delete(self) -> None:         self._action('delete')      def _action(self, action: str, *args, **kwargs) -> Any:         return getattr(self._object, action)(*args, **kwargs)

С его помощью мы закрываем четвёртое архитектурное требование.

Но тут еще надо разобраться с несколькими вопросами. Во-первых, что такое Storage? До этого такого класса у нас не было. И правильно, потому что каждый StorageObject мог принимать какой-то resource для формирования объекта и нам было не особо важно, откуда этот resource берётся. Сейчас же мы предполагаем, что хранилищ может быть множество и они могут меняться. Соответственно, работу по их инициализации и построению StorageObject есть смысл вынести непосредственно в хранилища Storage. Интерфейс у такого класса очень простой. Фактически нам требуется только один метод для создания StorageObject: build_object:

class Storage:     base_path: str     resource: Any     object_type: type[StorageObject]      def __init__(self, base_path: str | None = None, *args, **kwargs) -> None:         self.resource = self._build_resource(*args, **kwargs)          if base_path: self.base_path = base_path      def build_object(self, path: str, base_path: str | None = None) -> StorageObject:         return self.object_type(path, base_path or self.base_path, self.resource)      def _build_resource(*args, **kwargs) -> Any:         return None   class S3Storage(Storage):     object_type = S3StorageObject      def _build_resource(self, *args, **kwargs) -> BotoS3Resource:         return boto3.resource('s3', *args, **kwargs)  # type: ignore

Подход с хранилищем хорош тем, что можно определить его на уровне конфигурации приложения так:

# данный способ следует использовать с осторожностью File.storage = S3Storage(          aws_access_key_id=AWS_ACCESS_KEY_ID,     aws_secret_access_key=AWS_SECRET_ACCESS_KEY )  file = File(path_to_file, bucket_name)

или так:

# при необходимости меняется класс хранилища, но всё продолжает работать storage = S3Storage(     aws_access_key_id=AWS_ACCESS_KEY_ID,     aws_secret_access_key=AWS_SECRET_ACCESS_KEY )  file = File(path_to_file, bucket_name, storage)

Таким образом закрыто второе архитектурное требование.

И последний момент, касающийся File, это метод _action.

Мы используем его, чтобы определить единственную точку обращения непосредственно к методам хранилища. Благодаря этому есть возможность только в одном месте установить логгер, блок обработки ошибок или декорировать любым другим образом. При этом затрагивая сразу всё методы.

Им мы закрываем третье архитектурное требование.

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

storage = S3Storage(     bucket_name,     aws_access_key_id=AWS_ACCESS_KEY_ID,     aws_secret_access_key=AWS_SECRET_ACCESS_KEY )  file = File(path_to_file, storage=storage)  content = file.read() file.write(content) file.delete()

P.S. Конечно, этот пример — лишь иллюстрация. Что-то для конкретной задачи подойдёт, что-то нет, но возможно кому-то он будет интересен в качестве отправной точки для решения его задач.