Integrations: OAuth, Payments, Search, Email

En la cápsula 04 implementaste el core domain con SQLAlchemy 2.0 y FastAPI async. Funciona en desarrollo. Pero un E-commerce real depende de servicios externos: autenticación social (Google), procesamiento de pagos (Stripe), búsqueda eficiente (PostgreSQL FTS + pg_trgm), envío de emails transaccionales (SendGrid), y background tasks (Celery).

Esta cápsula te lleva a través de las integraciones críticas. No es exhaustiva — es decisional: para cada integración, qué hace, cómo se conecta con el core domain, qué decisiones tomas, y qué código exacto va al repo.

---

Mental model: integrations como adapters

                     ┌──────────────────────────────┐
                     │   E-commerce API (FastAPI)   │
                     └──────────────────────────────┘
                              │ │ │ │ │
        ┌─────────────────────┘ │ │ │ └────────────────────┐
        │            ┌──────────┘ │ └──────┐                │
        ▼            ▼            ▼        ▼                ▼
┌─────────────┐ ┌──────────┐ ┌──────────┐ ┌────────────┐ ┌──────────┐
│ Google OAuth│ │  Stripe  │ │  Redis   │ │  SendGrid  │ │  Celery  │
│   (auth)    │ │ (payments)│ │  (cache) │ │  (emails)  │ │  (jobs)  │
└─────────────┘ └──────────┘ └──────────┘ └────────────┘ └──────────┘

Cada integración es un adapter — un módulo que encapsula el servicio externo. El core domain no llama directamente a Stripe; llama a app/integrations/payments.py, que es lo único que conoce los detalles de Stripe. Esto hace el código testeable (puedes mockear el adapter) y reemplazable (cambiar Stripe por Lemon Squeezy = un solo módulo).

Regla: todo lo que toca un servicio externo vive bajo app/integrations/. Nada de import stripe directo en routers o services del dominio.

---

1. OAuth con Google (login social)

Por qué OAuth

Email + password requiere que el usuario:

1. Recuerde otra contraseña.
2. Confirme su email (otra integración).
3. Haga password reset cuando lo olvide.

OAuth con Google reduce fricción: el usuario hace login con un click, su email ya está verificado, y nunca te pasa una contraseña. Para un E-commerce nuevo, OAuth es el camino corto a más conversiones.

Decisión de diseño

Soporta ambos: email/password (cápsula 04) y OAuth Google. Esto cubre:

• Usuarios que ya tenían cuenta con email
• Usuarios nuevos que prefieren login social
• Casos donde Google está caído (fallback al método tradicional)

Schema cambio

# app/users/models.py — agregar a User
class User(Base):
    # ... existing fields ...
    google_id: Mapped[str | None] = mapped_column(String(100), unique=True, nullable=True, index=True)
    auth_provider: Mapped[str] = mapped_column(String(20), default="email")  # email, google

Migration: alembic revision --autogenerate -m "add google oauth fields".

Adapter

# app/integrations/google_oauth.py
import httpx
from fastapi import HTTPException

from app.config import settings


GOOGLE_TOKEN_URL = "https://oauth2.googleapis.com/token"
GOOGLE_USERINFO_URL = "https://www.googleapis.com/oauth2/v2/userinfo"


async def exchange_code_for_token(code: str) -> dict:
    """Intercambia el code de OAuth por un access_token."""
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.post(
            GOOGLE_TOKEN_URL,
            data={
                "code": code,
                "client_id": settings.google_client_id,
                "client_secret": settings.google_client_secret,
                "redirect_uri": settings.google_redirect_uri,
                "grant_type": "authorization_code",
            },
        )
    if response.status_code != 200:
        raise HTTPException(400, "Invalid Google authorization code")
    return response.json()


async def fetch_user_info(access_token: str) -> dict:
    """Trae perfil del usuario autenticado."""
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(
            GOOGLE_USERINFO_URL,
            headers={"Authorization": f"Bearer {access_token}"},
        )
    if response.status_code != 200:
        raise HTTPException(400, "Failed to fetch Google user info")
    return response.json()

Endpoints

# app/users/router.py — agregar
@router.get("/auth/google/login")
async def google_login_url():
    """Devuelve URL a la que el frontend redirige para iniciar OAuth."""
    url = (
        "https://accounts.google.com/o/oauth2/v2/auth?"
        f"client_id={settings.google_client_id}&"
        f"redirect_uri={settings.google_redirect_uri}&"
        "response_type=code&"
        "scope=openid email profile&"
        "access_type=online"
    )
    return {"auth_url": url}


