· Nacho Coll · Tutorials · 10 min de lectura
Subir a IPFS con Python: Guía Completa con Ejemplos de Código
Sube archivos a IPFS desde Python usando la API REST de IPFS.NINJA. Guía completa con ejemplos de la librería requests.
Los desarrolladores de Python que buscan subir archivos a IPFS no necesitan ejecutar un nodo IPFS local o lidiar con dependencias de librerías complejas. Aunque IPFS.NINJA no proporciona un SDK dedicado de Python, su API REST está perfectamente adaptada para aplicaciones Python usando la librería integrada requests.
Esta guía te muestra cómo subir archivos, imágenes y datos JSON a IPFS desde Python, gestionar tus archivos pineados, y aprovechar funciones avanzadas como tokens de subida firmados para subidas seguras del lado del cliente.
¿Por Qué Elegir IPFS.NINJA para Desarrollo en Python?
Antes de adentrarnos en los ejemplos de código, entendamos por qué IPFS.NINJA es una excelente opción para los desarrolladores de Python:
- Sin dependencias de SDK: Usa la librería integrada
requestsde Python - API RESTful: Endpoints limpios y predecibles que siguen los estándares HTTP
- JSON-nativo: Sube datos JSON directamente sin conversión a archivo
- Múltiples métodos de autenticación: Claves API, tokens firmados o JWT
- Análisis integrados: Rastrea el rendimiento y uso de subidas
- Gateways personalizados: Acceso IPFS de marca para tus aplicaciones
Si eres nuevo en los conceptos de pinning de IPFS, consulta nuestra guía completa sobre qué es el pinning de IPFS y cómo subir archivos a IPFS.
Configurando tu Entorno Python
Primero, asegúrate de tener Python 3.6+ e instala las librerías necesarias:
pip install requestsPara ejemplos de procesamiento de imágenes más adelante en esta guía:
pip install Pillow # Para manejo de imágenesConfiguración de Autenticación
Todas las llamadas a la API de IPFS.NINJA requieren autenticación. El método más simple para aplicaciones Python es usar claves API con el header X-Api-Key.
import requests
import json
import base64
# Tu clave API de IPFS.NINJA (formato: bws_ + 32 caracteres hex)
API_KEY = "bws_1234567890abcdef1234567890abcdef"
BASE_URL = "https://api.ipfs.ninja"
# Headers por defecto para todas las peticiones
headers = {
"X-Api-Key": API_KEY,
"Content-Type": "application/json"
}Obtén tu clave API desde el panel de IPFS.NINJA después de crear tu cuenta.
Subiendo Datos JSON a IPFS
Una de las características únicas de IPFS.NINJA es el soporte nativo de JSON. No necesitas convertir JSON a archivos — sube objetos directamente:
def upload_json_to_ipfs(data, description=None, metadata=None):
"""
Subir datos JSON directamente a IPFS
Args:
data (dict): Los datos JSON a subir
description (str, opcional): Descripción para el archivo
metadata (dict, opcional): Metadatos adicionales
Returns:
dict: Respuesta conteniendo CID y URLs de acceso
"""
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}")
# Ejemplo: Subir un perfil de usuario
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}")Subiendo Archivos con Codificación Base64
Para archivos binarios como imágenes, documentos o cualquier tipo de archivo, usa codificación base64:
def upload_file_to_ipfs(file_path, description=None, metadata=None):
"""
Subir un archivo a IPFS usando codificación base64
Args:
file_path (str): Ruta al archivo a subir
description (str, opcional): Descripción para el archivo
metadata (dict, opcional): Metadatos adicionales
Returns:
dict: Respuesta conteniendo CID y URLs de acceso
"""
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}")
# Ejemplo: Subir una imagen
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}")Función de Subida por Lotes
Para subir múltiples archivos eficientemente:
import os
from pathlib import Path
def batch_upload_files(directory_path, file_extensions=None):
"""
Subir múltiples archivos desde un directorio
Args:
directory_path (str): Directorio conteniendo archivos a subir
file_extensions (list, opcional): Filtrar por extensiones de archivo (ej., ['.png', '.jpg'])
Returns:
list: Resultados de cada subida
"""
results = []
directory = Path(directory_path)
if not directory.exists():
raise Exception(f"Directory not found: {directory_path}")
# Obtener todos los archivos o filtrar por extensiones
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
# Ejemplo de uso
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}")Gestionando tus Archivos Pineados
Recupera y gestiona tus archivos subidos:
def list_pinned_files(limit=50, offset=0):
"""
Listar tus archivos pineados
Args:
limit (int): Número de archivos a recuperar (máx 100)
offset (int): Número de archivos a omitir
Returns:
dict: Respuesta conteniendo lista de archivos y paginación
"""
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):
"""
Obtener detalles para un archivo específico
Args:
cid (str): El Identificador de Contenido del archivo
Returns:
dict: Detalles del archivo
"""
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}")
# Ejemplo: Listar subidas recientes
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}")Pinear Contenido IPFS Existente
Pinea contenido que ya existe en IPFS:
def pin_existing_cid(cid, description=None):
"""
Pinear contenido IPFS existente por CID
Args:
cid (str): El Identificador de Contenido a pinear
description (str, opcional): Descripción para el contenido pineado
Returns:
dict: Respuesta con estado del pin
"""
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}")
# Ejemplo: Pinear metadata NFT popular
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}")Usando Tokens de Subida Firmados
Para subidas seguras del lado del cliente o acceso temporal, usa tokens firmados:
def create_upload_token(expires_in_hours=24, max_uses=None):
"""
Crear un token de subida firmado para subidas temporales
Args:
expires_in_hours (int): Tiempo de expiración del token en horas
max_uses (int, opcional): Número máximo de usos (ilimitado si es None)
Returns:
dict: Detalles del token incluyendo la cadena del token firmado
"""
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):
"""
Subir usando un token firmado en lugar de clave API
Args:
data: Contenido a subir (dict para JSON, cadena base64 para archivos)
token (str): Token de subida firmado
description (str, opcional): Descripción para la subida
Returns:
dict: Respuesta de subida
"""
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}")
# Ejemplo: Crear token para subidas del lado del cliente
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')}")
# Usar el token para una subida
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}")Ejemplo Completo: Cargador de Metadata NFT
Aquí hay un ejemplo práctico que combina múltiples conceptos para la gestión de metadata 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=""):
"""Subir imagen NFT y devolver URL IPFS"""
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'] # Devolver URI IPFS
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):
"""Subir metadata NFT completa"""
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):
"""Subir imagen y metadata para un NFT completo"""
print(f"Creating NFT: {name}")
# Paso 1: Subir imagen
print(" 1. Uploading image...")
image_ipfs_url = self.upload_nft_image(image_path, name, description)
print(f" Image IPFS URL: {image_ipfs_url}")
# Paso 2: Subir 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']
}
# Ejemplo de uso
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}")Manejo de Errores y Mejores Prácticas
Implementa un manejo de errores robusto para aplicaciones de producción:
import time
from functools import wraps
def retry_on_failure(max_retries=3, delay=1):
"""Decorador para reintentar subidas fallidas"""
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):
"""Subir con reintento automático en caso de fallo"""
return upload_file_to_ipfs(file_path, description)
def validate_api_response(response, operation="operation"):
"""Validar respuesta API y proporcionar info de error detallada"""
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}")Próximos Pasos
¡Ahora tienes un kit de herramientas completo para subir archivos a IPFS desde Python! Esto cubre:
- Subidas JSON y de archivos con la API de IPFS.NINJA
- Procesamiento por lotes para múltiples archivos
- Gestión de archivos y pinning de contenido existente
- Subidas seguras con tokens firmados
- Manejo de errores listo para producción
Para funciones más avanzadas, explora nuestras otras guías:
- Tutorial Completo de la API de Subida IPFS para detalles de la API
- Comparación IPFS.NINJA vs Pinata para entender las ventajas
- Mejores Servicios de Pinning IPFS para comparaciones de servicios
La API de IPFS.NINJA proporciona todo lo que necesitas para aplicaciones IPFS basadas en Python sin la complejidad de ejecutar tu propia infraestructura.
¿Listo para empezar a pinear? Crea una cuenta gratuita — 50 archivos, 1 GB de almacenamiento, 2 GB de ancho de banda/mes. Sin tarjeta de crédito.
