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 requests

Cliente 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=5000

Usa python-dotenv para cargar las variables:

pip install python-dotenv
config.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")

Más Ejemplos