@router.get("/auth/google/callback")
async def google_callback(code: str, db: AsyncSession = Depends(get_db)):
    from app.integrations.google_oauth import exchange_code_for_token, fetch_user_info

    token_data = await exchange_code_for_token(code)
    user_info = await fetch_user_info(token_data["access_token"])

    # Find or create user
    google_id = user_info["id"]
    email = user_info["email"].lower()

    user = await db.scalar(select(User).where(User.google_id == google_id))
    if not user:
        # Maybe user exists with email but no Google link yet
        user = await db.scalar(select(User).where(User.email == email))
        if user:
            user.google_id = google_id
            user.auth_provider = "google"
        else:
            user = User(
                email=email,
                password_hash="",  # OAuth-only user
                name=user_info.get("name", ""),
                google_id=google_id,
                auth_provider="google",
            )
            db.add(user)

        await db.commit()
        await db.refresh(user)

    # Issue JWT
    access_token = create_token(str(user.id))
    return {"access_token": access_token, "token_type": "bearer", "user_id": str(user.id)}

Trampa común

No verificar el state parameter = vulnerabilidad CSRF. En producción, generas un state random antes de redirigir, lo guardas en sesión/cookie firmada, y verificas que coincida en el callback. Para esta versión, lo documentas como TODO en SECURITY.md.

---

2. Stripe payments (checkout + webhooks)

Flujo correcto

Usuario → Frontend → POST /orders (crea Order pending) → API
                                                              ↓
                              Stripe PaymentIntent ← API ← API
                                       ↓
                              Stripe Confirm ← Frontend ← client_secret
                                       ↓
                       Stripe webhook → API → Update Order (paid)
                                                       ↓
                                              Email confirmation

Patrón clave: la API nunca marca una Order como paid directamente. Solo cuando llega el webhook firmado de Stripe. Esto evita:

• Cliente malicioso que llama un endpoint /orders/{id}/mark-paid (no existe)
• Race conditions entre confirmación local y status real en Stripe

Adapter

# app/integrations/payments.py
import stripe
from app.config import settings

stripe.api_key = settings.stripe_secret_key


async def create_payment_intent(amount: Decimal, order_id: str, user_email: str) -> dict:
    """Crea un PaymentIntent y devuelve client_secret para el frontend."""
    # Stripe usa centavos
    amount_cents = int(amount * 100)

    intent = stripe.PaymentIntent.create(
        amount=amount_cents,
        currency="usd",
        metadata={"order_id": order_id},
        receipt_email=user_email,
        automatic_payment_methods={"enabled": True},
    )
    return {
        "client_secret": intent.client_secret,
        "payment_intent_id": intent.id,
    }


def verify_webhook_signature(payload: bytes, signature: str) -> stripe.Event:
    """Verifica que el webhook venga realmente de Stripe."""
    try:
        return stripe.Webhook.construct_event(
            payload, signature, settings.stripe_webhook_secret
        )
    except (stripe.error.SignatureVerificationError, ValueError):
        raise HTTPException(400, "Invalid webhook signature")

Endpoint: crear order + iniciar pago

# app/orders/router.py — modificar
@router.post("/")
async def create_order_with_payment(
    data: OrderCreate,
    user_id: str = Depends(get_current_user_id),
    db: AsyncSession = Depends(get_db),
):
    from app.integrations.payments import create_payment_intent

    order = await service.create_order(db, uuid.UUID(user_id), data.items)

    user = await db.get(User, uuid.UUID(user_id))
    payment = await create_payment_intent(order.total, str(order.id), user.email)

    # Guardar payment_intent_id en la order
    order.stripe_payment_intent_id = payment["payment_intent_id"]
    await db.commit()

    return {
        "order_id": str(order.id),
        "total": float(order.total),
        "client_secret": payment["client_secret"],
    }

Webhook handler

# app/payments/webhook.py
from fastapi import APIRouter, Request, HTTPException
router = APIRouter()


@router.post("/stripe-webhook")
async def stripe_webhook(request: Request, db: AsyncSession = Depends(get_db)):
    from app.integrations.payments import verify_webhook_signature

    payload = await request.body()
    signature = request.headers.get("stripe-signature", "")

    event = verify_webhook_signature(payload, signature)

    if event["type"] == "payment_intent.succeeded":
        intent = event["data"]["object"]
        order_id = intent["metadata"].get("order_id")

        order = await db.get(Order, uuid.UUID(order_id))
        if order and order.status == "pending":
            order.status = "paid"
            await db.commit()

            # Background task: email + reduce stock event
            from app.integrations.tasks import send_order_confirmation
            send_order_confirmation.delay(str(order.id))

    elif event["type"] == "payment_intent.payment_failed":
        intent = event["data"]["object"]
        order_id = intent["metadata"].get("order_id")
        order = await db.get(Order, uuid.UUID(order_id))
        if order:
            order.status = "cancelled"
            # Stock rollback
            await service.restore_stock(db, order)
            await db.commit()

    return {"received": True}

