async/await de verdad

"""Constantes y formas de datos del módulo enrichment (guideline 1: sin números mágicos)."""

from __future__ import annotations

from dataclasses import dataclass, field

# Timeout total por llamada saliente (segundos). En m-b-core el equivalente es FETCH_TIMEOUT_S.
HTTP_TIMEOUT_S = 5.0
# Máximo de llamadas concurrentes al upstream por request (no saturar al legacy).
MAX_CONCURRENCY = 3
# Radio de "cerca" para contar locales vecinos (km).
NEARBY_RADIUS_KM = 1.5


@dataclass(frozen=True)
class LocalEnriquecido:
    """Resultado de enriquecer un local con tres fuentes independientes."""

    local_id: int
    nombre: str
    rating: float | None
    fotos: int
    vecinos_cerca: int
    fuentes_ok: tuple[str, ...] = field(default_factory=tuple)

"""Un solo cliente HTTP saliente para todo el módulo, con timeout de la casa."""

from __future__ import annotations

import httpx

from .models import HTTP_TIMEOUT_S


def make_client(base_url: str) -> httpx.AsyncClient:
    """Cliente async compartido. Se crea una vez (lifespan de la app o fixture) y se inyecta."""
    return httpx.AsyncClient(
        base_url=base_url,
        timeout=httpx.Timeout(HTTP_TIMEOUT_S),
        headers={"User-Agent": "mesa-core-practicas/enrichment"},
    )

"""I/O: llamadas a los tres servicios que enriquecen un local. Sin lógica de negocio.

`r.json()` devuelve `Any`; con `warn_return_any` (config de m-b-core) mypy exige que
digamos qué esperamos. Anotar la variable es la forma barata de hacerlo explícito.
"""

from __future__ import annotations

from typing import Any

import httpx


async def fetch_rating(client: httpx.AsyncClient, local_id: int) -> dict[str, Any]:
    r = await client.get(f"/ratings/{local_id}")
    r.raise_for_status()
    data: dict[str, Any] = r.json()
    return data


async def fetch_fotos(client: httpx.AsyncClient, local_id: int) -> list[dict[str, Any]]:
    r = await client.get(f"/fotos/{local_id}")
    r.raise_for_status()
    data: list[dict[str, Any]] = r.json()
    return data


async def fetch_vecinos(client: httpx.AsyncClient, local_id: int) -> list[dict[str, Any]]:
    r = await client.get(f"/vecinos/{local_id}")
    r.raise_for_status()
    data: list[dict[str, Any]] = r.json()
    return data

"""Funciones puras: cálculo sin I/O. Se testean con parametrize, sin mocks."""

from __future__ import annotations

import math
from typing import Any

from .models import NEARBY_RADIUS_KM

_EARTH_KM = 6371.0


