· Nacho Coll · Tutorials · 10 min read
Python IPFS Upload: Complete Guide with Code Examples
Upload files to IPFS from Python using the IPFS.NINJA REST API. Complete guide with requests library examples.
Python developers looking to upload files to IPFS don’t need to run a local IPFS node or deal with complex library dependencies. While IPFS.NINJA doesn’t provide a dedicated Python SDK, its REST API is perfectly suited for Python applications using the built-in requests library.
This guide shows you how to upload files, images, and JSON data to IPFS from Python, manage your pinned files, and leverage advanced features like signed upload tokens for secure client-side uploads.
Why Choose IPFS.NINJA for Python Development?
Before diving into code examples, let’s understand why IPFS.NINJA is an excellent choice for Python developers:
- No SDK dependencies: Use Python’s built-in
requestslibrary - RESTful API: Clean, predictable endpoints that follow HTTP standards
- JSON-native: Upload JSON data directly without file conversion
- Multiple authentication methods: API keys, signed tokens, or JWT
- Built-in analytics: Track upload performance and usage
- Custom gateways: Branded IPFS access for your applications
If you’re new to IPFS pinning concepts, check out our comprehensive guide on what IPFS pinning is and how to upload files to IPFS.
Setting Up Your Python Environment
First, ensure you have Python 3.6+ and install the required libraries:
pip install requestsFor image processing examples later in this guide:
pip install Pillow # For image handlingAuthentication Setup
All IPFS.NINJA API calls require authentication. The simplest method for Python applications is using API keys with the X-Api-Key header.
import requests
import json
import base64
# Your IPFS.NINJA API key (format: bws_ + 32 hex characters)
API_KEY = "bws_1234567890abcdef1234567890abcdef"
BASE_URL = "https://api.ipfs.ninja"
# Default headers for all requests
headers = {
"X-Api-Key": API_KEY,
"Content-Type": "application/json"
}Get your API key from the IPFS.NINJA dashboard after creating your account.
Uploading JSON Data to IPFS
One of IPFS.NINJA’s unique features is native JSON support. You don’t need to convert JSON to files—upload objects directly:
def upload_json_to_ipfs(data, description=None, metadata=None):
"""
Upload JSON data directly to IPFS
Args:
data (dict): The JSON data to upload
description (str, optional): Description for the file
metadata (dict, optional): Additional metadata
Returns:
dict: Response containing CID and access URLs
"""
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}")
# Example: Upload a user profile
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}")Uploading Files with Base64 Encoding
For binary files like images, documents, or any file type, use base64 encoding:
def upload_file_to_ipfs(file_path, description=None, metadata=None):
"""
Upload a file to IPFS using base64 encoding
Args:
file_path (str): Path to the file to upload
description (str, optional): Description for the file
metadata (dict, optional): Additional metadata
Returns:
dict: Response containing CID and access URLs
"""
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}")
# Example: Upload an image
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}")Batch Upload Function
For uploading multiple files efficiently:
import os
from pathlib import Path
def batch_upload_files(directory_path, file_extensions=None):
"""
Upload multiple files from a directory
Args:
directory_path (str): Directory containing files to upload
file_extensions (list, optional): Filter by file extensions (e.g., ['.png', '.jpg'])
Returns:
list: Results from each upload
"""
results = []
directory = Path(directory_path)
if not directory.exists():
raise Exception(f"Directory not found: {directory_path}")
# Get all files or filter by extensions
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
# Example usage
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}")Managing Your Pinned Files
Retrieve and manage your uploaded files:
def list_pinned_files(limit=50, offset=0):
"""
List your pinned files
Args:
limit (int): Number of files to retrieve (max 100)
offset (int): Number of files to skip
Returns:
dict: Response containing files list and pagination
"""
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):
"""
Get details for a specific file
Args:
cid (str): The Content Identifier of the file
Returns:
dict: File details
"""
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}")
# Example: List recent uploads
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}")Pin Existing IPFS Content
Pin content that already exists on IPFS:
def pin_existing_cid(cid, description=None):
"""
Pin existing IPFS content by CID
Args:
cid (str): The Content Identifier to pin
description (str, optional): Description for the pinned content
Returns:
dict: Response with pin status
"""
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}")
# Example: Pin popular NFT metadata
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}")Using Signed Upload Tokens
For secure client-side uploads or temporary access, use signed tokens:
def create_upload_token(expires_in_hours=24, max_uses=None):
"""
Create a signed upload token for temporary uploads
Args:
expires_in_hours (int): Token expiration time in hours
max_uses (int, optional): Maximum number of uses (unlimited if None)
Returns:
dict: Token details including the signed token string
"""
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):
"""
Upload using a signed token instead of API key
Args:
data: Content to upload (dict for JSON, base64 string for files)
token (str): Signed upload token
description (str, optional): Description for the upload
Returns:
dict: Upload response
"""
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}")
# Example: Create token for client-side uploads
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')}")
# Use the token for an upload
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}")Complete Example: NFT Metadata Uploader
Here’s a practical example that combines multiple concepts for NFT metadata management:
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=""):
"""Upload NFT image and return 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'] # Return 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):
"""Upload complete NFT metadata"""
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):
"""Upload image and metadata for a complete NFT"""
print(f"Creating NFT: {name}")
# Step 1: Upload image
print(" 1. Uploading image...")
image_ipfs_url = self.upload_nft_image(image_path, name, description)
print(f" Image IPFS URL: {image_ipfs_url}")
# Step 2: Upload metadata
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']
}
# Example usage
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}")Error Handling and Best Practices
Implement robust error handling for production applications:
import time
from functools import wraps
def retry_on_failure(max_retries=3, delay=1):
"""Decorator to retry failed uploads"""
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):
"""Upload with automatic retry on failure"""
return upload_file_to_ipfs(file_path, description)
def validate_api_response(response, operation="operation"):
"""Validate API response and provide detailed error info"""
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}")Next Steps
You now have a complete toolkit for uploading files to IPFS from Python! This covers:
- JSON and file uploads with the IPFS.NINJA API
- Batch processing for multiple files
- File management and pinning existing content
- Secure uploads with signed tokens
- Production-ready error handling
For more advanced features, explore our other guides:
- Complete IPFS Upload API Tutorial for API details
- IPFS.NINJA vs Pinata comparison to understand the advantages
- Best IPFS Pinning Services for service comparisons
The IPFS.NINJA API provides everything you need for Python-based IPFS applications without the complexity of running your own infrastructure.
Ready to start pinning? Create a free account — 500 files, 1 GB storage, dedicated gateway. No credit card required.
