Для 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. Конечно, этот пример - лишь иллюстрация. Что-то для конкретной задачи подойдёт, что-то нет, но возможно кому-то он будет интересен в качестве отправной точки для решения его задач.

Код целиком: https://github.com/Destriery/file-storages