← назад к разделу

Файлы в Python-сервисе живут не в базе и не на диске контейнера, а в объектном хранилище с протоколом S3: в AWS это сам S3, в Yandex Cloud это Object Storage, в своём контуре это MinIO или Ceph. Протокол один, поэтому и клиент один: boto3 поверх botocore. Отличия между хранилищами укладываются в несколько строк настройки, а всё остальное, от потоковой отдачи файла до presigned-ссылок, работает одинаково.

boto3 синхронный, и это первое, с чем сталкивается сервис на FastAPI: вызов put_object блокирует поток. Поэтому его зовут через asyncio.to_thread, а не прямо из корутины. Асинхронные обёртки есть (aioboto3 поверх aiobotocore), но они переписывают внутренности botocore и привязаны к его точной версии; для сервиса, который грузит и отдаёт файлы, пула потоков хватает.

Обязательно

Клиент: один на процесс, несколько строк под хранилище

Клиент boto3 безопасен для потоков и держит пул HTTP-соединений, поэтому его создают один раз при старте и передают в хранилище файлов. Ресурсы (boto3.resource) потокобезопасными не являются, и в сервисе их не используют.

import boto3
from botocore.config import Config
from pydantic import BaseModel


class StorageSettings(BaseModel):
    endpoint: str | None = None
    region: str = "us-east-1"
    bucket: str
    access_key: str | None = None
    secret_key: str | None = None
    path_style: bool = False


def new_client(s: StorageSettings):
    config = Config(
        signature_version="s3v4",
        s3={"addressing_style": "path" if s.path_style else "virtual"},
        retries={"max_attempts": 3, "mode": "standard"},
        connect_timeout=5,
        read_timeout=60,
        request_checksum_calculation="when_required",
        response_checksum_validation="when_required",
    )
    return boto3.client(
        "s3",
        endpoint_url=s.endpoint,
        region_name=s.region,
        aws_access_key_id=s.access_key,
        aws_secret_access_key=s.secret_key,
        config=config,
    )

Что здесь меняется от хранилища к хранилищу.

  • AWS S3. endpoint пустой, path_style ложь, регион настоящий. Ключей в коде и в переменных окружения в проде нет: boto3.client без aws_access_key_id сам пройдёт цепочку (переменные окружения, файл профиля, роль пода через IRSA, роль инстанса), и сервис получает права от роли.
  • Yandex Object Storage. endpoint: https://storage.yandexcloud.net, region: ru-central1, статический ключ сервисного аккаунта из секрета Kubernetes. Виртуальные хосты (bucket.storage.yandexcloud.net) поддерживаются, path-style тоже.
  • MinIO в своём контуре и в тестах. endpoint: http://minio:9000, path_style: True обязательно: без него клиент пойдёт на bucket.minio, которого нет в DNS. Регион любой, MinIO его не проверяет.

Подпись. С чужим endpoint_url botocore подписывает запросы и presigned-ссылки старым алгоритмом второй версии: в ссылке появляется AWSAccessKeyId= вместо X-Amz-Signature. Часть хранилищ такую подпись отвергает, а главное, она не покрывает заголовки вроде Content-Type. signature_version="s3v4" ставят всегда, с настоящим S3 он и так по умолчанию.

Контрольные суммы. С начала 2025 года botocore по умолчанию считает CRC32 для каждого put_object и присылает его заголовком. S3 это понимает, часть совместимых хранилищ и старые версии MinIO отвечают ошибкой подписи или NotImplemented. Настройки when_required возвращают прежнее поведение: сумма считается только там, где её требует сама операция. С настоящим S3 они не мешают.

Таймауты и повторы. По умолчанию и connect_timeout, и read_timeout равны 60 секундам, а режим повторов устаревший с пятью попытками. read_timeout ограничивает паузу между байтами, а не вызов целиком, поэтому гигабайтный файл через него проходит; на вызов целиком лимита у SDK нет, его даёт asyncio.wait_for вокруг to_thread. Режим standard с тремя попытками повторяет сетевые ошибки и 5xx с нарастающей паузой; 404 и 403 не повторяются.

Загрузка файла

import asyncio
from typing import BinaryIO


class FileStore:
    def __init__(self, client, bucket: str) -> None:
        self.client = client
        self.bucket = bucket

    async def put(self, key: str, body: BinaryIO, size: int, content_type: str) -> None:
        await asyncio.to_thread(
            self.client.put_object,
            Bucket=self.bucket, Key=key, Body=body, ContentLength=size, ContentType=content_type,
        )

