[FEATURE] S3 이미지 읽기 구현

.gitignore

@@ -40,7 +40,7 @@ MANIFEST
 pip-log.txt
 pip-delete-this-directory.txt
 
-# Unit test / coverage reports
+# Unit tests / coverage reports
 htmlcov/
 .tox/
 .nox/

app/api/endpoints/predictions.py

@@ -14,7 +14,7 @@ def get_redis_client(request: Request) -> redis.Redis:
 async def create_prediction_job(job_request: JobRequest, redis_client: redis.Redis = Depends(get_redis_client)):
     correlation_id = str(uuid.uuid4())
     job = ImageJob(
-        correlationId=correlationId,
+        correlationId=correlation_id,
         presignedUrl=job_request.presigned_url,
         replyQueue=settings.STREAM_RESULT,
         callbackUrl=None,

app/core/lifespan.py

@@ -1,26 +1,24 @@
 
 import asyncio
 from contextlib import asynccontextmanager
-import redis.asyncio as redis
+
+import anyio.to_thread
+import redis
 from fastapi import FastAPI
 
+from app.worker.redis_client import redis_client
 from app.core.config import settings
 from app.worker.worker import JobWorker
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    redis_client = redis.from_url(settings.REDIS_URL, decode_responses=True)
+    limiter = anyio.to_thread.current_default_thread_limiter()
+    limiter.total_tokens = 100
 
-    try:
-        await redis_client.xgroup_create(
-            name=settings.STREAM_JOB,
-            groupname=settings.GROUP_NAME,
-            id="$",
-            mkstream=True
-        )
-    except redis.ResponseError as e:
-        if "BUSYGROUP" not in str(e):
-            raise
+    redis_client.xgroup_create(
+        stream_name=settings.STREAM_JOB,
+        group_name=settings.GROUP_NAME,
+    )
 
     worker = JobWorker(redis_client)
     worker_task = asyncio.create_task(worker.run())

app/services/predictor_service.py

@@ -6,6 +6,7 @@
 from PIL import Image
 import json
 from pathlib import Path
+from io import BytesIO
 
 class LightCNN(nn.Module):
     def __init__(self, num_classes):
@@ -53,8 +54,8 @@ def _load_model(self, model_path: Path) -> LightCNN:
         model.eval()
         return model
 
-    def predict(self, image_path: Path) -> tuple[str, str, float]:
-        image = Image.open(image_path).convert('RGB')
+    def predict(self, stream_file: BytesIO) -> tuple[str, str, float]:
+        image = Image.open(stream_file).convert('RGB')
         input_tensor = self.transform(image).unsqueeze(0).to(self.device)
 
         with torch.no_grad():

app/services/s3_service.py

@@ -1,8 +1,8 @@
 
 import boto3
 from botocore.config import Config as BotoConfig
-from pathlib import Path
 import requests
+from io import BytesIO
 
 from app.core.config import settings
 
@@ -18,10 +18,10 @@ def __init__(self):
             ),
         )
 
-    def download_file_from_presigned_url(self, presigned_url: str, destination: Path):
+    def download_file_from_presigned_url(self, presigned_url: str) -> BytesIO:
         response = requests.get(presigned_url)
         response.raise_for_status()
-        with open(destination, "wb") as f:
-            f.write(response.content)
+
+        return BytesIO(response.content) # response 안의 content Stream으로 처리
 
 s3_service = S3Service()

app/worker/redis_client.py

@@ -1,10 +1,107 @@
+import asyncio
 import redis
+
+from redis.asyncio import Redis as AsyncRedis
 from pydantic import BaseModel, Field
-from typing import Any, Dict
+from typing import Any, Dict, Optional, List
 from app.core.config import settings
 
-redis_client = redis.asyncio.from_url(settings.REDIS_URL, decode_responses=True)
-
+# Redis Stream Definition
 class PublishRequest(BaseModel):
