Volver a Inicio
Ejemplo Python
Integra la API de ReciGo en tu aplicación Python usando requests.
Instalación
Instala la librería requests:
pip install requestsCliente de API
Crea un cliente reutilizable para interactuar con la API de ReciGo:
recigo_client.py
import os
import requests
from typing import Dict, Any, Optional, List
class ReciGoClient:
"""Cliente para la API de ReciGo Partner."""
BASE_URL = "https://api.recigo.es/v1"
def __init__(self, api_key: Optional[str] = None):
"""
Inicializa el cliente de ReciGo.
Args:
api_key: Tu API key de ReciGo. Si no se proporciona,
se busca en la variable de entorno RECIGO_API_KEY.
"""
self.api_key = api_key or os.environ.get("RECIGO_API_KEY")
if not self.api_key:
raise ValueError(
"API key es requerida. Proporciona api_key o "
"define RECIGO_API_KEY en variables de entorno."
)
self.session = requests.Session()
self.session.headers.update({
"X-API-Key": self.api_key,
"Content-Type": "application/json",
})
def _request(
self,
method: str,
endpoint: str,
data: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""Realiza una petición HTTP a la API."""
url = f"{self.BASE_URL}{endpoint}"
response = self.session.request(
method=method,
url=url,
json=data
)
# Manejo de errores
if not response.ok:
try:
error_data = response.json()
raise Exception(
f"ReciGo API Error ({response.status_code}): "
f"{error_data.get('message', 'Error desconocido')}"
)
except ValueError:
raise Exception(
f"ReciGo API Error ({response.status_code}): "
f"{response.text}"
)
return response.json()
def estimate_price(
self,
size: str,
pickup_address: str,
dropoff_address: str,
pickup_floor: Optional[int] = None,
pickup_has_elevator: Optional[bool] = None,
dropoff_floor: Optional[int] = None,
dropoff_has_elevator: Optional[bool] = None,
) -> Dict[str, Any]:
"""
Estima el precio de un trabajo.
Args:
size: Tamaño del objeto ('S', 'M', 'L', 'XL')
pickup_address: Dirección de recogida
dropoff_address: Dirección de entrega
pickup_floor: Piso de recogida (opcional)
pickup_has_elevator: Si hay ascensor en recogida (opcional)
dropoff_floor: Piso de entrega (opcional)
dropoff_has_elevator: Si hay ascensor en entrega (opcional)
Returns:
Diccionario con la estimación de precio
"""
data = {
"size": size,
"pickup": {"address": pickup_address},
"dropoff": {"address": dropoff_address},
}
if pickup_floor is not None:
data["pickup"]["floor"] = pickup_floor
if pickup_has_elevator is not None:
data["pickup"]["has_elevator"] = pickup_has_elevator
if dropoff_floor is not None:
data["dropoff"]["floor"] = dropoff_floor
if dropoff_has_elevator is not None:
data["dropoff"]["has_elevator"] = dropoff_has_elevator
return self._request("POST", "/pricing/estimate", data)
def create_job(
self,
title: str,
description: str,
size: str,
pickup_address: str,
dropoff_address: str,
customer_phone: str,
external_reference: Optional[str] = None,
photos: Optional[List[str]] = None,
pickup_floor: Optional[int] = None,
pickup_has_elevator: Optional[bool] = None,
dropoff_floor: Optional[int] = None,
dropoff_has_elevator: Optional[bool] = None,
) -> Dict[str, Any]:
"""
Crea un nuevo trabajo.
Args:
title: Título del trabajo
description: Descripción detallada
size: Tamaño del objeto ('S', 'M', 'L', 'XL')
pickup_address: Dirección de recogida
dropoff_address: Dirección de entrega
customer_phone: Teléfono del cliente (formato internacional, ej: +34612345678)
external_reference: Tu referencia interna (opcional)
photos: Lista de URLs de fotos (opcional)
pickup_floor: Piso de recogida (opcional)
pickup_has_elevator: Si hay ascensor en recogida (opcional)
dropoff_floor: Piso de entrega (opcional)
dropoff_has_elevator: Si hay ascensor en entrega (opcional)
Returns:
Diccionario con los datos del trabajo creado
"""
data = {
"title": title,
"description": description,
"size": size,
"pickup": {"address": pickup_address},
"dropoff": {"address": dropoff_address},
"customer_phone": customer_phone,
}
if external_reference:
data["external_reference"] = external_reference
if photos:
data["photos"] = photos
if pickup_floor is not None:
data["pickup"]["floor"] = pickup_floor
if pickup_has_elevator is not None:
data["pickup"]["has_elevator"] = pickup_has_elevator
if dropoff_floor is not None:
data["dropoff"]["floor"] = dropoff_floor
if dropoff_has_elevator is not None:
data["dropoff"]["has_elevator"] = dropoff_has_elevator
return self._request("POST", "/jobs", data)
def get_job_status(self, job_id: str) -> Dict[str, Any]:
"""
Consulta el estado de un trabajo.
Args:
job_id: ID del trabajo
Returns:
Diccionario con el estado del trabajo
"""
return self._request("GET", f"/jobs/{job_id}")
# Instancia global del cliente
recigo = ReciGoClient()Ejemplo 1: Estimación de Precio
estimate_price.py
from recigo_client import recigo
def main():
try:
# Estimar precio
estimate = recigo.estimate_price(
size="L",
pickup_address="Calle Gran Vía 1, Madrid",
dropoff_address="Calle Alcalá 100, Madrid",
pickup_floor=3,
pickup_has_elevator=False,
dropoff_floor=1,
dropoff_has_elevator=True,
)
print("Estimación de precio:")
print(f" Total cliente: €{estimate['total_customer_pays_eur']}")
print(f" Precio base: €{estimate['base_price_eur']}")
print(f" Comisión plataforma: €{estimate['platform_fee_eur']}")
print(f" Transportista recibe: €{estimate['helper_receives_eur']}")
print(f" Distancia: {estimate['calculation_breakdown']['distance_km']} km")
print(f" Duración estimada: {estimate['estimated_duration_minutes']} min")
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
main()Ejemplo 2: Crear Trabajo Completo
create_job.py
from recigo_client import recigo
def main():
try:
# Paso 1: Estimar precio (opcional pero recomendado)
print("Estimando precio...")
estimate = recigo.estimate_price(
size="L",
pickup_address="Calle Gran Vía 1, Madrid",
dropoff_address="Calle Alcalá 100, Madrid",
)
print(f"Precio estimado: €{estimate['total_customer_pays_eur']}")
# Paso 2: Crear el trabajo
print("\nCreando trabajo...")
job = recigo.create_job(
external_reference="listing-12345",
title="Mover sofá de 2 plazas",
description="Sofá gris en buen estado, necesita 2 personas",
size="L",
pickup_address="Calle Gran Vía 1, Madrid",
dropoff_address="Calle Alcalá 100, Madrid",
customer_phone="+34612345678",
photos=[
"https://mi-plataforma.com/images/sofa1.jpg",
"https://mi-plataforma.com/images/sofa2.jpg",
],
pickup_floor=3,
pickup_has_elevator=False,
dropoff_floor=1,
dropoff_has_elevator=True,
)
print("\n✅ Trabajo creado exitosamente:")
print(f" Job ID: {job['job_id']}")
print(f" Estado: {job['status']}")
print(f" SMS enviado: {'Sí' if job['sms_sent'] else 'No'}")
print(f" Deep Link: {job['deep_link_url']}")
print(f" Precio total: €{job['pricing']['total_customer_pays_eur']}")
return job
except Exception as e:
print(f"❌ Error: {e}")
return None
if __name__ == "__main__":
main()Ejemplo 3: Consultar Estado del Trabajo
check_status.py
from recigo_client import recigo
def check_job_status(job_id: str):
try:
status = recigo.get_job_status(job_id)
print(f"Estado del trabajo {job_id}:")
print(f" Estado actual: {status['status']}")
print(f" Referencia externa: {status.get('external_reference', 'N/A')}")
print(f" Creado: {status['created_at']}")
print(f" Actualizado: {status['updated_at']}")
# Mensajes según el estado
status_messages = {
"draft": "⏳ El cliente aún no ha completado el pago",
"posted": "✅ El trabajo está publicado y visible para transportistas",
"accepted": "🚗 Un transportista ha aceptado el trabajo",
"completed": "🎉 El trabajo ha sido completado",
"cancelled": "❌ El trabajo ha sido cancelado",
}
message = status_messages.get(status['status'], "Estado desconocido")
print(f"\n{message}")
return status
except Exception as e:
print(f"❌ Error: {e}")
return None
if __name__ == "__main__":
# Reemplaza con tu job_id real
check_job_status("550e8400-e29b-41d4-a716-446655440000")Ejemplo 4: Integración con Flask
Ejemplo de cómo integrar la API de ReciGo en una aplicación Flask:
app.py
from flask import Flask, request, jsonify
from recigo_client import recigo
import os
app = Flask(__name__)
@app.route('/api/delivery/estimate', methods=['POST'])
def estimate_delivery():
"""Estima el precio de un trabajo de entrega."""
try:
data = request.json
estimate = recigo.estimate_price(
size=data['size'],
pickup_address=data['pickup_address'],
dropoff_address=data['dropoff_address'],
)
return jsonify({
'success': True,
'estimate': estimate
})
except Exception as e:
return jsonify({
'success': False,
'error': str(e)
}), 500
@app.route('/api/delivery/create', methods=['POST'])
def create_delivery():
"""Crea un nuevo trabajo de entrega."""
try:
data = request.json
job = recigo.create_job(
external_reference=data.get('listing_id'),
title=data['title'],
description=data['description'],
size=data['size'],
pickup_address=data['pickup_address'],
dropoff_address=data['dropoff_address'],
customer_phone=data['customer_phone'],
photos=data.get('photo_urls', []),
)
# Guardar en tu base de datos
save_to_database(
listing_id=data.get('listing_id'),
recigo_job_id=job['job_id'],
status=job['status'],
deep_link=job['deep_link_url'],
)
return jsonify({
'success': True,
'job_id': job['job_id'],
'deep_link': job['deep_link_url'],
'sms_sent': job['sms_sent'],
})
except Exception as e:
return jsonify({
'success': False,
'error': str(e)
}), 500
@app.route('/api/delivery/<job_id>/status', methods=['GET'])
def get_delivery_status(job_id):
"""Consulta el estado de un trabajo."""
try:
status = recigo.get_job_status(job_id)
return jsonify({
'success': True,
'status': status
})
except Exception as e:
return jsonify({
'success': False,
'error': str(e)
}), 500
def save_to_database(**kwargs):
"""Guarda los datos en tu base de datos."""
# Implementa tu lógica de base de datos aquí
print(f"Guardando en DB: {kwargs}")
if __name__ == '__main__':
port = int(os.environ.get('PORT', 5000))
app.run(host='0.0.0.0', port=port, debug=True)Manejo de Errores
error_handling.py
from recigo_client import recigo
def create_job_with_error_handling():
"""Ejemplo de manejo robusto de errores."""
try:
job = recigo.create_job(
title="Mover sofá",
description="Sofá de 2 plazas",
size="L",
pickup_address="Calle Gran Vía 1, Madrid",
dropoff_address="Calle Alcalá 100, Madrid",
customer_phone="+34612345678",
)
return {'success': True, 'job': job}
except Exception as e:
error_message = str(e)
# Manejo específico de errores
if 'validation_error' in error_message:
return {
'success': False,
'error': 'Los datos proporcionados son inválidos',
'details': error_message
}
elif 'unauthorized' in error_message or '401' in error_message:
return {
'success': False,
'error': 'API key inválida o faltante',
'details': error_message
}
elif 'rate_limit_exceeded' in error_message or '429' in error_message:
return {
'success': False,
'error': 'Demasiadas solicitudes, intenta más tarde',
'details': error_message
}
elif 'geocoding_failed' in error_message or '422' in error_message:
return {
'success': False,
'error': 'Una o más direcciones no son válidas',
'details': error_message
}
else:
# Error genérico
return {
'success': False,
'error': 'Ocurrió un error al crear el trabajo',
'details': error_message
}
# Ejemplo de uso
result = create_job_with_error_handling()
if result['success']:
print(f"✅ Trabajo creado: {result['job']['job_id']}")
else:
print(f"❌ Error: {result['error']}")
if 'details' in result:
print(f" Detalles: {result['details']}")Variables de Entorno
Crea un archivo .env para guardar tu API key:
.env
# ReciGo API Configuration
RECIGO_API_KEY=recigo_live_abc123...
# Flask Configuration
FLASK_APP=app.py
FLASK_ENV=production
PORT=5000Usa python-dotenv para cargar las variables:
pip install python-dotenvconfig.py
import os
from dotenv import load_dotenv
# Cargar variables de entorno
load_dotenv()
# Acceder a las variables
RECIGO_API_KEY = os.environ.get('RECIGO_API_KEY')
if not RECIGO_API_KEY:
raise ValueError("RECIGO_API_KEY no está definida")