def haversine_km(lat1: float, lng1: float, lat2: float, lng2: float) -> float:
    """Distancia en km entre dos coordenadas."""
    p1, p2 = math.radians(lat1), math.radians(lat2)
    dphi = math.radians(lat2 - lat1)
    dl = math.radians(lng2 - lng1)
    a = math.sin(dphi / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2
    return 2 * _EARTH_KM * math.asin(math.sqrt(a))


def contar_vecinos_cerca(
    origen: tuple[float, float], vecinos: list[dict[str, Any]], radio_km: float = NEARBY_RADIUS_KM
) -> int:
    """Cuenta vecinos dentro del radio. CPU pura: en el servicio corre en to_thread."""
    lat0, lng0 = origen
    return sum(
        1
        for v in vecinos
        if v.get("lat") is not None
        and v.get("lng") is not None
        and haversine_km(lat0, lng0, float(v["lat"]), float(v["lng"])) <= radio_km
    )

"""Orquestador: tres fuentes EN PARALELO, con límite de concurrencia, degradando por fuente.

Patrón de m-b-core: services/api/app/availability_engine/legacy_queries.py:136
    results = await asyncio.gather(*[_query(q, params, enabled) for q in QUERIES])
y el Semaphore del workbench (availability_engine/workbench/parity.py:920).
"""

from __future__ import annotations

import asyncio
import logging
from typing import Any

import httpx

from . import queries
from .models import MAX_CONCURRENCY, LocalEnriquecido
from .utils import contar_vecinos_cerca

log = logging.getLogger(__name__)


async def _guardado(sem: asyncio.Semaphore, nombre: str, coro: Any) -> tuple[str, Any | None]:
    """Ejecuta una fuente bajo el semáforo; un fallo de red degrada a None y se registra.

    Solo capturamos errores de red/HTTP (lo que sabemos manejar); cualquier otra
    excepción sube.
    """
    async with sem:
        try:
            return nombre, await coro
        except (httpx.HTTPError, TimeoutError) as e:
            log.warning("enrichment.fuente_fallo", extra={"fuente": nombre, "error": type(e).__name__})
            return nombre, None


async def enriquecer_local(
    client: httpx.AsyncClient,
    local_id: int,
    nombre: str,
    coords: tuple[float, float],
) -> LocalEnriquecido:
    sem = asyncio.Semaphore(MAX_CONCURRENCY)
    resultados = await asyncio.gather(
        _guardado(sem, "rating", queries.fetch_rating(client, local_id)),
        _guardado(sem, "fotos", queries.fetch_fotos(client, local_id)),
        _guardado(sem, "vecinos", queries.fetch_vecinos(client, local_id)),
    )
    por_fuente: dict[str, Any] = dict(resultados)
    ok = tuple(k for k, v in por_fuente.items() if v is not None)

    rating = por_fuente["rating"]["rating"] if por_fuente["rating"] else None
    fotos = len(por_fuente["fotos"] or [])
    # CPU pura sobre una lista potencialmente grande: fuera del loop.
    vecinos = await asyncio.to_thread(contar_vecinos_cerca, coords, por_fuente["vecinos"] or [])

    return LocalEnriquecido(
        local_id=local_id,
        nombre=nombre,
        rating=rating,
        fotos=fotos,
        vecinos_cerca=vecinos,
        fuentes_ok=ok,
    )

import pytest

from services.api.app.enrichment.client import make_client

BASE = "https://upstream.test"


@pytest.fixture()
async def client():
    """Un cliente por test, cerrado al terminar (async fixture: asyncio_mode = auto)."""
    c = make_client(BASE)
    yield c
    await c.aclose()


@pytest.fixture()
def coords_lima() -> tuple[float, float]:
    return (-12.1211, -77.0300)

"""Funciones puras → parametrizar bordes, sin mocks (TESTING.md §1)."""

import pytest

from services.api.app.enrichment.utils import contar_vecinos_cerca, haversine_km

LIMA = (-12.1211, -77.0300)


@pytest.mark.parametrize(
    "a, b, esperado_km",
    [
        (LIMA, LIMA, 0.0),
        (LIMA, (-12.1211, -77.0400), 1.09),  # ~1 km al oeste
        (LIMA, (-33.4489, -70.6693), 2458.0),  # Santiago
    ],
)
def test_haversine(a, b, esperado_km):
    assert haversine_km(*a, *b) == pytest.approx(esperado_km, abs=0.5)


@pytest.mark.parametrize(
    "vecinos, esperado",
    [
        ([], 0),
        ([{"lat": -12.1211, "lng": -77.0300}], 1),  # el mismo punto
        ([{"lat": -12.1211, "lng": -77.0400}, {"lat": -33.4, "lng": -70.6}], 1),  # 1 km sí, Santiago no
        ([{"lat": None, "lng": -77.0}, {"lng": -77.0}], 0),  # filas sucias no cuentan (dict.get)
    ],
)
def test_contar_vecinos_cerca(vecinos, esperado):
    assert contar_vecinos_cerca(LIMA, vecinos) == esperado

"""Orquestador → red simulada con respx (mock en el borde de red, no en la función bajo prueba)."""

import httpx
import respx

from services.api.app.enrichment.service import enriquecer_local

from .conftest import BASE


@respx.mock(base_url=BASE)
async def test_tres_fuentes_ok(respx_mock, client, coords_lima):
    respx_mock.get("/ratings/11").mock(return_value=httpx.Response(200, json={"rating": 4.6}))
    respx_mock.get("/fotos/11").mock(return_value=httpx.Response(200, json=[{"id": 1}, {"id": 2}]))
    respx_mock.get("/vecinos/11").mock(
        return_value=httpx.Response(200, json=[{"lat": -12.1211, "lng": -77.0310}])
    )

    r = await enriquecer_local(client, 11, "Maido", coords_lima)

    assert r.rating == 4.6
    assert r.fotos == 2
    assert r.vecinos_cerca == 1
    assert r.fuentes_ok == ("rating", "fotos", "vecinos")


@respx.mock(base_url=BASE)
async def test_una_fuente_caida_degrada_no_rompe(respx_mock, client, coords_lima):
    respx_mock.get("/ratings/11").mock(side_effect=httpx.ConnectError("boom"))  # servidor caído
    respx_mock.get("/fotos/11").mock(return_value=httpx.Response(200, json=[]))
    respx_mock.get("/vecinos/11").mock(return_value=httpx.Response(200, json=[]))

    r = await enriquecer_local(client, 11, "Maido", coords_lima)

    assert r.rating is None
    assert r.fuentes_ok == ("fotos", "vecinos")


@respx.mock(base_url=BASE)
async def test_500_del_upstream_es_fuente_caida(respx_mock, client, coords_lima):
    respx_mock.get("/ratings/11").mock(return_value=httpx.Response(500))
    respx_mock.get("/fotos/11").mock(return_value=httpx.Response(200, json=[]))
    respx_mock.get("/vecinos/11").mock(return_value=httpx.Response(200, json=[]))

    r = await enriquecer_local(client, 11, "Maido", coords_lima)

    assert "rating" not in r.fuentes_ok


@respx.mock(base_url=BASE)
async def test_timeout_no_cuelga_el_request(respx_mock, client, coords_lima):
    """respx no simula el paso del tiempo: el timeout se simula lanzando la excepción que
    lanzaría el transporte real (httpx.ReadTimeout). Lo que probamos es que el orquestador
    la trata como fuente caída y no la deja subir."""
    respx_mock.get("/ratings/11").mock(side_effect=httpx.ReadTimeout("timed out"))
    respx_mock.get("/fotos/11").mock(return_value=httpx.Response(200, json=[]))
    respx_mock.get("/vecinos/11").mock(return_value=httpx.Response(200, json=[]))

    r = await enriquecer_local(client, 11, "Maido", coords_lima)

    assert r.rating is None
    assert r.fuentes_ok == ("fotos", "vecinos")


async def test_el_cliente_de_la_casa_tiene_timeout(client):
    """Guardarraíl: si alguien quita el timeout del cliente, este test lo pilla."""
    assert client.timeout.connect == pytest.approx(5.0)
    assert client.timeout.read == pytest.approx(5.0)

"""Servidor de pruebas: tres endpoints que tardan 300 ms cada uno (simulan el legacy)."""

import asyncio

from fastapi import FastAPI

app = FastAPI()
LATENCIA_S = 0.3


@app.get("/ratings/{local_id}")
async def ratings(local_id: int):
    await asyncio.sleep(LATENCIA_S)
    return {"rating": 4.6}


@app.get("/fotos/{local_id}")
async def fotos(local_id: int):
    await asyncio.sleep(LATENCIA_S)
    return [{"id": i} for i in range(3)]


@app.get("/vecinos/{local_id}")
async def vecinos(local_id: int):
    await asyncio.sleep(LATENCIA_S)
    return [{"lat": -12.1211 + i * 0.001, "lng": -77.03} for i in range(2000)]

"""Mide: tres llamadas en serie (como en PHP) vs gather (como en m-b-core)."""

import asyncio
import time

from services.api.app.enrichment import queries
from services.api.app.enrichment.client import make_client
from services.api.app.enrichment.service import enriquecer_local

BASE = "http://127.0.0.1:8099"


async def en_serie(client, local_id):
    await queries.fetch_rating(client, local_id)
    await queries.fetch_fotos(client, local_id)
    await queries.fetch_vecinos(client, local_id)


async def main():
    async with make_client(BASE) as client:
        t0 = time.perf_counter()
        await en_serie(client, 11)
        t_serie = time.perf_counter() - t0

        t0 = time.perf_counter()
        r = await enriquecer_local(client, 11, "Maido", (-12.1211, -77.03))
        t_par = time.perf_counter() - t0

    print(f"en serie  : {t_serie:.2f} s  (3 x 300 ms + red)")
    print(f"con gather: {t_par:.2f} s  -> {r}")


asyncio.run(main())

"""Error clásico 1: olvidar el await. No da error: devuelve una corrutina sin ejecutar."""

import asyncio


async def buscar_local(local_id: int) -> dict:
    await asyncio.sleep(0.01)
    return {"id": local_id, "nombre": "Maido"}


async def main():
    local = buscar_local(11)  # <- falta el await
    print("tipo:", type(local).__name__)
    print("nombre:", local["nombre"])  # TypeError: 'coroutine' object is not subscriptable


asyncio.run(main())

"""Error clásico 2: time.sleep dentro de async bloquea el loop; asyncio.sleep lo cede."""

import asyncio
import time


async def tarea(nombre: str, dormir):
    await dormir(0.5)
    return nombre


async def correr(dormir, etiqueta):
    t0 = time.perf_counter()
    await asyncio.gather(*[tarea(f"t{i}", dormir) for i in range(5)])
    print(f"{etiqueta:<22} 5 tareas de 0.5 s -> {time.perf_counter() - t0:.2f} s")


async def dormir_bloqueante(s: float):
    time.sleep(s)  # <- bloquea el loop: nadie más corre mientras tanto


asyncio.run(correr(asyncio.sleep, "asyncio.sleep (cede):"))
asyncio.run(correr(dormir_bloqueante, "time.sleep (bloquea):"))

"""Error clásico 3: un cliente HTTP síncrono (requests) dentro de async def.
Funciona... y bloquea el loop en cada llamada. Comparado con httpx.AsyncClient."""

import asyncio
import time

import httpx
import requests

BASE = "http://127.0.0.1:8099"


async def con_requests(local_id: int):
    return requests.get(f"{BASE}/ratings/{local_id}", timeout=5).json()  # bloqueante


async def con_httpx(client: httpx.AsyncClient, local_id: int):
    r = await client.get(f"/ratings/{local_id}")
    return r.json()


async def main():
    t0 = time.perf_counter()
    await asyncio.gather(*[con_requests(i) for i in range(5)])
    print(f"requests en async def: 5 llamadas -> {time.perf_counter() - t0:.2f} s (en serie, aunque uses gather)")

    async with httpx.AsyncClient(base_url=BASE, timeout=5) as client:
        t0 = time.perf_counter()
        await asyncio.gather(*[con_httpx(client, i) for i in range(5)])
        print(f"httpx.AsyncClient   : 5 llamadas -> {time.perf_counter() - t0:.2f} s (de verdad en paralelo)")


asyncio.run(main())