95 lines
3.0 KiB
Python
95 lines
3.0 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from app.schemas.addresses import AddressCreate, AddressResponse
|
|
from app.core.deps import get_current_user, require_role
|
|
from app.core.supabase_client import supabase_admin
|
|
from app.services.simulation import get_colonias
|
|
|
|
router = APIRouter(prefix="/addresses", tags=["addresses"])
|
|
|
|
|
|
def _resolve_route_id(colonia: str) -> str:
|
|
"""Deriva routeId desde colonias-rutas.json; lanza 404 si la colonia no existe."""
|
|
mapping = get_colonias()
|
|
match = next(
|
|
(c for c in mapping if c.get("colonia", "").lower() == colonia.lower()), None
|
|
)
|
|
if not match:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Colonia '{colonia}' no encontrada. Usa GET /colonias para ver las opciones.",
|
|
)
|
|
return match["routeId"]
|
|
|
|
|
|
@router.post("", response_model=AddressResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_address(
|
|
body: AddressCreate,
|
|
current_user: dict = Depends(require_role("citizen", "admin")),
|
|
):
|
|
"""Alta de domicilio. El routeId se deriva automáticamente de la colonia elegida."""
|
|
route_id = _resolve_route_id(body.colonia)
|
|
|
|
result = (
|
|
supabase_admin.table("addresses")
|
|
.insert(
|
|
{
|
|
"user_id": current_user["user_id"],
|
|
"label": body.label,
|
|
"calle": body.calle,
|
|
"colonia": body.colonia,
|
|
"route_id": route_id,
|
|
"verified": False,
|
|
}
|
|
)
|
|
.execute()
|
|
)
|
|
|
|
if not result.data:
|
|
raise HTTPException(status_code=500, detail="No se pudo guardar el domicilio")
|
|
|
|
return result.data[0]
|
|
|
|
|
|
@router.get("", response_model=list[AddressResponse])
|
|
def list_addresses(current_user: dict = Depends(get_current_user)):
|
|
"""
|
|
Lista de domicilios.
|
|
- Ciudadano: solo sus propios (filtro por user_id en código + RLS en BD).
|
|
- Admin: todos los domicilios.
|
|
"""
|
|
if current_user["role"] == "admin":
|
|
result = supabase_admin.table("addresses").select("*").execute()
|
|
else:
|
|
result = (
|
|
supabase_admin.table("addresses")
|
|
.select("*")
|
|
.eq("user_id", current_user["user_id"])
|
|
.execute()
|
|
)
|
|
|
|
return result.data or []
|
|
|
|
|
|
@router.get("/{address_id}", response_model=AddressResponse)
|
|
def get_address(
|
|
address_id: str,
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""Detalle de un domicilio. El ciudadano solo puede ver los suyos."""
|
|
result = (
|
|
supabase_admin.table("addresses")
|
|
.select("*")
|
|
.eq("id", address_id)
|
|
.maybe_single()
|
|
.execute()
|
|
)
|
|
|
|
if not result.data:
|
|
raise HTTPException(status_code=404, detail="Domicilio no encontrado")
|
|
|
|
address = result.data
|
|
if current_user["role"] != "admin" and address["user_id"] != current_user["user_id"]:
|
|
raise HTTPException(status_code=403, detail="No tienes acceso a este domicilio")
|
|
|
|
return address
|