64 lines
2.2 KiB
Python
64 lines
2.2 KiB
Python
"""Azure Blob Storage 업로드.
|
|
|
|
SAS 토큰을 붙인 URL로 PUT 한 번. 경로 규칙은 부르는 쪽이 정한다.
|
|
클라이언트를 모듈에서 공유해 커넥션 풀을 재사용한다.
|
|
"""
|
|
import mimetypes
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
from settings import settings
|
|
|
|
BLOB_TYPE_HEADER = {"x-ms-blob-type": "BlockBlob"}
|
|
DEFAULT_CONTENT_TYPE = "application/octet-stream"
|
|
UPLOAD_TIMEOUT = httpx.Timeout(180.0, connect=10.0)
|
|
POOL_LIMITS = httpx.Limits(max_keepalive_connections=10, max_connections=20)
|
|
|
|
_client: httpx.AsyncClient | None = None
|
|
|
|
|
|
def get_client() -> httpx.AsyncClient:
|
|
global _client
|
|
if _client is None or _client.is_closed:
|
|
_client = httpx.AsyncClient(timeout=UPLOAD_TIMEOUT, limits=POOL_LIMITS)
|
|
return _client
|
|
|
|
|
|
async def close_client() -> None:
|
|
global _client
|
|
if _client is not None and not _client.is_closed:
|
|
await _client.aclose()
|
|
_client = None
|
|
|
|
|
|
def guess_content_type(path: str) -> str:
|
|
return mimetypes.guess_type(path)[0] or DEFAULT_CONTENT_TYPE
|
|
|
|
|
|
def public_url(path: str) -> str:
|
|
return f"{settings.azure_blob_base_url.rstrip('/')}/{path.lstrip('/')}"
|
|
|
|
|
|
def upload_url(path: str) -> str:
|
|
# SAS 토큰이 따옴표나 ? 로 감싸여 오는 경우가 있다
|
|
token = settings.azure_blob_sas_token.strip("?'\"")
|
|
return f"{public_url(path)}?{token}"
|
|
|
|
|
|
async def upload_bytes(data: bytes, path: str, content_type: str | None = None) -> str:
|
|
"""path 예: playreel/nol_26007416/clip.mp4. 공개 URL을 돌려준다."""
|
|
if not settings.azure_blob_base_url or not settings.azure_blob_sas_token:
|
|
raise RuntimeError("AZURE_BLOB_BASE_URL / AZURE_BLOB_SAS_TOKEN 없음")
|
|
|
|
headers = {"Content-Type": content_type or guess_content_type(path), **BLOB_TYPE_HEADER}
|
|
response = await get_client().put(upload_url(path), content=data, headers=headers)
|
|
if response.status_code not in (200, 201):
|
|
raise RuntimeError(f"blob 업로드 실패 {response.status_code}: {response.text[:300]}")
|
|
return public_url(path)
|
|
|
|
|
|
async def upload_file(file_path: Path, path: str, content_type: str | None = None) -> str:
|
|
return await upload_bytes(file_path.read_bytes(), path,
|
|
content_type or guess_content_type(file_path.name))
|