Mina Choi
3280cab088
POST /v1/item/image — SAS 토큰을 URL에 붙여 httpx PUT 으로 Azure Blob 업로드. 이미지 타입/용량 검증 + IMAGE_* 에러코드. config 에 azure_blob_base_url/sas/root 추가. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
51 lines
1.8 KiB
Python
51 lines
1.8 KiB
Python
import uuid
|
|
|
|
import httpx
|
|
|
|
from config.server_configs import storage_config
|
|
|
|
# 허용 content-type -> 저장 확장자. 목록 밖이면 업로드 거부.
|
|
_EXT_BY_TYPE = {
|
|
"image/jpeg": "jpg",
|
|
"image/png": "png",
|
|
"image/gif": "gif",
|
|
"image/webp": "webp",
|
|
}
|
|
|
|
# 업로드용 공유 클라이언트(연결 풀링).
|
|
_shared_client: httpx.AsyncClient | None = None
|
|
|
|
|
|
def _get_client() -> httpx.AsyncClient:
|
|
global _shared_client
|
|
if _shared_client is None or _shared_client.is_closed:
|
|
_shared_client = httpx.AsyncClient(
|
|
timeout=httpx.Timeout(60.0, connect=10.0),
|
|
limits=httpx.Limits(max_keepalive_connections=10, max_connections=20),
|
|
)
|
|
return _shared_client
|
|
|
|
|
|
def is_allowed_image(content_type: str) -> bool:
|
|
return content_type in _EXT_BY_TYPE
|
|
|
|
|
|
async def upload_image(company_id: str, content: bytes, content_type: str) -> str:
|
|
"""Azure Blob 에 이미지를 올리고 접근 URL 을 돌려준다.
|
|
blob 경로: <blob_root>/<company_id>/items/<uuid>.<ext>
|
|
— company_id 로 멀티테넌트 격리.
|
|
실패 시 예외를 던진다(호출측에서 IMAGE_UPLOAD_FAILED 처리). 저장값은 SAS 를 뗀 public URL."""
|
|
ext = _EXT_BY_TYPE.get(content_type, "bin")
|
|
blob_path = f"{storage_config.blob_root}/{company_id}/items/{uuid.uuid4().hex}.{ext}"
|
|
|
|
base = storage_config.azure_blob_base_url.rstrip("/")
|
|
sas = storage_config.azure_blob_sas_token.strip("?'\"")
|
|
public_url = f"{base}/{blob_path}"
|
|
upload_url = f"{public_url}?{sas}"
|
|
headers = {"Content-Type": content_type, "x-ms-blob-type": "BlockBlob"}
|
|
|
|
resp = await _get_client().put(upload_url, content=content, headers=headers)
|
|
if resp.status_code not in (200, 201):
|
|
raise RuntimeError(f"Azure Blob upload failed: status={resp.status_code} body={resp.text[:200]}")
|
|
return public_url
|