45ffba69b2
Co-authored-by: MENDOZA BALLARDO GAEL RICARDO <gael-meb123@users.noreply.github.com> Co-authored-by: Azareth-Tr <Azareth-Tr@users.noreply.github.com> modificacion de las vistas principales para el usuario ciudadano, primer avance para el panel admin
99 lines
3.1 KiB
Python
99 lines
3.1 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)
|
|
|
|
insert_data: dict = {
|
|
"user_id": current_user["user_id"],
|
|
"label": body.label,
|
|
"calle": body.calle,
|
|
"colonia": body.colonia,
|
|
"route_id": route_id,
|
|
"verified": False,
|
|
}
|
|
if body.lat is not None:
|
|
insert_data["lat"] = body.lat
|
|
if body.lng is not None:
|
|
insert_data["lng"] = body.lng
|
|
|
|
result = (
|
|
supabase_admin.table("addresses")
|
|
.insert(insert_data)
|
|
.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
|