MinIo, как система объектного хранилища данных, заслуженно пользуется любовью разработчиков: инструмент приятный и, довольно, простой в использовании и освоении. Вот и для одного из наших крупных проектов на работе недавно возникла потребность в использовании S3 хранилища, мы, однако, по корпоративным соображениям выбрали для применения в продакшене другой инструмент, а именно - IONOS (компания у нас немецкая и на ионосе много еще чего завязано), но для тестов и для локального запуска скриптов ничего лучше MinIo в голову нам не пришло. Подобное сочетание при этом вызвало необходимость в использовании такой Python библиотеки, которая могла бы работать и "на наших, и на ваших", а в нашем случае и на MinIo, и на IONOS (поменял параметры в конфиге и тот же самый код, что работал локально, начинает работать и с продакшеном) и этой библиотекой стал Boto3 (стандартный пакет minio для этих целей не подходил). Именно об этой констелляции - Python, MinIo и Boto3 - дальше мне и хотелось бы рассказать, ну а если вместо MinIo вы захотите использовать что-то другое, то "поменял параметры в конфиге и тот же самый код, что работал локально, начинает работать и с продакшеном".

В начале был docker compose файл...

Итак, поскольку MinIo нам главным образом нужен для локальной разработки и тестирования, то с локального запуска и начнем. Для этого создадим в нашем проекте docker-compose.yml файл и поместим там следующий сервис:

services:
  minio:
    image: minio/minio
    entrypoint: sh
    command: > 
      -c 'mkdir -p /data/test-bucket #  Этой командой мы сразу создаем нужный
      && minio server /data'         #  нам бакет в MinIo (test-bucket)
    ports: 
      - 9000:9000
      - 9001:9001
    environment:                         #  Эта часть кода нам нужна, чтобы 
      MINIO_ROOT_USER: 'USERNAME'        #  запустить пользовательскую консоль
      MINIO_ROOT_PASSWORD: 'PASSWORD'
      MINIO_ADDRESS: ':9000'
      MINIO_CONSOLE_ADDRESS: ':9001'

Далее запускаем docker compose командой docker compose up и ждем пока подтянется имидж и запустится контейнер. После завершения этих процоессов переходим по адресу http://localhost:9001/login вводим логин и пароль (USERNAME и PASSWORD) и проверяем, что наш тестовый бакет создался. Должна получиться такая картинка:

S3 сервис на Boto3

Итак MinIo запущен, теперь можно перейти к созданию скрипта на библиотеке boto3, который даст нам возможность взаимодействовать с нашим бакетом. Установим интересующую нас библиотеку командой pip install boto3 (или какой-нибудь другой командой, которая используется вашим любимым менеджером зависимостей), а затем создадим файл s3_service.py и поместим в него следующий код:

from io import BytesIO
from pathlib import Path
from typing import Optional, Union

import boto3
from botocore.client import Config
from botocore.exceptions import ClientError
from botocore.response import StreamingBody


class S3BucketService:
    def __init__(
        self,
        bucket_name: str,
        endpoint: str,
        access_key: str,
        secret_key: str,
    ) -> None:
        self.bucket_name = bucket_name
        self.endpoint = endpoint
        self.access_key = access_key
        self.secret_key = secret_key

    def create_s3_client(self) -> boto3.client:
        client = boto3.client(
            "s3",
            endpoint_url=self.endpoint,
            aws_access_key_id=self.access_key,
            aws_secret_access_key=self.secret_key,
            config=Config(signature_version="s3v4"),
        )
        return client

    def upload_file_object(
        self,
        prefix: str,
        source_file_name: str,
        content: Union[str, bytes],
    ) -> None:
        client = self.create_s3_client()
        destination_path = str(Path(prefix, source_file_name))

        if isinstance(content, bytes):
            buffer = BytesIO(content)
        else:
            buffer = BytesIO(content.encode("utf-8"))
        client.upload_fileobj(buffer, self.bucket_name, destination_path)

    def list_objects(self, prefix: str) -> list[str]:
        client = self.create_s3_client()

        response = client.list_objects_v2(Bucket=self.bucket_name, Prefix=prefix)
        storage_content: list[str] = []

        try:
            contents = response["Contents"]
        except KeyError:
            return storage_content

        for item in contents:
            storage_content.append(item["Key"])

        return storage_content

    def delete_file_object(self, prefix: str, source_file_name: str) -> None:
        client = self.create_s3_client()
        path_to_file = str(Path(prefix, source_file_name))
        client.delete_object(Bucket=self.bucket_name, Key=path_to_file)
        