Idempotencia

Stripe puede reintentar un webhook si tu API responde lento o falla. Si el mismo payment_intent.succeeded llega dos veces, no debes contar dos veces el pago. La solución: verificar if order.status == "pending" antes de marcar como paid. Si ya está paid, el segundo evento no hace nada.

---

3. Búsqueda con PostgreSQL FTS + pg_trgm

Por qué PostgreSQL FTS y no Elasticsearch

Para un E-commerce con < 1M productos:

• PostgreSQL FTS es suficiente: tsvector + GIN index, búsqueda en milisegundos.
• pg_trgm agrega tolerancia a typos ("iphne" → "iphone").
• Elasticsearch agrega un servicio nuevo (otro deploy, otro costo, otra DB que sincronizar). Sobreingeniería para este scale.

Si crece a > 10M productos o necesitas faceted search avanzado, Elasticsearch es el upgrade. Hasta entonces, PostgreSQL gana en simplicidad.

Schema (ya hecho en cap 04)

Recordatorio: la migration agregó search_vector tsvector GENERATED ALWAYS AS (...) STORED con GIN index sobre él, y un índice trigram sobre name.

Service mejorado

# app/products/service.py
from sqlalchemy import text


async def search_products(
    db: AsyncSession,
    redis: Redis,
    q: str,
    page: int = 1,
    page_size: int = 20,
) -> list[dict]:
    """
    Búsqueda híbrida:
    1. FTS (search_vector) — exact match en términos.
    2. Trigram (pg_trgm) — fallback para typos.
    3. Ranking por similarity + recency.
    """
    cache_key = f"search:{q.lower()}:{page}:{page_size}"
    cached = await redis.get(cache_key)
    if cached:
        return json.loads(cached)

    query = text("""
        SELECT
            id, name, description, price, image_url,
            ts_rank(search_vector, plainto_tsquery('spanish', :q)) AS fts_rank,
            similarity(name, :q) AS trgm_sim
        FROM products
        WHERE
            search_vector @@ plainto_tsquery('spanish', :q)
            OR name % :q
        ORDER BY
            fts_rank DESC NULLS LAST,
            trgm_sim DESC NULLS LAST,
            created_at DESC
        LIMIT :limit OFFSET :offset
    """)

    result = await db.execute(query, {
        "q": q,
        "limit": page_size,
        "offset": (page - 1) * page_size,
    })
    rows = result.mappings().all()
    data = [dict(row) for row in rows]

    await redis.setex(cache_key, 60, json.dumps(data, default=str))
    return data

Endpoint

@router.get("/search")
async def search(
    q: str,
    page: int = 1,
    page_size: int = 20,
    db: AsyncSession = Depends(get_db),
    redis: Redis = Depends(get_redis),
):
    if len(q) < 2:
        raise HTTPException(400, "Query must be at least 2 chars")
    return await service.search_products(db, redis, q, page, page_size)

---

4. Emails transaccionales con SendGrid

Decisión

SendGrid sobre SMTP custom porque:

• Deliverability garantizada (reputation manejada por ellos)
• Templates editables sin redeploy
• Tracking de opens/clicks
• $0 en free tier hasta 100 emails/día (suficiente para portfolio)

Adapter

# app/integrations/email.py
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail

from app.config import settings


_client = SendGridAPIClient(settings.sendgrid_api_key)


def send_email(
    to_email: str,
    subject: str,
    template_id: str,
    dynamic_data: dict,
) -> bool:
    message = Mail(
        from_email=(settings.from_email, settings.from_name),
        to_emails=to_email,
    )
    message.template_id = template_id
    message.dynamic_template_data = dynamic_data

    try:
        response = _client.send(message)
        return response.status_code in (200, 202)
    except Exception as e:
        # Log a Sentry; no crashear el caller
        import sentry_sdk
        sentry_sdk.capture_exception(e)
        return False

Templates en SendGrid (configuras en su dashboard)

• order_confirmation — recibo del pago
• password_reset — reset link
• shipping_notification — order shipped
• welcome — al registrarse

Cuando llamarlo: nunca síncrono

Anti-pattern crítico: mandar email en el handler del request. Si SendGrid está lento (500ms+), tu API responde lenta. Si SendGrid está caído, tu API tira 500.