Три договорённости, которые стоит зафиксировать в хранилище файлов с первого дня.

  • Ключ строит сервис, а не пользователь. orders/2026/10/0192f1e4-....pdf: префикс по сущности и дате, идентификатор вместо имени файла. Имя, которое прислал пользователь, живёт в базе и возвращается в Content-Disposition при отдаче. Так в ключ не попадут пробелы, кириллица и ../.
  • ContentLength известен заранее. UploadFile из FastAPI отдаёт его как upload.size, а само тело — файловым объектом upload.file, который SDK читает потоком. Читать файл целиком через await upload.read(), чтобы завернуть в BytesIO, нельзя: каждый файл на 200 МБ это 200 МБ кучи на запрос.
  • ContentType задаёт сервис по проверенному содержимому (первые байты через библиотеку вроде filetype), а не по расширению от клиента и не по upload.content_type: именно его потом отдаст браузер.

Большие файлы: multipart через TransferConfig

Один put_object ограничен 5 ГБ и идёт одним соединением. Для видео, бэкапов и выгрузок есть многочастная загрузка: файл режется на части от 5 МБ, части уходят параллельно, хранилище склеивает их по команде. Руками это три вызова и обработка сбоев, в SDK это upload_fileobj с настройками переноса:

from boto3.s3.transfer import TransferConfig

MIB = 1 << 20


class FileStore:
    transfer = TransferConfig(multipart_threshold=16 * MIB, multipart_chunksize=16 * MIB, max_concurrency=4)

    async def put_large(self, key: str, body: BinaryIO, content_type: str) -> None:
        await asyncio.to_thread(
            self.client.upload_fileobj,
            body, self.bucket, key,
            ExtraArgs={"ContentType": content_type},
            Config=self.transfer,
        )

Загрузчик сам решает, слать файл одним запросом или частями (порог по умолчанию 8 МБ), и принимает поток без длины. multipart_chunksize на max_concurrency это память, которую он займёт на одну загрузку (здесь 64 МБ), и это надо помнить, когда таких загрузок десятки одновременно; части грузят потоки из своего пула, по умолчанию десять. Прерванная многочастная загрузка оставляет в бакете части, которые не видны в списке объектов, но занимают место и стоят денег: правило жизненного цикла AbortIncompleteMultipartUpload через сутки обязательно для любого бакета, куда грузят большие файлы.

Отдача файла

Скачивание это тоже поток: тело ответа S3 переливается в тело HTTP-ответа кусками, без буфера на весь файл.

from botocore.exceptions import ClientError
from fastapi import HTTPException
from fastapi.responses import StreamingResponse


class FileStore:
    async def get(self, key: str):
        try:
            return await asyncio.to_thread(self.client.get_object, Bucket=self.bucket, Key=key)
        except self.client.exceptions.NoSuchKey as e:
            raise FileNotFound(key) from e


@router.get("/files/{key:path}")
async def serve_file(key: str, store: FileStoreDep) -> StreamingResponse:
    try:
        obj = await store.get(key)
    except FileNotFound:
        raise HTTPException(status_code=404)
    except ClientError:
        raise HTTPException(status_code=502, detail="storage unavailable")
    return StreamingResponse(
        obj["Body"].iter_chunks(64 * 1024),
        media_type=obj["ContentType"],
        headers={"Content-Length": str(obj["ContentLength"])},
    )

Ошибки SDK частично типизированы: client.exceptions.NoSuchKey у get_object, NoSuchBucket, NoSuchUpload. У head_object тела в ответе нет, поэтому отсутствующий ключ приходит общим ClientError с e.response["Error"]["Code"] == "404". Их распознают на границе хранилища и переводят в ошибки своего домена, чтобы обработчик не знал о botocore. Частичное чтение (докачка, видео с середины) это аргумент Range="bytes=0-1048575" у get_object.

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

Presigned-ссылки: хранилище работает вместо сервиса

Подписанная ссылка это обычный URL с подписью и сроком жизни. Тот, у кого она есть, может выполнить ровно одну операцию над ровно одним объектом, пока срок не вышел. Сервис подписывает ссылку своими ключами и ничего не передаёт через себя; сетевого вызова при этом нет, подпись считается локально.

клиент: POST /uploads проверка прав, запись о файле сервис: подписанная ссылка с content-length-range байты идут мимо сервиса клиент: PUT файла прямо в S3 подтверждение клиента или событие хранилища сервис: помечает файл готовым

