Python IPFS アップロード：コード例付き完全ガイド
IPFS.NINJA REST APIを使ってPythonからファイル・画像・JSONをIPFSにアップロード。専用SDKは不要—requestsライブラリだけでOK。動作確認済みのコード例付き完全ガイド。
要約
- 標準のrequestsライブラリだけを使って、Pythonからファイル、画像、JSONをIPFSにアップロードできます。
- 専用のPython SDKは不要です—REST APIはHTTPに対応したどの言語からでも同じように動作します。
- Pythonから署名付きアップロードトークンを生成すれば、ブラウザやモバイルのクライアントがAPIキーなしでアップロードできます。
- すべての呼び出しは単一のX-Api-Keyヘッダーで認証されます—OAuthフローもSDKのインストールも不要です。
IPFS にファイルをアップロードしたい Python 開発者は、ローカルの IPFS ノードを実行したり、複雑なライブラリ依存関係を扱う必要はありません。IPFS.NINJA は専用の Python SDK を提供していませんが、その REST API は組み込みの requests ライブラリを使用する Python アプリケーションに完全に適しています。
このガイドでは、Python からファイル、画像、JSON データを IPFS にアップロードし、ピン留めしたファイルを管理し、安全なクライアント側アップロードのための署名済みアップロードトークンなどの高度な機能を活用する方法を示します。
なぜ Python 開発に IPFS.NINJA を選ぶのか？#
コード例に入る前に、IPFS.NINJA が Python 開発者にとって優れた選択肢である理由を理解しましょう：
- SDK 依存関係なし：Python の組み込み
requestsライブラリを使用 - RESTful API：HTTP 標準に従ったクリーンで予測可能なエンドポイント
- JSON ネイティブ：ファイル変換なしで JSON データを直接アップロード
- 複数の認証方法：API キー、署名トークン、または JWT
- 組み込み分析：アップロードのパフォーマンスと使用状況を追跡
- カスタムゲートウェイ：アプリケーション用のブランド化された IPFS アクセス
IPFS ピン留めの概念に馴染みがない場合は、IPFS ピン留めとは何かの包括的なガイドとIPFS にファイルをアップロードする方法をご覧ください。
Python 環境のセットアップ#
まず、Python 3.6+ があることを確認し、必要なライブラリをインストールします：
pip install requestsこのガイドの後半で画像処理の例について：
pip install Pillow # 画像処理用認証セットアップ#
すべての IPFS.NINJA API 呼び出しには認証が必要です。Python アプリケーションの最も簡単な方法は、X-Api-Key ヘッダーで API キーを使用することです。
import requests
import json
import base64
# あなたの IPFS.NINJA API キー（形式：bws_ + 32 文字の 16 進数）
API_KEY = "bws_1234567890abcdef1234567890abcdef"
BASE_URL = "https://api.ipfs.ninja"
# すべてのリクエストのデフォルトヘッダー
headers = {
"X-Api-Key": API_KEY,
"Content-Type": "application/json"
}アカウントを作成した後、IPFS.NINJA ダッシュボードから API キーを取得します。
JSON データを IPFS にアップロード#
IPFS.NINJA のユニークな機能の 1 つは、ネイティブの JSON サポートです。JSON をファイルに変換する必要はなく、オブジェクトを直接アップロードできます：
def upload_json_to_ipfs(data, description=None, metadata=None):
"""
JSON データを直接 IPFS にアップロード
引数:
data (dict): アップロードする JSON データ
description (str, オプション): ファイルの説明
metadata (dict, オプション): 追加のメタデータ
戻り値:
dict: CID とアクセス URL を含むレスポンス
"""
payload = {
"content": data,
"description": description,
"metadata": metadata
}
response = requests.post(
f"{BASE_URL}/upload/new",
headers=headers,
json=payload
)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"Upload failed: {response.status_code} - {response.text}")
# 例：ユーザープロファイルをアップロード
user_profile = {
"name": "Alice Smith",
"bio": "Python developer and IPFS enthusiast",
"website": "https://alice.dev",
"social": {
"github": "alicesmith",
"twitter": "@alice_dev"
}
}
try:
result = upload_json_to_ipfs(
data=user_profile,
description="User profile for Alice Smith",
metadata={"type": "profile", "version": "1.0"}
)
print(f"✅ Upload successful!")
print(f"CID: {result['cid']}")
print(f"Size: {result['sizeMB']} MB")
print(f"IPFS URL: {result['uris']['ipfs']}")
print(f"Gateway URL: {result['uris']['url']}")
except Exception as e:
print(f"❌ Upload failed: {e}")Base64 エンコーディングでファイルをアップロード#
画像、ドキュメント、その他のファイルタイプなどのバイナリファイルの場合は、base64 エンコーディングを使用します：
def upload_file_to_ipfs(file_path, description=None, metadata=None):
"""
base64 エンコーディングを使用して IPFS にファイルをアップロード
引数:
file_path (str): アップロードするファイルのパス
description (str, オプション): ファイルの説明
metadata (dict, オプション): 追加のメタデータ
戻り値:
dict: CID とアクセス URL を含むレスポンス
"""
try:
with open(file_path, "rb") as file:
file_content = base64.b64encode(file.read()).decode('utf-8')
payload = {
"content": file_content,
"description": description or f"File upload: {file_path}",
"metadata": metadata
}
response = requests.post(
f"{BASE_URL}/upload/new",
headers=headers,
json=payload
)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"Upload failed: {response.status_code} - {response.text}")
except FileNotFoundError:
raise Exception(f"File not found: {file_path}")
# 例：画像をアップロード
try:
result = upload_file_to_ipfs(
file_path="./logo.png",
description="Company logo",
metadata={
"type": "image",
"format": "png",
"purpose": "branding"
}
)
print(f"✅ Image uploaded successfully!")
print(f"CID: {result['cid']}")
print(f"View at: {result['uris']['url']}")
except Exception as e:
print(f"❌ Upload failed: {e}")バッチアップロード関数#
複数のファイルを効率的にアップロードする：
import os
from pathlib import Path
def batch_upload_files(directory_path, file_extensions=None):
"""
ディレクトリから複数のファイルをアップロード
引数:
directory_path (str): アップロードするファイルを含むディレクトリ
file_extensions (list, オプション): ファイル拡張子でフィルタ（例：['.png', '.jpg']）
戻り値:
list: 各アップロードの結果
"""
results = []
directory = Path(directory_path)
if not directory.exists():
raise Exception(f"Directory not found: {directory_path}")
# すべてのファイルを取得するか、拡張子でフィルタ
if file_extensions:
files = []
for ext in file_extensions:
files.extend(directory.glob(f"*{ext}"))
else:
files = [f for f in directory.iterdir() if f.is_file()]
print(f"Found {len(files)} files to upload...")
for file_path in files:
try:
print(f"Uploading {file_path.name}...")
result = upload_file_to_ipfs(
file_path=str(file_path),
description=f"Batch upload: {file_path.name}",
metadata={
"batch_upload": True,
"original_name": file_path.name,
"file_size": file_path.stat().st_size
}
)
results.append({
"filename": file_path.name,
"success": True,
"cid": result['cid'],
"url": result['uris']['url']
})
except Exception as e:
results.append({
"filename": file_path.name,
"success": False,
"error": str(e)
})
print(f"Failed to upload {file_path.name}: {e}")
return results
# 使用例
try:
results = batch_upload_files("./images", file_extensions=['.png', '.jpg', '.jpeg'])
successful = [r for r in results if r['success']]
failed = [r for r in results if not r['success']]
print(f"\n✅ Successfully uploaded: {len(successful)} files")
print(f"❌ Failed uploads: {len(failed)} files")
for result in successful:
print(f" - {result['filename']}: {result['url']}")
except Exception as e:
print(f"Batch upload error: {e}")拡張子フィルタリングと失敗の追跡#
batch_upload_files 関数はすでに file_extensions 引数で拡張子によるフィルタリングをサポートしていますが、実運用では許可リスト（アップロードを許可する拡張子）と拒否リスト（明示的に除外する拡張子）の両方を扱えると便利です。テンポラリファイルやシステムファイル（.tmp、.DS_Store など）を誤ってアップロードしてしまう事故を防げます。また、失敗したファイルをただ数えるだけでなく、失敗理由ごとに集計しておくと、ネットワークタイムアウトが多いのか、ファイルサイズ超過が多いのかといった傾向を把握でき、再試行ロジックの改善につながります。
from collections import Counter
def batch_upload_with_tracking(directory_path, allowed_extensions=None, excluded_extensions=None):
"""
拡張子フィルタリングと失敗理由の集計を行うバッチアップロード
引数:
directory_path (str): アップロードするファイルを含むディレクトリ
allowed_extensions (list, オプション): 許可する拡張子（許可リスト）
excluded_extensions (list, オプション): 除外する拡張子（拒否リスト）
"""
results = batch_upload_files(
directory_path,
file_extensions=allowed_extensions
)
if excluded_extensions:
results = [
r for r in results
if not any(r["filename"].endswith(ext) for ext in excluded_extensions)
]
failure_reasons = Counter(
r["error"].split(":")[0] for r in results if not r["success"]
)
print("失敗理由の内訳:")
for reason, count in failure_reasons.most_common():
print(f" - {reason}: {count} 件")
return resultsこのように許可リストと拒否リストを組み合わせることで、意図しないファイルタイプの混入を防ぎつつ、失敗パターンを継続的に監視できます。CI/CD パイプラインで定期的にバッチアップロードを実行している場合は、この集計結果をSlack通知やダッシュボードに連携すると、障害の早期発見に役立ちます。
ピン留めしたファイルの管理#
アップロードしたファイルを取得および管理します：
def list_pinned_files(limit=50, offset=0):
"""
ピン留めしたファイルをリスト
引数:
limit (int): 取得するファイル数（最大 100）
offset (int): スキップするファイル数
戻り値:
dict: ファイルリストとページネーションを含むレスポンス
"""
params = {
"limit": limit,
"offset": offset
}
response = requests.get(
f"{BASE_URL}/files",
headers=headers,
params=params
)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"Failed to list files: {response.status_code} - {response.text}")
def get_file_details(cid):
"""
特定のファイルの詳細を取得
引数:
cid (str): ファイルのコンテンツ識別子
戻り値:
dict: ファイルの詳細
"""
response = requests.get(
f"{BASE_URL}/files/{cid}",
headers=headers
)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"Failed to get file details: {response.status_code} - {response.text}")
# 例：最近のアップロードをリスト
try:
files = list_pinned_files(limit=10)
print(f"Total files: {files.get('total', 0)}")
print("\nRecent uploads:")
for file_info in files.get('files', []):
print(f" - {file_info['cid'][:12]}... | {file_info['sizeMB']} MB | {file_info['description']}")
except Exception as e:
print(f"Failed to list files: {e}")既存の IPFS コンテンツをピン留め#
すでに IPFS に存在するコンテンツをピン留めします：
def pin_existing_cid(cid, description=None):
"""
CID で既存の IPFS コンテンツをピン留め
引数:
cid (str): ピン留めするコンテンツ識別子
description (str, オプション): ピン留めされたコンテンツの説明
戻り値:
dict: ピンステータスを含むレスポンス
"""
payload = {
"cid": cid,
"description": description or f"Pinned existing content: {cid}"
}
response = requests.post(
f"{BASE_URL}/pin",
headers=headers,
json=payload
)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"Pin failed: {response.status_code} - {response.text}")
# 例：人気のある NFT メタデータをピン留め
try:
result = pin_existing_cid(
cid="QmYwAPJzv5CZsnA625s3Xf2nemtYgPpHdWEz79ojWnPbdG",
description="Popular NFT metadata backup"
)
print(f"✅ Successfully pinned existing content!")
print(f"CID: {result['cid']}")
except Exception as e:
print(f"❌ Pin failed: {e}")署名付きアップロードトークンの使用#
安全なクライアント側アップロードまたは一時的なアクセスのために、署名付きトークンを使用します：
def create_upload_token(expires_in_hours=24, max_uses=None):
"""
一時的なアップロード用の署名付きアップロードトークンを作成
引数:
expires_in_hours (int): トークンの有効期限（時間）
max_uses (int, オプション): 最大使用回数（None の場合は無制限）
戻り値:
dict: 署名されたトークン文字列を含むトークンの詳細
"""
from datetime import datetime, timedelta
expiry_time = datetime.utcnow() + timedelta(hours=expires_in_hours)
payload = {
"expiry": expiry_time.isoformat() + "Z",
"maxUses": max_uses
}
response = requests.post(
f"{BASE_URL}/upload-tokens",
headers=headers,
json=payload
)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"Token creation failed: {response.status_code} - {response.text}")
def upload_with_token(data, token, description=None):
"""
API キーではなく署名付きトークンを使用してアップロード
引数:
data: アップロードするコンテンツ（JSON 用の dict、ファイル用の base64 文字列）
token (str): 署名付きアップロードトークン
description (str, オプション): アップロードの説明
戻り値:
dict: アップロードレスポンス
"""
token_headers = {
"Authorization": f"Signed {token}",
"Content-Type": "application/json"
}
payload = {
"content": data,
"description": description
}
response = requests.post(
f"{BASE_URL}/upload/new",
headers=token_headers,
json=payload
)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"Token upload failed: {response.status_code} - {response.text}")
# 例：クライアント側アップロード用のトークンを作成
try:
token_info = create_upload_token(expires_in_hours=1, max_uses=10)
upload_token = token_info['token']
print(f"Created upload token: {upload_token[:20]}...")
print(f"Expires: {token_info['expiry']}")
print(f"Max uses: {token_info.get('maxUses', 'unlimited')}")
# アップロードにトークンを使用
test_data = {"message": "Hello from signed token!"}
result = upload_with_token(test_data, upload_token, "Token test upload")
print(f"✅ Token upload successful: {result['cid']}")
except Exception as e:
print(f"❌ Token operation failed: {e}")完全な例：NFT メタデータアップローダー#
NFT メタデータ管理のために複数の概念を組み合わせた実用的な例です：
class NFTMetadataUploader:
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "https://api.ipfs.ninja"
self.headers = {
"X-Api-Key": api_key,
"Content-Type": "application/json"
}
def upload_nft_image(self, image_path, name, description=""):
"""NFT 画像をアップロードして IPFS URL を返す"""
try:
with open(image_path, "rb") as file:
image_content = base64.b64encode(file.read()).decode('utf-8')
payload = {
"content": image_content,
"description": f"NFT Image: {name}",
"metadata": {
"type": "nft_image",
"name": name,
"description": description
}
}
response = requests.post(
f"{self.base_url}/upload/new",
headers=self.headers,
json=payload
)
if response.status_code == 200:
result = response.json()
return result['uris']['ipfs'] # IPFS URI を返す
else:
raise Exception(f"Image upload failed: {response.text}")
except Exception as e:
raise Exception(f"Failed to upload image {image_path}: {e}")
def upload_nft_metadata(self, name, description, image_ipfs_url, attributes=None, external_url=None):
"""完全な NFT メタデータをアップロード"""
metadata = {
"name": name,
"description": description,
"image": image_ipfs_url,
}
if attributes:
metadata["attributes"] = attributes
if external_url:
metadata["external_url"] = external_url
payload = {
"content": metadata,
"description": f"NFT Metadata: {name}",
"metadata": {
"type": "nft_metadata",
"standard": "ERC-721"
}
}
response = requests.post(
f"{self.base_url}/upload/new",
headers=self.headers,
json=payload
)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"Metadata upload failed: {response.text}")
def create_complete_nft(self, image_path, name, description, attributes=None, external_url=None):
"""完全な NFT のために画像とメタデータをアップロード"""
print(f"Creating NFT: {name}")
# ステップ 1：画像をアップロード
print(" 1. Uploading image...")
image_ipfs_url = self.upload_nft_image(image_path, name, description)
print(f" Image IPFS URL: {image_ipfs_url}")
# ステップ 2：メタデータをアップロード
print(" 2. Uploading metadata...")
metadata_result = self.upload_nft_metadata(
name=name,
description=description,
image_ipfs_url=image_ipfs_url,
attributes=attributes,
external_url=external_url
)
print(f" Metadata CID: {metadata_result['cid']}")
print(f" Metadata URL: {metadata_result['uris']['url']}")
return {
"image_ipfs": image_ipfs_url,
"metadata_cid": metadata_result['cid'],
"metadata_url": metadata_result['uris']['url'],
"metadata_ipfs": metadata_result['uris']['ipfs']
}
# 使用例
if __name__ == "__main__":
uploader = NFTMetadataUploader("bws_1234567890abcdef1234567890abcdef")
try:
nft_result = uploader.create_complete_nft(
image_path="./nft_artwork.png",
name="Cosmic Cat #001",
description="A rare cosmic cat exploring the digital universe",
attributes=[
{"trait_type": "Background", "value": "Space"},
{"trait_type": "Species", "value": "Cosmic Cat"},
{"trait_type": "Rarity", "value": "Legendary"},
{"trait_type": "Power Level", "value": 95}
],
external_url="https://myproject.com/nft/1"
)
print("\n🎉 NFT created successfully!")
print(f"Use this metadata URI in your smart contract: {nft_result['metadata_ipfs']}")
except Exception as e:
print(f"❌ NFT creation failed: {e}")エラー処理とベストプラクティス#
本番アプリケーションのための堅牢なエラー処理を実装します：
import time
from functools import wraps
def retry_on_failure(max_retries=3, delay=1):
"""失敗したアップロードを再試行するためのデコレータ"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_retries - 1:
raise e
print(f"Attempt {attempt + 1} failed: {e}")
print(f"Retrying in {delay} seconds...")
time.sleep(delay)
return None
return wrapper
return decorator
@retry_on_failure(max_retries=3, delay=2)
def robust_upload(file_path, description=None):
"""失敗時に自動再試行でアップロード"""
return upload_file_to_ipfs(file_path, description)
def validate_api_response(response, operation="operation"):
"""API レスポンスを検証し、詳細なエラー情報を提供"""
if response.status_code == 200:
return response.json()
elif response.status_code == 401:
raise Exception(f"Authentication failed for {operation}. Check your API key.")
elif response.status_code == 413:
raise Exception(f"File too large for {operation}. Check your plan limits.")
elif response.status_code == 429:
raise Exception(f"Rate limit exceeded for {operation}. Please wait and retry.")
else:
raise Exception(f"{operation} failed: {response.status_code} - {response.text}")レート制限に対する指数バックオフ#
上記の retry_on_failure デコレータは固定の待機時間で再試行しますが、本番環境では HTTP 429（レート制限超過）に対して指数バックオフを実装することが重要です。固定間隔での再試行は、レート制限が解除される前に何度もリクエストを送ってしまい、かえって制限期間を延長させる可能性があります。以下は、試行ごとに待機時間を倍増させ、ジッター（ランダムな揺らぎ）を加えることで複数クライアントが同時に再試行するのを防ぐ実装です。
import random
def exponential_backoff_upload(file_path, description=None, max_retries=5, base_delay=1):
"""
レート制限（HTTP 429）に対して指数バックオフで再試行する
引数:
file_path (str): アップロードするファイルのパス
description (str, オプション): ファイルの説明
max_retries (int): 最大再試行回数
base_delay (int): 最初の待機時間（秒）
"""
for attempt in range(max_retries):
try:
return upload_file_to_ipfs(file_path, description)
except Exception as e:
if "429" not in str(e) or attempt == max_retries - 1:
raise
wait_time = base_delay * (2 ** attempt) + random.uniform(0, 1)
print(f"レート制限を検出。{wait_time:.1f}秒待機後に再試行 {attempt + 1}/{max_retries}...")
time.sleep(wait_time)
return Noneこのパターンは、大量のファイルをバッチアップロードする際に特に重要です。プランごとのレート制限に近づくと 429 が返ってきますが、指数バックオフを使うことでリクエストの再試行間隔が自然に広がり、制限が解除されるまで API に余計な負荷をかけません。
アップロードログの記録#
本番運用では、どのファイルがいつアップロードされ、成功したか失敗したかを記録しておくことが不可欠です。Python の標準 logging モジュールを使えば、コンソール出力だけに頼らず、ファイルへの永続的なログ記録が可能になり、後から障害調査や監査を行う際に役立ちます。
import logging
logging.basicConfig(
filename="ipfs_uploads.log",
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s"
)
def logged_upload(file_path, description=None):
"""アップロードの結果をログファイルに記録する"""
try:
result = upload_file_to_ipfs(file_path, description)
logging.info(f"SUCCESS file={file_path} cid={result['cid']} size={result.get('sizeMB')}MB")
return result
except Exception as e:
logging.error(f"FAILED file={file_path} error={e}")
raiseこのログは、アップロードの失敗パターンを後から分析したり、特定の CID がいつ作成されたかを監査したりする際に役立ちます。CI/CD パイプラインでバッチアップロードを自動化している場合は、ログファイルをアーティファクトとして保存しておくと、障害調査が格段に楽になります。
次のステップ#
これで、Python から IPFS にファイルをアップロードするための完全なツールキットができました！これは以下をカバーしています：
- IPFS.NINJA API を使用した JSON とファイルのアップロード
- 複数ファイルのバッチ処理
- ファイル管理と既存コンテンツのピン留め
- 署名付きトークンによる安全なアップロード
- 本番環境対応のエラー処理
より高度な機能については、他のガイドを参照してください：
- API の詳細については完全な IPFS アップロード API チュートリアル
- 利点を理解するためのIPFS.NINJA vs Pinata の比較
- サービス比較のための最高の IPFS ピン留めサービス
IPFS.NINJA API は、独自のインフラストラクチャを実行する複雑さなしに、Python ベースの IPFS アプリケーションに必要なすべてを提供します。
ピン留めを始める準備はできましたか？ 無料アカウントを作成 — 50 ファイル、1 GB ストレージ、2 GB 帯域幅/月。クレジットカード不要。
See also: IPFS CIDs explained.
著者について
Nacho Coll
Founder & Engineer at IPFS.NINJA
Nacho founded IPFS.NINJA to make content-addressed storage feel as simple as an S3 PUT — a single API call, a permanent CID, no wallets or peer discovery to reason about. Writes about IPFS internals, decentralized storage patterns, and the pinning-service landscape from the operator side of the wire.
