このチュートリアルでは、Python SDKを使用してBoxにコンテンツをアップロードする2つの方法 (と) を取り上げます。分割アップロードには、SDKの便利なメソッドを使う方法と、アップロードセッションを自分で管理する方法があります。各方法の仕組みと、と競合処理を組み合わせて新規ファイルをアップロードしたり、既存のファイルを更新したりする方法について説明します。
| メソッド | 最適な用途 | 最小サイズ | 最大サイズ |
|---|
| サイズの小さいファイル、シンプルなワークフロー | 下限なし | 50 MB |
| サイズの大きいファイル | 20 MB | 上限なし |
このチュートリアルのコードサンプルでは、 を使用しています。
前提条件
- Boxアプリケーションで [] スコープ (
root_readwrite) が有効になっている
- コンピュータにPython 3.8以降がインストールされている
- Box Python SDKがインストールされている (
pip install "boxsdk~=10")
- が認証済みである
直接アップロード
ファイルをアップロードするには、upload_fileメソッドを呼び出します。直接アップロードAPIの詳細については、を参照してください。
最初にを実行します。同名のファイルがすでに存在する場合は、エラーレスポンスに競合ファイルのIDが含まれます。その名前のファイルのコンテンツを上書きしたくない場合は、名前を変更して新規ファイルとしてアップロードします。以下のサンプルでは、ファイルのコンテンツを上書きする必要があると想定し、そのIDを使用してをアップロードしています。
import os
from box_sdk_gen import (
BoxAPIError,
PreflightFileUploadCheckParent,
UploadFileAttributes,
UploadFileAttributesParentField,
UploadFileVersionAttributes,
)
def upload_file_or_file_version(client, file_path, folder_id):
file_size = os.path.getsize(file_path)
file_name = os.path.basename(file_path)
with open(file_path, "rb") as file_stream:
try:
client.uploads.preflight_file_upload_check(
name=file_name,
size=file_size,
parent=PreflightFileUploadCheckParent(id=folder_id),
)
files = client.uploads.upload_file(
UploadFileAttributes(
name=file_name,
parent=UploadFileAttributesParentField(id=folder_id),
),
file_stream,
)
return files.entries[0]
except BoxAPIError as err:
if err.response_info.code != "item_name_in_use":
raise
file_id = err.response_info.context_info["conflicts"]["id"]
file_stream.seek(0)
files = client.uploads.upload_file_version(
file_id,
UploadFileVersionAttributes(name=file_name),
file_stream,
)
return files.entries[0]
upload_fileを呼び出すと新規ファイルが作成されます。既存ファイルでupload_file_versionを呼び出すと新しいバージョンがアップロードされます。
分割アップロード
20 MB以上のファイルにはを使用します。各分割アップロードはを通じて行われます。アップロードセッションでは、セッションを作成して、し、することでファイルを確定します。これには、2つの方法があります。
- 便利なメソッド:
upload_big_fileを使用すると、3つのステップすべてが自動で行われます。
- 手動のアップロードセッション: 詳細な制御が必要な場合は、セッションの作成、各パーツのアップロード、コミットを自分で行います。
便利なメソッド
upload_big_fileメソッドを使用すると、パーツ単位でファイルが自動的にアップロードされます。アップロード開始前に名前の競合を処理できるよう、最初に事前チェックを実行します。同名のファイルがすでに存在する場合は、upload_chunked_fileによって既存ファイル用のセッションが作成され、次のセクションで示すようにupload_file_via_sessionヘルパーによって新しいバージョンがアップロードされます。
import os
from box_sdk_gen import BoxAPIError, PreflightFileUploadCheckParent
def upload_chunked_file(client, file_path, folder_id):
file_size = os.path.getsize(file_path)
file_name = os.path.basename(file_path)
parent = PreflightFileUploadCheckParent(id=folder_id)
try:
client.uploads.preflight_file_upload_check(
name=file_name,
size=file_size,
parent=parent,
)
with open(file_path, "rb") as file_stream:
return client.chunked_uploads.upload_big_file(
file_stream, file_name, file_size, folder_id
)
except BoxAPIError as err:
if err.response_info.code != "item_name_in_use":
raise
file_id = err.response_info.context_info["conflicts"]["id"]
upload_session = (
client.chunked_uploads.create_file_upload_session_for_existing_file(
file_id, file_size, file_name
)
)
return upload_file_via_session(
client, file_path, file_size, upload_session
)
手動のアップロードセッション
アップロードを詳細に制御するには、自分でセッションを管理します。自分が渡したセッションについて、upload_file_via_sessionヘルパーによってすべてのパートがアップロードされ、セッションがコミットされます。
import base64
import hashlib
from io import BytesIO
from box_sdk_gen import UploadPart
def upload_file_via_session(client, file_path, file_size, upload_session):
file_sha1 = hashlib.sha1()
parts: list[UploadPart] = []
upload_part_url = upload_session.session_endpoints.upload_part
commit_url = upload_session.session_endpoints.commit
with open(file_path, "rb") as file_stream:
part_size = upload_session.part_size
for part_num in range(upload_session.total_parts):
offset = part_num * part_size
remaining = file_size - offset
chunk_size = min(part_size, remaining)
chunk = file_stream.read(chunk_size)
part_sha1 = hashlib.sha1(chunk).digest()
digest = f"sha={base64.b64encode(part_sha1).decode()}"
content_range = (
f"bytes {offset}-{offset + len(chunk) - 1}/{file_size}"
)
uploaded_part = client.chunked_uploads.upload_file_part_by_url(
upload_part_url,
BytesIO(chunk),
digest,
content_range,
)
parts.append(uploaded_part.part)
file_sha1.update(chunk)
content_sha1 = file_sha1.digest()
digest = f"sha={base64.b64encode(content_sha1).decode()}"
files = client.chunked_uploads.create_file_upload_session_commit_by_url(
commit_url, parts, digest
)
return files.entries[0]
create_file_upload_sessionを使用して新しいファイルを作成します。同名のファイルがすでに存在する場合、upload_file_manualはcreate_file_upload_session_for_existing_fileにフォールバックして新しいバージョンをアップロードします。
import os
from box_sdk_gen import BoxAPIError, PreflightFileUploadCheckParent
def upload_file_manual(client, file_path, folder_id):
file_size = os.path.getsize(file_path)
file_name = os.path.basename(file_path)
parent = PreflightFileUploadCheckParent(id=folder_id)
try:
client.uploads.preflight_file_upload_check(
name=file_name,
size=file_size,
parent=parent,
)
upload_session = client.chunked_uploads.create_file_upload_session(
folder_id, file_size, file_name
)
except BoxAPIError as err:
if err.response_info.code != "item_name_in_use":
raise
file_id = err.response_info.context_info["conflicts"]["id"]
upload_session = (
client.chunked_uploads.create_file_upload_session_for_existing_file(
file_id, file_size, file_name
)
)
return upload_file_via_session(
client, file_path, file_size, upload_session
)
リソース
