Workers에서 R2 멀티파트 API 사용하기 (Use the R2 multipart API from Workers)
이 가이드를 따라가면 애플리케이션이 멀티파트 업로드를 수행할 수 있는 Worker를 만들게 돼요. 이 예시 Worker는 Worker에 인증을 추가하거나 각 파트를 업로드할 때 추가 검증 로직을 넣는 식으로 자신의 사용 사례의 기반이 될 수 있어요. 이 가이드에는 이 Worker에 파일을 업로드하는 예시 Python 애플리케이션도 포함되어 있어요.
출처: 문서
본문
이 가이드는 Worker에 R2 바인딩을 설정했다고 가정해요. R2 바인딩 설정 방법은 Workers에서 R2 사용하기를 참고해 주세요.
멀티파트 API를 사용하는 예시 Worker
다음 예시 Worker는 애플리케이션이 Worker를 통해 멀티파트 API를 사용할 수 있게 해 주는 HTTP API를 노출해요.
이 예시에서는 각 요청이 HTTP 메서드와 action 요청 파라미터에 따라 라우팅돼요. Worker가 복잡해지면 Hono ↗︎ 같은 서버리스 웹 프레임워크를 사용해 라우팅을 처리하는 것을 고려해 보세요.
다음 예시 Worker는 멀티파트 업로드 상태에 대한 새로운 정보를 각 요청의 응답에 포함해요. 멀티파트 업로드를 생성하는 요청에서는 uploadId가 반환돼요. 파트를 업로드하는 요청에서는 파트 번호와 etag가 반환돼요. 그러면 클라이언트가 이 상태를 추적하고, 이후 요청에 uploadId를 포함하며, 멀티파트 업로드를 완료할 때 각 파트의 etag와 파트 번호를 포함해요.
다음 코드를 프로젝트의 index.ts 파일에 추가하고 MY_BUCKET을 버킷 이름으로 바꿔요:
interface Env {
MY_BUCKET: R2Bucket;
}
export default {
async fetch(
request,
env,
ctx
): Promise<Response> {
const bucket = env.MY_BUCKET;
const url = new URL(request.url);
const key = url.pathname.slice(1);
const action = url.searchParams.get("action");
if (action === null) {
return new Response("Missing action type", { status: 400 });
}
// Route the request based on the HTTP method and action type
switch (request.method) {
case "POST":
switch (action) {
case "mpu-create": {
const multipartUpload = await bucket.createMultipartUpload(key);
return new Response(
JSON.stringify({
key: multipartUpload.key,
uploadId: multipartUpload.uploadId,
})
);
}
case "mpu-complete": {
const uploadId = url.searchParams.get("uploadId");
if (uploadId === null) {
return new Response("Missing uploadId", { status: 400 });
}
const multipartUpload = env.MY_BUCKET.resumeMultipartUpload(
key,
uploadId
);
interface completeBody {
parts: R2UploadedPart[];
}
const completeBody: completeBody = await request.json();
if (completeBody === null) {
return new Response("Missing or incomplete body", {
status: 400,
});
}
// Error handling in case the multipart upload does not exist anymore
try {
const object = await multipartUpload.complete(completeBody.parts);
return new Response(null, {
headers: {
etag: object.httpEtag,
},
});
} catch (error: any) {
return new Response(error.message, { status: 400 });
}
}
default:
return new Response(`Unknown action ${action} for POST`, {
status: 400,
});
}
case "PUT":
switch (action) {
case "mpu-uploadpart": {
const uploadId = url.searchParams.get("uploadId");
const partNumberString = url.searchParams.get("partNumber");
if (partNumberString === null || uploadId === null) {
return new Response("Missing partNumber or uploadId", {
status: 400,
});
}
if (request.body === null) {
return new Response("Missing request body", { status: 400 });
}
const partNumber = parseInt(partNumberString);
const multipartUpload = env.MY_BUCKET.resumeMultipartUpload(
key,
uploadId
);
try {
const uploadedPart: R2UploadedPart =
await multipartUpload.uploadPart(partNumber, request.body);
return new Response(JSON.stringify(uploadedPart));
} catch (error: any) {
return new Response(error.message, { status: 400 });
}
}
default:
return new Response(`Unknown action ${action} for PUT`, {
status: 400,
});
}
case "GET":
if (action !== "get") {
return new Response(`Unknown action ${action} for GET`, {
status: 400,
});
}
const object = await env.MY_BUCKET.get(key);
if (object === null) {
return new Response("Object Not Found", { status: 404 });
}
const headers = new Headers();
object.writeHttpMetadata(headers);
headers.set("etag", object.httpEtag);
return new Response(object.body, { headers });
case "DELETE":
switch (action) {
case "mpu-abort": {
const uploadId = url.searchParams.get("uploadId");
if (uploadId === null) {
return new Response("Missing uploadId", { status: 400 });
}
const multipartUpload = env.MY_BUCKET.resumeMultipartUpload(
key,
uploadId
);
try {
multipartUpload.abort();
} catch (error: any) {
return new Response(error.message, { status: 400 });
}
return new Response(null, { status: 204 });
}
case "delete": {
await env.MY_BUCKET.delete(key);
return new Response(null, { status: 204 });
}
default:
return new Response(`Unknown action ${action} for DELETE`, {
status: 400,
});
}
default:
return new Response("Method Not Allowed", {
status: 405,
headers: { Allow: "PUT, POST, GET, DELETE" },
});
}
},
} satisfies ExportedHandler<Env>;
from workers import WorkerEntrypoint, Response
from urllib.parse import urlparse, parse_qs
import json
class Default(WorkerEntrypoint):
async def fetch(self, request):
bucket = self.env.MY_BUCKET
url = urlparse(request.url)
key = url.path[1:]
params = parse_qs(url.query)
action = params.get("action", [None])[0]
if action is None:
return Response("Missing action type", status=400)
if request.method == "POST":
if action == "mpu-create":
multipart_upload = await bucket.createMultipartUpload(key)
return Response.json({
"key": multipart_upload.key,
"uploadId": multipart_upload.uploadId,
})
elif action == "mpu-complete":
upload_id = params.get("uploadId", [None])[0]
if upload_id is None:
return Response("Missing uploadId", status=400)
multipart_upload = bucket.resumeMultipartUpload(key, upload_id)
complete_body = await request.json()
if complete_body is None:
return Response("Missing or incomplete body", status=400)
try:
obj = await multipart_upload.complete(complete_body.parts)
return Response(None, headers={"etag": obj.httpEtag})
except Exception as error:
return Response(str(error), status=400)
else:
return Response(f"Unknown action {action} for POST", status=400)
elif request.method == "PUT":
if action == "mpu-uploadpart":
upload_id = params.get("uploadId", [None])[0]
part_number_str = params.get("partNumber", [None])[0]
if part_number_str is None or upload_id is None:
return Response("Missing partNumber or uploadId", status=400)
if request.body is None:
return Response("Missing request body", status=400)
part_number = int(part_number_str)
multipart_upload = bucket.resumeMultipartUpload(key, upload_id)
try:
uploaded_part = await multipart_upload.uploadPart(part_number, request.body)
return Response.json(uploaded_part)
except Exception as error:
return Response(str(error), status=400)
else:
return Response(f"Unknown action {action} for PUT", status=400)
elif request.method == "GET":
if action != "get":
return Response(f"Unknown action {action} for GET", status=400)
obj = await bucket.get(key)
if obj is None:
return Response("Object Not Found", status=404)
body = await obj.text()
headers = {"etag": obj.httpEtag}
return Response(body, headers=headers)
elif request.method == "DELETE":
if action == "mpu-abort":
upload_id = params.get("uploadId", [None])[0]
if upload_id is None:
return Response("Missing uploadId", status=400)
multipart_upload = bucket.resumeMultipartUpload(key, upload_id)
try:
await multipart_upload.abort()
except Exception as error:
return Response(str(error), status=400)
return Response(None, status=204)
elif action == "delete":
await bucket.delete(key)
return Response(None, status=204)
else:
return Response(f"Unknown action {action} for DELETE", status=400)
else:
return Response(
"Method Not Allowed",
status=405,
headers={"Allow": "PUT, POST, GET, DELETE"},
)
Worker를 위 코드로 업데이트한 후 npx wrangler deploy를 실행해요.
이제 이 Worker로 멀티파트 업로드를 수행할 수 있어요. 기존 애플리케이션에서 이 Worker로 요청을 보내 업로드를 수행하거나, 스크립트로 이 Worker를 통해 파일을 업로드해도 돼요.
다음 섹션은 선택 사항이며, 컴퓨터의 선택한 파일을 Worker에 업로드하는 예시 Python 스크립트를 보여줘요.
Worker로 멀티파트 업로드 수행하기 (선택 사항)
이 예시 애플리케이션은 로컬 파일을 여러 파트로 나눠 Worker에 업로드해요. Python 내장 ThreadPoolExecutor를 사용해 파트 업로드를 Worker에 병렬화해서 업로드 속도를 높여요. Worker로의 HTTP 요청은 requests ↗︎ 라이브러리로 만들어요.
이런 방식으로 멀티파트 API를 활용하면 Workers 요청 본문 크기 제한보다 큰 파일을 업로드할 때도 Worker를 사용할 수 있어요. 개별 파트의 업로드는 여전히 이 제한을 따르는 점을 유의해요.
다음 코드를 로컬 컴퓨터의 mpuscript.py 파일에 저장해요. worker_endpoint 변수를 Worker가 배포된 위치로 바꿔요. 이 스크립트를 실행할 때 업로드할 파일을 인자로 전달해요: python3 mpuscript.py myfile. 그러면 컴퓨터의 myfile 파일이 Worker를 통해 버킷에 업로드돼요.
import math
import os
import requests
from requests.adapters import HTTPAdapter, Retry
import sys
import concurrent.futures
# Take the file to upload as an argument
filename = sys.argv[1]
# The endpoint for our worker, change this to wherever you deploy your worker
worker_endpoint = "https://myworker.myzone.workers.dev/"
# Configure the part size to be 10MB. 5MB is the minimum part size, except for the last part
partsize = 10 * 1024 * 1024
def upload_file(worker_endpoint, filename, partsize):
url = f"{worker_endpoint}{filename}"
# Create the multipart upload
uploadId = requests.post(url, params={"action": "mpu-create"}).json()["uploadId"]
part_count = math.ceil(os.stat(filename).st_size / partsize)
# Create an executor for up to 25 concurrent uploads.
executor = concurrent.futures.ThreadPoolExecutor(25)
# Submit a task to the executor to upload each part
futures = [
executor.submit(upload_part, filename, partsize, url, uploadId, index)
for index in range(part_count)
]
concurrent.futures.wait(futures)
# get the parts from the futures
uploaded_parts = [future.result() for future in futures]
# complete the multipart upload
response = requests.post(
url,
params={"action": "mpu-complete", "uploadId": uploadId},
json={"parts": uploaded_parts},
)
if response.status_code == 200:
print("🎉 successfully completed multipart upload")
else:
print(response.text)
def upload_part(filename, partsize, url, uploadId, index):
# Open the file in rb mode, which treats it as raw bytes rather than attempting to parse utf-8
with open(filename, "rb") as file:
file.seek(partsize * index)
part = file.read(partsize)
# Retry policy for when uploading a part fails
s = requests.Session()
retries = Retry(total=3, status_forcelist=[400, 500, 502, 503, 504])
s.mount("https://", HTTPAdapter(max_retries=retries))
return s.put(
url,
params={
"action": "mpu-uploadpart",
"uploadId": uploadId,
"partNumber": str(index + 1),
},
data=part,
).json()
upload_file(worker_endpoint, filename, partsize)
상태 관리 (State management)
멀티파트 업로드의 상태 저장 특성은 본질적으로 상태가 없는(stateless) Workers의 사용 모델에 쉽게 매핑되지 않아요. 일반적인 멀티파트 업로드에서는 보통 클라이언트 애플리케이션의 연속적인 단일 실행으로 진행돼요. 하지만 Worker에서의 멀티파트 업로드는 Worker의 여러 호출에 걸쳐 완료되는 경우가 많아서 이것과 다르고, 상태 관리를 더 어렵게 만들어요.
이를 극복하려면 멀티파트 업로드와 관련된 상태, 즉 uploadId와 어떤 파트가 업로드되었는지를 Worker 밖 어딘가에서 추적해야 해요.
이 가이드에서 설명하는 예시 Worker와 Python 애플리케이션에서는 멀티파트 업로드의 상태를 Worker에 요청을 보내는 클라이언트 애플리케이션에서 추적하고, 필요한 상태를 각 요청에 포함해요. 클라이언트 애플리케이션에서 멀티파트 상태를 추적하면 최대한의 유연성이 확보되고, 각 파트를 병렬로 순서에 상관없이 업로드할 수 있어요.
클라이언트에서 이 상태를 추적하는 것이 불가능할 때는 대안 설계를 고려할 수 있어요. 예를 들어 uploadId와 어떤 파트가 업로드되었는지를 Durable Object나 다른 데이터베이스에서 추적할 수 있어요.
더 알아보기 (Learn more)
- Workers API 참조 —
createMultipartUpload,resumeMultipartUpload참조 - Workers에서 R2 사용하기 — R2 바인딩 설정