Срез 0 [5.2]: S3Storage (src/core/clients.py): get_bytes, put_bytes, put_json, from_settings с secure
Ошибки не проглатываются: S3ObjectNotFoundError (NoSuchKey) и S3ReadError для чтения, S3WriteError для записи, исходная причина — в __cause__. JSON пишется в UTF-8 без экранирования, Content-Type application/json. Шаблонные put_object/get_object удалены (не использовались); удалён и шаблонный nats_callback.py — его единственный импорт S3Storage, сам он нигде не используется (слой NATS — следующим коммитом в src/core/nats.py). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
b9e1be37ee
commit
c6dfb3e22c
@ -1,28 +1,105 @@
|
||||
"""Клиенты инфраструктуры. Каркас: оставлен S3Storage; outbox-паблишер (PollWorker) реализуется в Срезе 0
|
||||
по docs/context/03-architecture.md (референс — reference-old-arm/api/src/core/clients.py).
|
||||
"""S3-клиент сервиса (MinIO через `miniopy-async`): чтение исходника, запись сырого ответа ИС и файла результата.
|
||||
|
||||
Ошибки не проглатываются и не превращаются в `None`: каждая операция бросает исключение из иерархии
|
||||
`S3StorageError` с исходной причиной в `__cause__`, чтобы воркер мог классифицировать сбой (design D-7):
|
||||
|
||||
- `S3ObjectNotFoundError` (подкласс `S3ReadError`) — объекта нет (`NoSuchKey`);
|
||||
- `S3ReadError` — любая другая ошибка чтения (нет доступа, нет бакета, сеть) → воркер отвечает `FILE_READ_ERROR`;
|
||||
- `S3WriteError` — ошибка записи (`put_bytes` / `put_json`).
|
||||
|
||||
`asyncio.CancelledError` не оборачивается — отмена задачи проходит как есть.
|
||||
"""
|
||||
|
||||
import json
|
||||
from io import BytesIO
|
||||
from typing import Any
|
||||
|
||||
from miniopy_async import Minio
|
||||
from miniopy_async.error import S3Error
|
||||
|
||||
from src.core.config import MinioSettings
|
||||
|
||||
OCTET_STREAM = "application/octet-stream"
|
||||
JSON_CONTENT_TYPE = "application/json"
|
||||
NOT_FOUND_CODES = frozenset({"NoSuchKey"})
|
||||
|
||||
|
||||
class S3StorageError(Exception):
|
||||
"""Ошибка операции с S3; `bucket` и `key` — объект, на котором она случилась."""
|
||||
|
||||
def __init__(self, message: str, *, bucket: str, key: str) -> None:
|
||||
super().__init__(message)
|
||||
self.bucket = bucket
|
||||
self.key = key
|
||||
|
||||
|
||||
class S3ReadError(S3StorageError):
|
||||
"""Объект не прочитан (нет доступа, нет бакета, сбой сети, ...)."""
|
||||
|
||||
|
||||
class S3ObjectNotFoundError(S3ReadError):
|
||||
"""Объекта с таким ключом в бакете нет."""
|
||||
|
||||
|
||||
class S3WriteError(S3StorageError):
|
||||
"""Объект не записан."""
|
||||
|
||||
|
||||
class S3Storage:
|
||||
def __init__(self, client: Minio, bucket_name: str):
|
||||
"""Операции сервиса над одним бакетом (`OPENBAO.MINIO.MINIO_BUCKET`); чтение может явно указать другой бакет."""
|
||||
|
||||
def __init__(self, client: Minio, bucket: str) -> None:
|
||||
self.client = client
|
||||
self.bucket = bucket_name
|
||||
self.bucket = bucket
|
||||
|
||||
async def put_object(self, obj_name, file_data):
|
||||
try:
|
||||
stream = BytesIO(file_data)
|
||||
await self.client.put_object(self.bucket, obj_name, stream, length=len(file_data))
|
||||
except:
|
||||
raise
|
||||
@classmethod
|
||||
def from_settings(cls, settings: MinioSettings) -> "S3Storage":
|
||||
"""Клиент MinIO из настроек: `MINIO_SECURE=false` — http (compose), `true` — https (платформа)."""
|
||||
client = Minio(
|
||||
settings.MINIO_URL,
|
||||
access_key=settings.MINIO_ADMIN_LOGIN,
|
||||
secret_key=settings.MINIO_ADMIN_PASSWORD,
|
||||
secure=settings.MINIO_SECURE,
|
||||
)
|
||||
return cls(client, settings.MINIO_BUCKET)
|
||||
|
||||
async def get_object(self, obj_name):
|
||||
async def get_bytes(self, key: str, *, bucket: str | None = None) -> bytes:
|
||||
"""Содержимое объекта целиком. `bucket` — из запроса (`s3.bucket`), по умолчанию — бакет сервиса.
|
||||
|
||||
Нет объекта → `S3ObjectNotFoundError`; иная ошибка → `S3ReadError`.
|
||||
"""
|
||||
bucket_name = bucket or self.bucket
|
||||
try:
|
||||
response = await self.client.get_object(self.bucket, obj_name)
|
||||
# file_bytes = await response.content.read()
|
||||
return response
|
||||
except:
|
||||
raise
|
||||
response = await self.client.get_object(bucket_name, key)
|
||||
try:
|
||||
data: bytes = await response.read()
|
||||
finally:
|
||||
response.release()
|
||||
except S3Error as exc:
|
||||
error_class = S3ObjectNotFoundError if exc.code in NOT_FOUND_CODES else S3ReadError
|
||||
raise error_class(
|
||||
f"S3: объект {bucket_name}/{key} не прочитан ({exc.code})", bucket=bucket_name, key=key
|
||||
) from exc
|
||||
except Exception as exc:
|
||||
raise S3ReadError(
|
||||
f"S3: объект {bucket_name}/{key} не прочитан ({type(exc).__name__})", bucket=bucket_name, key=key
|
||||
) from exc
|
||||
return data
|
||||
|
||||
async def put_bytes(self, key: str, data: bytes, content_type: str = OCTET_STREAM) -> None:
|
||||
"""Записывает `data` в бакет сервиса; ошибка → `S3WriteError`."""
|
||||
try:
|
||||
await self.client.put_object(self.bucket, key, BytesIO(data), length=len(data), content_type=content_type)
|
||||
except Exception as exc:
|
||||
raise S3WriteError(
|
||||
f"S3: объект {self.bucket}/{key} не записан ({type(exc).__name__})", bucket=self.bucket, key=key
|
||||
) from exc
|
||||
|
||||
async def put_json(self, key: str, obj: Any) -> None:
|
||||
"""Записывает `obj` как JSON в UTF-8 без `\\u`-экранирования, `Content-Type: application/json`."""
|
||||
data = json.dumps(obj, ensure_ascii=False).encode("utf-8")
|
||||
await self.put_bytes(key, data, content_type=JSON_CONTENT_TYPE)
|
||||
|
||||
async def close(self) -> None:
|
||||
"""Закрывает HTTP-сессию клиента (при остановке приложения)."""
|
||||
await self.client.close_session()
|
||||
|
||||
@ -1,19 +0,0 @@
|
||||
"""Обработчики сообщений NATS.
|
||||
|
||||
Каркас: доменный обработчик удалён. Consumer `recognition.result` (дедуп по recognition_job.result_msg_id,
|
||||
маппинг JSON из S3 в document/osv_entry/financial_statement_*) реализуется в Срезе 0–2 по
|
||||
docs/context/03-architecture.md. Для референса — reference-old-arm/api/src/core/nats_callback.py.
|
||||
"""
|
||||
|
||||
from loguru import logger
|
||||
from nats.aio.client import Client as NATS
|
||||
|
||||
from src.core.clients import S3Storage
|
||||
|
||||
|
||||
def build_result_handler(nc: NATS, storage: S3Storage):
|
||||
async def handler(msg):
|
||||
logger.info("NATS message received. subject={subject}, size={size}", subject=msg.subject, size=len(msg.data))
|
||||
await msg.ack()
|
||||
|
||||
return handler
|
||||
201
api/tests/unit/test_clients.py
Normal file
201
api/tests/unit/test_clients.py
Normal file
@ -0,0 +1,201 @@
|
||||
"""S3-клиент сервиса (`src.core.clients.S3Storage`) поверх фейкового `Minio`: аргументы вызовов, JSON в UTF-8,
|
||||
классификация ошибок чтения/записи. Настоящий MinIO — в tests/integration/test_nats_minio_integration.py.
|
||||
"""
|
||||
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from src.core import clients
|
||||
from src.core.clients import (
|
||||
S3ObjectNotFoundError,
|
||||
S3ReadError,
|
||||
S3Storage,
|
||||
S3StorageError,
|
||||
S3WriteError,
|
||||
)
|
||||
from src.core.config import get_settings
|
||||
|
||||
|
||||
class FakeResponse:
|
||||
def __init__(self, data: bytes) -> None:
|
||||
self._data = data
|
||||
self.released = False
|
||||
|
||||
async def read(self) -> bytes:
|
||||
return self._data
|
||||
|
||||
def release(self) -> None:
|
||||
self.released = True
|
||||
|
||||
|
||||
class FakeS3Error(Exception):
|
||||
"""Подмена `miniopy_async.error.S3Error`: у настоящего — атрибут `code` (`NoSuchKey`, `AccessDenied`, ...)."""
|
||||
|
||||
def __init__(self, code: str) -> None:
|
||||
super().__init__(code)
|
||||
self.code = code
|
||||
|
||||
|
||||
class FakeMinio:
|
||||
def __init__(self) -> None:
|
||||
self.objects: dict[tuple[str, str], tuple[bytes, str]] = {}
|
||||
self.get_calls: list[tuple[str, str]] = []
|
||||
self.put_calls: list[dict[str, Any]] = []
|
||||
self.responses: list[FakeResponse] = []
|
||||
self.get_error: BaseException | None = None
|
||||
self.put_error: BaseException | None = None
|
||||
self.session_closed = False
|
||||
|
||||
async def get_object(self, bucket_name: str, object_name: str) -> FakeResponse:
|
||||
self.get_calls.append((bucket_name, object_name))
|
||||
if self.get_error is not None:
|
||||
raise self.get_error
|
||||
response = FakeResponse(self.objects[(bucket_name, object_name)][0])
|
||||
self.responses.append(response)
|
||||
return response
|
||||
|
||||
async def put_object(
|
||||
self, bucket_name: str, object_name: str, data: Any, length: int, content_type: str = ""
|
||||
) -> None:
|
||||
body = data.read()
|
||||
self.put_calls.append(
|
||||
{"bucket": bucket_name, "key": object_name, "data": body, "length": length, "content_type": content_type}
|
||||
)
|
||||
if self.put_error is not None:
|
||||
raise self.put_error
|
||||
self.objects[(bucket_name, object_name)] = (body, content_type)
|
||||
|
||||
async def close_session(self) -> None:
|
||||
self.session_closed = True
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def minio() -> FakeMinio:
|
||||
return FakeMinio()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def storage(minio: FakeMinio) -> S3Storage:
|
||||
return S3Storage(minio, "arm-ka")
|
||||
|
||||
|
||||
async def test_put_bytes_passes_bucket_key_length_and_content_type(storage: S3Storage, minio: FakeMinio):
|
||||
await storage.put_bytes("uploads/a.pdf", b"%PDF-1.7", content_type="application/pdf")
|
||||
|
||||
assert minio.put_calls == [
|
||||
{
|
||||
"bucket": "arm-ka",
|
||||
"key": "uploads/a.pdf",
|
||||
"data": b"%PDF-1.7",
|
||||
"length": 8,
|
||||
"content_type": "application/pdf",
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
async def test_put_bytes_default_content_type_is_octet_stream(storage: S3Storage, minio: FakeMinio):
|
||||
await storage.put_bytes("k", b"x")
|
||||
assert minio.put_calls[0]["content_type"] == "application/octet-stream"
|
||||
|
||||
|
||||
async def test_put_json_writes_utf8_without_ascii_escaping(storage: S3Storage, minio: FakeMinio):
|
||||
payload = {"error": {"code": "INVALID_DOCUMENT", "message": "Документ не распознан"}, "n": 1}
|
||||
|
||||
await storage.put_json("raw/job.json", payload)
|
||||
|
||||
[call] = minio.put_calls
|
||||
assert call["key"] == "raw/job.json"
|
||||
assert call["content_type"] == "application/json"
|
||||
assert "Документ не распознан".encode() in call["data"]
|
||||
assert call["length"] == len(call["data"])
|
||||
assert json.loads(call["data"].decode("utf-8")) == payload
|
||||
|
||||
|
||||
async def test_get_bytes_reads_and_releases_response(storage: S3Storage, minio: FakeMinio):
|
||||
minio.objects[("arm-ka", "uploads/a.pdf")] = (b"%PDF", "application/pdf")
|
||||
|
||||
assert await storage.get_bytes("uploads/a.pdf") == b"%PDF"
|
||||
assert minio.get_calls == [("arm-ka", "uploads/a.pdf")]
|
||||
assert minio.responses[0].released is True
|
||||
|
||||
|
||||
async def test_get_bytes_from_explicit_bucket(storage: S3Storage, minio: FakeMinio):
|
||||
minio.objects[("other", "k")] = (b"data", "")
|
||||
assert await storage.get_bytes("k", bucket="other") == b"data"
|
||||
assert minio.get_calls == [("other", "k")]
|
||||
|
||||
|
||||
async def test_get_bytes_missing_key_raises_not_found(
|
||||
storage: S3Storage, minio: FakeMinio, monkeypatch: pytest.MonkeyPatch
|
||||
):
|
||||
monkeypatch.setattr(clients, "S3Error", FakeS3Error)
|
||||
minio.get_error = FakeS3Error("NoSuchKey")
|
||||
|
||||
with pytest.raises(S3ObjectNotFoundError) as exc_info:
|
||||
await storage.get_bytes("uploads/missing.pdf")
|
||||
|
||||
assert isinstance(exc_info.value, S3ReadError)
|
||||
assert exc_info.value.key == "uploads/missing.pdf"
|
||||
assert exc_info.value.bucket == "arm-ka"
|
||||
assert exc_info.value.__cause__ is minio.get_error
|
||||
|
||||
|
||||
@pytest.mark.parametrize("error", [FakeS3Error("AccessDenied"), ConnectionRefusedError(), RuntimeError("x")])
|
||||
async def test_get_bytes_other_errors_raise_read_error(
|
||||
storage: S3Storage, minio: FakeMinio, monkeypatch: pytest.MonkeyPatch, error: BaseException
|
||||
):
|
||||
monkeypatch.setattr(clients, "S3Error", FakeS3Error)
|
||||
minio.get_error = error
|
||||
|
||||
with pytest.raises(S3ReadError) as exc_info:
|
||||
await storage.get_bytes("uploads/a.pdf")
|
||||
|
||||
assert not isinstance(exc_info.value, S3ObjectNotFoundError)
|
||||
assert exc_info.value.__cause__ is error
|
||||
|
||||
|
||||
async def test_put_errors_raise_write_error(storage: S3Storage, minio: FakeMinio):
|
||||
minio.put_error = ConnectionRefusedError()
|
||||
|
||||
with pytest.raises(S3WriteError) as exc_info:
|
||||
await storage.put_json("raw/job.json", {"a": 1})
|
||||
|
||||
assert isinstance(exc_info.value, S3StorageError)
|
||||
assert exc_info.value.key == "raw/job.json"
|
||||
assert exc_info.value.__cause__ is minio.put_error
|
||||
|
||||
|
||||
async def test_close_closes_http_session(storage: S3Storage, minio: FakeMinio):
|
||||
await storage.close()
|
||||
assert minio.session_closed is True
|
||||
|
||||
|
||||
def test_from_settings_builds_minio_with_secure_flag(monkeypatch: pytest.MonkeyPatch):
|
||||
created: list[dict[str, Any]] = []
|
||||
|
||||
def fake_minio(endpoint: str, **kwargs: Any) -> FakeMinio:
|
||||
created.append({"endpoint": endpoint, **kwargs})
|
||||
return FakeMinio()
|
||||
|
||||
monkeypatch.setattr(clients, "Minio", fake_minio)
|
||||
settings = get_settings() # .env.test: 127.0.0.1:9000, minioadmin, бакет arm-ka, MINIO_SECURE=false
|
||||
|
||||
storage = S3Storage.from_settings(settings.OPENBAO.MINIO)
|
||||
|
||||
assert created == [
|
||||
{"endpoint": "127.0.0.1:9000", "access_key": "minioadmin", "secret_key": "minioadmin", "secure": False}
|
||||
]
|
||||
assert storage.bucket == "arm-ka"
|
||||
|
||||
|
||||
def test_from_settings_secure_true(monkeypatch: pytest.MonkeyPatch):
|
||||
created: list[dict[str, Any]] = []
|
||||
monkeypatch.setattr(clients, "Minio", lambda endpoint, **kwargs: created.append(kwargs) or FakeMinio())
|
||||
monkeypatch.setenv("OPENBAO__MINIO__MINIO_SECURE", "true")
|
||||
get_settings.cache_clear()
|
||||
|
||||
S3Storage.from_settings(get_settings().OPENBAO.MINIO)
|
||||
|
||||
assert created[0]["secure"] is True
|
||||
Loading…
x
Reference in New Issue
Block a user