· Nacho Coll · Tutorials · 13 分で読了
Python IPFS アップロード：コード例付き完全ガイド
IPFS.NINJA REST API を使用して Python から IPFS にファイルをアップロード。requests ライブラリの例を含む完全ガイド。
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}")ピン留めしたファイルの管理
アップロードしたファイルを取得および管理します：
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}")次のステップ
これで、Python から IPFS にファイルをアップロードするための完全なツールキットができました！これは以下をカバーしています：
- IPFS.NINJA API を使用した JSON とファイルのアップロード
- 複数ファイルのバッチ処理
- ファイル管理と既存コンテンツのピン留め
- 署名付きトークンによる安全なアップロード
- 本番環境対応のエラー処理
より高度な機能については、他のガイドを参照してください：
- API の詳細については完全な IPFS アップロード API チュートリアル
- 利点を理解するためのIPFS.NINJA vs Pinata の比較
- サービス比較のための最高の IPFS ピン留めサービス
IPFS.NINJA API は、独自のインフラストラクチャを実行する複雑さなしに、Python ベースの IPFS アプリケーションに必要なすべてを提供します。
ピン留めを始める準備はできましたか？ 無料アカウントを作成 — 50 ファイル、1 GB ストレージ、2 GB 帯域幅/月。クレジットカード不要。