-    stream: str = Field(default=settings.JOB_STREAM, description="Redis Stream Job name")
-    payload: Dict[str, Any]
+    stream: str = Field(default=settings.STREAM_JOB, description="Redis Stream Job name")
+    payload: Dict[str, Any]
+
+class RedisStreamClient:
+
+    def __init__(self):
+        self.redis_client = AsyncRedis.from_url(
+            url=settings.REDIS_URL,
+            decode_responses=True,
+        )
+
+    @classmethod
+    def init(cls):
+        broker = cls()
+        return broker
+
+    # Fast API 에서 Publish
+    async def xadd(self, stream_name: str, fields: Dict[str, Any]) -> str:
+        return await self.redis_client.xadd(stream_name, fields)
+
+    # Group 단위로 읽어오기
+    async def xreadgroup(
+            self,
+            group_name: str,
+            consumer_name: str,
+            stream_name: str,
+            count: Optional[int] = None,
+            block: Optional[int] = None,  # ms 단위
+            id: str = ">",  # 새 메시지만 읽기
+    ) -> List[tuple]:
+        try:
+            # Create the consumer group (존재하지 않을 때)
+            self.redis_client.xgroup_create(
+                stream_name, group_name, id="$", mkstream=True
+            )
+        except redis.exceptions.ResponseError as e:
+            # 이미 존재할 때
+            if "BUSYGROUP" not in str(e):
+                raise
+
+        streams = {stream_name: id}
+        response = await self.redis_client.xreadgroup(
+            group_name,  # groupname (positional)
+            consumer_name,  # consumername (positional)
+            streams,  # {stream: id} (positional)
+            count=count,
+            block=block,
+        )
+        return response
+
+    # Consumer 처리 완료
+    async def xack(self, stream_name: str, group_name: str, message_ids: List[str]) -> int:
+        return await self.redis_client.xack(stream_name, group_name, *message_ids)
+
+    # 완료 시 삭제
+    async def xack_and_del(self, stream_name: str, group_name: str, message_ids: str) -> int:
+
+        acked_count = await self.redis_client.xack(stream_name, group_name, *message_ids)
+
+        # XACK가 성공하면 스트림에서 해당 메시지를 삭제 (XDEL)
+        if acked_count > 0:
+            await self.redis_client.xdel(stream_name, *message_ids)
+
+        return acked_count
+
+    async def xgroup_create(self, stream_name: str, group_name: str, id: str = "$") -> bool:
+        try:
+            self.redis_client.xgroup_create(stream_name, group_name, id, mkstream=True)
+            return True
+        except redis.exceptions.ResponseError as e:
+            if "BUSYGROUP" in str(e):
+                print(f"Consumer group '{group_name}' already exists.")
+                return False
+            raise e
+
+    # 메시지 재처리 지원
+    async def xclaim(
+            self,
+            stream_name: str,
+            group_name: str,
+            consumer_name: str,
+            min_idle_time: int,
+            message_ids: List[str],
+    ) -> List[tuple]:
+
+        return await self.redis_client.xclaim(
+            stream_name=stream_name,
+            group_name=group_name,
+            consumer_name=consumer_name,
+            min_idle_time=min_idle_time,
+            message_ids=message_ids,
+        )
+
+    async def aclose(self):
+        await self.redis_client.close()
+
+redis_client = RedisStreamClient.init()

app/worker/tasks.py

@@ -1,7 +1,5 @@
 
 import asyncio
-import json
-from pathlib import Path
 import redis.asyncio as redis
 from datetime import datetime
 
@@ -13,13 +11,10 @@
 async def process_image_scan(job: ImageJob, redis_client: redis.Redis):
     correlation_id = job.correlationId
     print(f"[task] Start image scan for job_id={correlation_id}")
-
-    temp_image_path = Path(f"/tmp/{correlation_id}.jpg")
-
     try:
-        s3_service.download_file_from_presigned_url(job.presignedUrl, temp_image_path)
+        stream_file = await asyncio.to_thread(s3_service.download_file_from_presigned_url(job.presignedUrl))
 
-        pill_name, label, confidence = predictor_service.predict(temp_image_path)
+        pill_name, label, confidence = await asyncio.to_thread(predictor_service.predict(stream_file))
 
         finished_at = datetime.utcnow().isoformat()
 
@@ -43,5 +38,4 @@ async def process_image_scan(job: ImageJob, redis_client: redis.Redis):
     except Exception as e:
         print(f"[task] Failed to process job_id={correlation_id}: {e}")
     finally:
-        if temp_image_path.exists():
-            temp_image_path.unlink()
+        print(f"[task] Image scan finished for job_id={correlation_id}")

app/worker/worker.py

@@ -1,12 +1,12 @@
 
 import asyncio
-import redis.asyncio as redis
-from app.core.config import Settings
+from app.worker.redis_client import redis_client
+from app.core.config import settings
 from app.schemas.job import ImageJob
 from app.worker.tasks import process_image_scan
 
 class JobWorker:
-    def __init__(self, redis_client: redis.Redis):
+    def __init__(self, redis_client: redis_client):
         self.redis_client = redis_client
 
     async def run(self):
@@ -17,9 +17,9 @@ async def run(self):
         while True:
             try:
                 resp = await self.redis_client.xreadgroup(
-                    groupname=settings.GROUP_NAME,
-                    consumername=settings.CONSUMER_NAME,
-                    streams={settings.STREAM_JOB: ">"},
+                    group_name=settings.GROUP_NAME,
+                    consumer_name=settings.CONSUMER_NAME,
+                    stream_name=settings.STREAM_JOB,
                     count=10,
                     block=5000,
                 )
@@ -28,9 +28,15 @@ async def run(self):
                     for msg_id, fields in entries:
                         try:
                             job = ImageJob.model_validate_json(fields["json"])
-                            await process_image_scan(job, redis_client)
-                            # 처리 성공 시에만 ack
-                            await redis_client.xack(settings.STREAM_JOB, settings.GROUP_NAME, msg_id)
+                            task = asyncio.create_task(process_image_scan(job, redis_client))
+                            print(f"[worker] {task} 발행 성공")
+                            # 처리 성공 시에만 ack 후 del
+                            task.add_done_callback(lambda t: asyncio.create_task(
+                                self.redis_client.xack_and_del(settings.STREAM_JOB, settings.GROUP_NAME, msg_id)
+                                if not t.exception() else
+                                self.redis_client.xadd(f"{settings.STREAM_JOB}:DLQ",
+                                                       {"id": msg_id, "error": str(t.exception()), **fields})
+                            ))
                         except asyncio.CancelledError:
                             # 취소되면 재전송되도록 ack 하지 않음
                             raise
@@ -54,8 +60,15 @@ async def run(self):
                     for msg_id, fields in claimed:
                         try:
                             job = ImageJob.model_validate_json(fields["json"])
-                            asyncio.create_task(process_image_scan(job, self.redis_client))
-                            await self.redis_client.xack(settings.STREAM_JOB, settings.GROUP_NAME, msg_id)
+                            task = asyncio.create_task(process_image_scan(job, redis_client))
+                            print(f"[worker] {task} 발행 성공")
+                            # 처리 성공 시에만 ack 후 del
+                            task.add_done_callback(lambda t: asyncio.create_task(
+                                self.redis_client.xack_and_del(settings.STREAM_JOB, settings.GROUP_NAME, msg_id)
+                                if not t.exception() else
+                                self.redis_client.xadd(f"{settings.STREAM_JOB}:DLQ",
+                                                       {"id": msg_id, "error": str(t.exception()), **fields})
+                            ))
                         except Exception as e:
                             await self.redis_client.xadd(
                                 f"{settings.STREAM_JOB}:DLQ",

tests/s3_service_test.py

@@ -0,0 +1,39 @@
+import io
+import pytest
+from unittest.mock import patch, MagicMock
+
+from app.services.s3_service import s3_service
+
+class TestDownloadFile:
+    @patch("app.services.s3_service.requests.get")
+    def test_download_file_success(self, mock_get):
+        # arrange
+        mock_get.return_value = MagicMock()
+        mock_get.return_value.content = b"test file content"
+        mock_get.return_value.raise_for_status = MagicMock()
+
+        # act
+        obj = s3_service
+        result = obj.download_file_from_presigned_url(
+            "http://fake-url.com"
+        )
+
+        # assert
+        assert isinstance(result, io.BytesIO)
+        assert result.getvalue() == b"test file content"
+        mock_get.assert_called_once_with(
+            "http://fake-url.com"
+        )
+        mock_get.return_value.raise_for_status.assert_called_once()
+
+    @patch("app.services.s3_service.requests.get")
+    def test_download_file_http_error(self, mock_get):
+        # arrange
+        mock_get.return_value = MagicMock()
+        mock_get.return_value.raise_for_status.side_effect = Exception("HTTP Error")
+
+        obj = s3_service
+
+        # act & assert
+        with pytest.raises(Exception, match="HTTP Error"):
+            obj.download_file_from_presigned_url("http://fake-url.com")

tests/test_main.py

@@ -0,0 +1,10 @@
+from fastapi.testclient import TestClient
+
+from app.main import app
+
+client = TestClient(app=app)
+
+def test_health_check():
+    response = client.get("/api/v1/health")
+    assert response.status_code == 200
+    assert response.json() == {"status": "ok"}