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:
Код также можно найти в репозитории.