Сервис выдаёт подписанную ссылку и ведёт запись о файле, а сами байты идут в хранилище напрямую; приложение не держит ни памяти, ни соединения под загрузку.

class Presigner:
    def __init__(self, client, bucket: str) -> None:
        self.client = client
        self.bucket = bucket

    def download_url(self, key: str, file_name: str) -> str:
        return self.client.generate_presigned_url(
            "get_object",
            Params={
                "Bucket": self.bucket,
                "Key": key,
                "ResponseContentDisposition": f'attachment; filename="{file_name}"',
            },
            ExpiresIn=15 * 60,
        )

    def upload_form(self, key: str, content_type: str, max_size: int) -> dict:
        return self.client.generate_presigned_post(
            self.bucket,
            key,
            Fields={"Content-Type": content_type},
            Conditions=[{"Content-Type": content_type}, ["content-length-range", 0, max_size]],
            ExpiresIn=10 * 60,
        )

Для скачивания хватает generate_presigned_url. Для загрузки есть два пути. Ссылка на put_object подписывает Content-Type (он попадает в подписанные заголовки), но не размер: пользователь загрузит гигабайт вместо обещанных ста килобайт. Поэтому для загрузки из браузера берут presigned POST: он возвращает адрес и набор полей формы, а условие content-length-range ограничивает размер на стороне хранилища. Браузер отправляет multipart/form-data с этими полями и файлом последним; нарушил тип или размер — хранилище ответит 403. Это не неудобство, а защита: пользователь не загрузит исполняемый файл под видом картинки. Срок жизни ссылки короткий, минуты, потому что отозвать её нельзя.

Загрузка из браузера напрямую в бакет требует CORS на бакете: разрешённый источник (домен приложения), методы POST, PUT и GET, заголовок Content-Type. Без этого запрос из браузера упадёт ещё на preflight, а в логах сервиса не будет ничего. CORS это настройка бакета (put_bucket_cors), её делают один раз при создании инфраструктуры, а не из кода сервиса.

База и хранилище: кто кому верит

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

  • Загрузка: сначала объект, потом запись. Сервис выдаёт presigned-форму на ключ, клиент загружает файл, затем вызывает «подтвердить загрузку». Обработчик проверяет head_object (объект есть, размер и тип совпадают) и только тогда пишет строку в базу. Упал клиент посередине: в бакете лежит сирота, в базе нет ничего, ссылки на несуществующий файл не появилось.
  • Удаление: сначала запись, потом объект, и не в запросе. Обработчик помечает файл удалённым в базе (в той же транзакции, что и бизнес-операцию), а объект удаляет фоновая задача по этой отметке с повторами. Удалять из обработчика нельзя: откат транзакции после удаления объекта оставит в базе ссылку в никуда.
  • Сироты чистит сверка. Раз в сутки задача проходит по префиксу постранично и удаляет объекты старше часа, на которые нет строки в базе. delete_objects принимает до тысячи ключей за вызов и возвращает список тех, что удалить не удалось.
def list_objects(client, bucket: str, prefix: str):
    paginator = client.get_paginator("list_objects_v2")
    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        for obj in page.get("Contents", []):
            yield obj["Key"], obj["Size"], obj["LastModified"]


def delete_many(client, bucket: str, keys: list[str]) -> list[str]:
    failed = []
    for i in range(0, len(keys), 1000):
        chunk = keys[i:i + 1000]
        resp = client.delete_objects(Bucket=bucket, Delete={"Objects": [{"Key": k} for k in chunk], "Quiet": True})
        failed.extend(e["Key"] for e in resp.get("Errors", []))
    return failed

Пагинатор сам ходит за следующими страницами по ContinuationToken; у страницы без объектов ключа Contents нет вовсе, отсюда page.get.

Когда хранилище недоступно

Чтение файла это 502 с коротким дедлайном, а не ожидание минуту: asyncio.wait_for(to_thread(...), timeout=5) вокруг вызова, и пользователь видит «файл временно недоступен», остальное приложение работает. Повторы SDK (три попытки с нарастающей паузой) закрывают короткие сбои, дольше их растягивать не нужно. Загрузка через presigned-форму недоступности сервиса вообще не касается, а недоступность хранилища клиент увидит сам и повторит. Фоновые удаления и сверка просто повторятся в следующий запуск. В проверке здоровья сервиса хранилище значится как зависимость, но не как причина снять под с балансировщика.