Patrón correcto: background task con Celery (siguiente sección).

---

5. Background tasks con Celery

Cuándo background y cuándo no

Operación	¿Background?	Por qué
Crear order + payment intent	❌ Síncrono	El user espera el client_secret
Mandar email confirmation	✅ Background	Email puede tardar; user no espera
Generar invoice PDF	✅ Background	CPU-heavy, no bloquea response
Update inventory analytics	✅ Background	Eventually consistent OK
Webhook de Stripe	❌ Síncrono	Stripe espera 200 en < 5s
Recompute search index	✅ Background	Puede tardar minutos

Setup

# app/integrations/celery_app.py
from celery import Celery
from app.config import settings


celery_app = Celery(
    "ecommerce",
    broker=settings.redis_url,
    backend=settings.redis_url,
)

celery_app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
    task_track_started=True,
    task_time_limit=300,  # hard limit 5 min
    task_soft_time_limit=240,  # soft limit 4 min
)

Tasks

# app/integrations/tasks.py
from app.integrations.celery_app import celery_app
from app.integrations.email import send_email


@celery_app.task(bind=True, max_retries=3, default_retry_delay=60)
def send_order_confirmation(self, order_id: str):
    """Background task — manda email de confirmación de orden."""
    import asyncio
    from app.database import async_session
    from app.orders.models import Order
    from app.users.models import User

    async def _run():
        async with async_session() as db:
            order = await db.get(Order, order_id)
            if not order:
                return
            user = await db.get(User, order.user_id)
            ok = send_email(
                to_email=user.email,
                subject="Tu pedido ha sido confirmado",
                template_id="d-xxxxxxxxxxxxxxx",  # SendGrid template ID
                dynamic_data={
                    "user_name": user.name,
                    "order_id": str(order.id),
                    "total": float(order.total),
                },
            )
            if not ok:
                raise Exception("SendGrid failed")

    try:
        asyncio.run(_run())
    except Exception as exc:
        raise self.retry(exc=exc)

bind=True + self.retry(...) = retry exponencial automático. Si SendGrid falla 3 veces, el task termina con FAILURE y queda en logs/Sentry.

Worker en producción

# En Render: agregar Celery worker como servicio adicional
celery -A app.integrations.celery_app worker --loglevel=info --concurrency=4

---

Trampas comunes en integraciones

1. Hardcodear URLs y keys en código

stripe.api_key = "sk_test_..." directo en payments.py se filtra al primer push. Todas las keys van por pydantic-settings desde environment variables.

2. No manejar timeouts

httpx.AsyncClient() sin timeout = tu request queda colgado si Google está caído. Siempre timeout=10.0 o menos.

3. Webhooks sin verificación de firma

Cualquiera puede hacer POST a /stripe-webhook con un payload falso de "payment.succeeded". Sin verificar la firma, marcas órdenes como pagadas sin que se haya pagado nada. Siempre stripe.Webhook.construct_event(...).

4. Síncrono lo que debería ser async

Mandar email en el endpoint = el response time del endpoint depende de SendGrid. Cuando SendGrid se cae, tu API parece caída. Background tasks resuelven esto.

5. Idempotencia ignorada

Webhooks pueden llegar duplicados. Background tasks pueden ejecutarse dos veces. Si tu código asume "esto se ejecuta una sola vez", produces datos duplicados (cobros dobles, emails dobles).

---

Checkpoint

Al terminar esta cápsula, tu API debe poder:

• ✅ Login con Google OAuth (genera JWT compatible con email/password)
• ✅ Crear orden + iniciar pago Stripe → webhook actualiza status
• ✅ Búsqueda de productos con FTS + trigram fallback
• ✅ Mandar emails transaccionales vía SendGrid (background task)
• ✅ Celery worker corriendo separado de la API web

Probado localmente: deberías poder registrarte con Google, comprar un producto con tarjeta de testing de Stripe (4242 4242 4242 4242), y recibir email de confirmación (en SendGrid sandbox).

Cápsula siguiente: deployment + infrastructure. Vas a llevar todo esto a Render + Supabase + Upstash + Cloudflare con CI/CD automático.

---

Recursos

1. Stripe API — PaymentIntents — referencia oficial.
2. Stripe Webhooks Guide — incluye signature verification.
3. Google OAuth 2.0 docs — flow completo.
4. PostgreSQL Full Text Search — referencia oficial.
5. Celery 5 Documentation — para tasks complejas.
6. SendGrid Python Library — incluye dynamic templates.

---

Cápsula 05 de 08 — Módulo 7 — Deployment & System Design Guide