/
systemsstrategyy
/
Treker
Обзор
Документация
Войти
/
systemsstrategyy
/
Treker
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
master
api/shared/storage/base.py
57 строк
2 KB
SystemsStrategy
Initial import
02 июл 2026, 15:15
02 июл 2026, 15:15
9624ed1
Код
Авторство
О чём код?
"""FileStorage Protocol — публичный интерфейс хранилища файлов. Coding_Principles ADR-0001 §FileStorage: - Абстракция через Protocol (structural typing), не ABC. - LocalFileStorage сейчас, S3/MinIO когда упрёмся в объёмы (отдельный ADR). - nginx X-Accel-Redirect для отдачи: backend ставит заголовок с относительным путём, nginx стримит файл сам. Подписанные ссылки не нужны при local FS — права проверяет FastAPI до того как поставить X-Accel-Redirect header. """ from __future__ import annotations from collections.abc import AsyncIterator from dataclasses import dataclass from typing import BinaryIO, Protocol @dataclass(frozen=True, slots=True) class StoredFile: """Результат загрузки. `key` — стабильный идентификатор для запросов read/delete/url. `size` — байты на диске (после возможной компрессии).""" key: str size: int content_type: str class FileStorage(Protocol): """Контракт хранилища. Реализации: LocalFileStorage (Day 5+), S3FileStorage (когда понадобится).""" async def save( self, *, stream: BinaryIO, key: str, content_type: str, ) -> StoredFile: """Сохранить поток байт под указанным key. Если файл уже существует — перезаписать.""" ... async def read(self, key: str) -> AsyncIterator[bytes]: """Стримит содержимое файла. Бросает FileNotFoundError если ключа нет.""" ... async def delete(self, key: str) -> bool: """Удаляет файл. True если был удалён, False если ключа не было.""" ... async def exists(self, key: str) -> bool: """Проверка существования без чтения.""" ... def internal_path(self, key: str) -> str: """Internal-путь для nginx X-Accel-Redirect (local FS only). Для S3 эта операция будет недоступна — реализация бросит NotImplementedError.""" ...