Отдельная забота пула потоков: каждый вызов to_thread занимает поток из общего пула цикла событий, а зависший read_timeout в 60 секунд держит его минуту. Сотня одновременных скачиваний при недоступном хранилище исчерпает пул, и встанут все остальные to_thread, включая обращения к базе через синхронные драйверы. Для хранилища заводят свой ThreadPoolExecutor на 8–16 потоков и передают его в loop.run_in_executor, тогда очередь за файлами не трогает остальных.

Тесты на MinIO

Хранилище файлов это интерфейс (put, get, delete, download_url), и обработчики тестируются с заглушкой в памяти. Сама реализация на SDK тестируется против MinIO в контейнере: он совместим с S3 настолько, что код не отличает их.

import pytest
from testcontainers.minio import MinioContainer


@pytest.fixture(scope="session")
def storage_client():
    with MinioContainer("minio/minio:RELEASE.2024-01-16T16-07-38Z") as minio:
        cfg = minio.get_config()
        client = new_client(StorageSettings(
            endpoint=f"http://{cfg['endpoint']}", region="us-east-1", bucket="files",
            access_key=cfg["access_key"], secret_key=cfg["secret_key"], path_style=True,
        ))
        client.create_bucket(Bucket="files")
        yield client

Что проверяют: загрузка и чтение возвращают те же байты и тот же Content-Type, отсутствующий ключ даёт FileNotFound, а не ошибку SDK, presigned-ссылка на скачивание открывается обычным httpx.get без ключей, delete_many чистит префикс. Контейнер один на сессию, бакет или префикс на каждый тест свой. Для модульных тестов кода, который зовёт S3 ради одного вызова, есть moto: декоратор mock_aws подменяет AWS внутри процесса без Docker и без сети, но проверяет он поведение самого AWS, а не MinIO или Yandex.

Дополнительно: при первом чтении можно пропустить

Глубже: многочастная загрузка рукамирасширенное

Когда файл грузит клиент напрямую, а не сервис, upload_fileobj не поможет: его части подписывает сервис. Схема такая: сервис вызывает create_multipart_upload и получает UploadId, выдаёт клиенту presigned-ссылки на upload_part для каждого номера части, клиент грузит части параллельно и возвращает ETag каждой, сервис завершает complete_multipart_upload со списком частей. Минимальный размер части 5 МБ (кроме последней), максимум частей 10 000. Незавершённую загрузку закрывает abort_multipart_upload или правило жизненного цикла бакета.

Глубже: события бакетарасширенное

Хранилище умеет сообщать о появлении объекта: в AWS это уведомления в SQS или EventBridge, в MinIO это события в Kafka, NATS или вебхук, в Yandex Object Storage это очередь сообщений. Так строят обработку загруженного: превью картинок, антивирус, разбор выгрузок. Потребитель на Python читает событие с ключом объекта, забирает объект get_object и пишет результат рядом под другим префиксом. Событие может прийти дважды, поэтому обработчик идемпотентен: перед работой он проверяет head_object на уже существующий результат.

Коротко

  • Один клиент boto3 на процесс, вызовы через asyncio.to_thread со своим пулом потоков; для AWS права от роли без ключей в коде, для совместимых хранилищ endpoint_url, path-style и статический ключ из секрета.
  • signature_version="s3v4" всегда: с чужим адресом botocore иначе подписывает по старому алгоритму; контрольные суммы when_required, иначе часть S3-совместимых хранилищ отвергает put_object.
  • Таймауты по умолчанию 60 секунд на соединение и на паузу между байтами; режим повторов standard с тремя попытками, лимит на вызов целиком даёт asyncio.wait_for.
  • Ключ объекта строит сервис, имя файла пользователя живёт в базе и возвращается в Content-Disposition; загрузка и отдача потоком через upload.file и iter_chunks, без чтения файла в память.
  • Большие файлы через upload_fileobj с TransferConfig; правило AbortIncompleteMultipartUpload на бакете обязательно.
  • Ошибки: NoSuchKey типизирована, у head_object — ClientError с кодом 404; переводить в ошибки домена на границе хранилища.
  • Presigned-ссылка для скачивания, presigned POST с content-length-range для загрузки из браузера, CORS на бакете.
  • Сначала объект, потом строка в базе; удаление объекта фоновой задачей по отметке; сверка сирот пагинатором list_objects_v2.
  • Недоступность хранилища это 502 на файловых ручках с коротким дедлайном, а не падение сервиса и не исчерпанный пул потоков.
  • Тесты реализации на MinIO в контейнере, обработчики на заглушке интерфейса, moto для кода против самого AWS.

Что почитать дальше