В приведенном коде мы создали класс S3BucketService, который позволяет нам настроить соединение с нашим хранилищем, дает возможность добавления и удаления объектов, а также получения их списка. Самое время настроить параметры соединения с MinIo и проверить, правильно ли все работает. Создадим файл конфигурации (вы можете использовать любой привычный вам формат, лично я буду применять .ini файл и конфигпарсер) и назовем его default.ini. Запишем внутрь такой конфиг:

[s3_storage]
bucket_name = test-bucket
endpoint = http://localhost:9000
access_key = USERNAME
secret_key = PASSWORD

Затем добавим в уже созданный файл s3_service.py следующую функцию:

def s3_bucket_service_factory(config: configparser.ConfigParser) -> S3BucketService:
    return S3BucketService(
        config["s3_storage"]["bucket_name"],
        config["s3_storage"]["endpoint"],
        config["s3_storage"]["access_key"],
        config["s3_storage"]["secret_key"],
    )

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

import configparser

from s3_service import s3_bucket_service_factory

config = configparser.ConfigParser()
config.read('default.ini')

s3 = s3_bucket_service_factory(config)
s3.upload_file_object("test", "test.txt", "test")

Запустим получившийся скрипт, подождем завершения его выполнения и перейдем в пользовательскую консоль MinIo, нажмем на бакет и порадуемся появившейся директории test, с файлом test.txt и его содержанием "test" (ну тут, конечно, прежде чем радоваться, придется сначала файл скачать, но это также можно сделать через GUI, предоставляемое MinIo)!

Тоже самое для любопытства можно проделать и с другими, определенными в классе S3BucketService методами, такими как получение списка объектов (тут проще всего будет просто принтовать результат в консоль) и удаление объекта.

Тесты (ну или что-то типа того)

В описываемом примере нет никакой бизнес логики, которая бы требовала загрузки-выгрузки объектов, их удаления и пр. Отсюда написание тестов в целом представляется избыточным (ну не будем же мы в самом деле на голубом глазу и с серьезным лицом тестить MinIo), но в показательных целях почему бы и нет. Напишем небольшой тест, который проверяет, что наши методы загрузки объектов и их удаления работают. Переименуем уже имеющийся файл test.py в test_minio.py и напишем в нем следующий код:

import configparser

from s3_service import s3_bucket_service_factory


OBJECTS_TO_UPLOAD = [1, 2, 3]
config = configparser.ConfigParser()
config.read("default.ini")
S3 = s3_bucket_service_factory(config)


def test_object_is_created():
    for obj in OBJECTS_TO_UPLOAD:
        S3.upload_file_object("test", f"{obj}.txt", "")

    objects_in_bucket = S3.list_objects("test")
    assert len(objects_in_bucket) == len(OBJECTS_TO_UPLOAD)


def test_object_is_deleted():
    for obj in OBJECTS_TO_UPLOAD:
        S3.delete_file_object("test", f"{obj}.txt")

    objects_in_bucket = S3.list_objects("test")
    assert len(objects_in_bucket) == 0

Запустим pytest и вуаля, все работает.

Gitlab-ci для pytest и Minio

В завершении картины приведу также скрипт для gitlab-ci, который запускает pipeline с pytest, может кому-то пригодится.

test-pytest:
  image: 'python:3.9-slim-bullseye'
  stage: test
  needs: []
  variables:
    MINIO_BASE_URL: http://minio:9000
  services:
    - name: minio/minio
      alias: minio
      entrypoint: ['sh']
      command:
        - -c
        - >
          mkdir -p /data/test-bucket
          && minio server /data
      variables:
        MINIO_ROOT_USER: 'USERNAME'
        MINIO_ROOT_PASSWORD: 'PASSWORD'
  before_script:    # тут нам нужно убедиться, что MinIo запущен 
    - apt update    # и лишь потом стартовать тесты
    - apt install -y curl
    - | 
      until curl --output /dev/null --silent --head --fail $MINIO_BASE_URL/minio/health/live; do
        printf '.'
        sleep 1
      done
    - pip install -r requirements.txt
  script:
    - |
      pytest -v .

PS:
Код также можно найти в репозитории.