115 lines
3.5 KiB
Python
115 lines
3.5 KiB
Python
from fastapi import APIRouter, HTTPException, Depends
|
|
from pydantic import BaseModel
|
|
from typing import Optional
|
|
|
|
from app.db.database import get_db
|
|
from app.core.dependencies import get_current_user
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
class AddressCreate(BaseModel):
|
|
lat: float
|
|
lng: float
|
|
alias: Optional[str] = None
|
|
address_text: str
|
|
|
|
|
|
@router.post("/", summary="Crear nueva dirección")
|
|
async def create_address(
|
|
address: AddressCreate,
|
|
current_user: dict = Depends(get_current_user)
|
|
):
|
|
"""Crear dirección para usuario autenticado."""
|
|
# Determinar ruta basada en ubicación (simplificado - usar PostGIS en producción)
|
|
route_id = "RUTA-01" # Mock: calcular basado en lat/lng
|
|
|
|
db = get_db()
|
|
try:
|
|
result = db.table("addresses").insert({
|
|
"user_id": current_user["id"],
|
|
"alias": address.alias,
|
|
"lat": address.lat,
|
|
"lng": address.lng,
|
|
"route_id": route_id,
|
|
}).execute()
|
|
|
|
new_address = result.data[0]
|
|
return {
|
|
"id": new_address["id"],
|
|
"user_id": new_address["user_id"],
|
|
"alias": new_address["alias"],
|
|
"lat": new_address["lat"],
|
|
"lng": new_address["lng"],
|
|
"route_id": new_address["route_id"],
|
|
}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.get("/", summary="Obtener direcciones del usuario")
|
|
async def get_addresses(current_user: dict = Depends(get_current_user)):
|
|
"""Obtener todas las direcciones del usuario autenticado."""
|
|
db = get_db()
|
|
try:
|
|
result = db.table("addresses").select(
|
|
"id, alias, lat, lng, route_id"
|
|
).eq("user_id", current_user["id"]).execute()
|
|
|
|
return result.data
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.get("/{address_id}", summary="Obtener dirección específica")
|
|
async def get_address(
|
|
address_id: int,
|
|
current_user: dict = Depends(get_current_user)
|
|
):
|
|
"""Obtener detalle de una dirección específica (solo del usuario)."""
|
|
db = get_db()
|
|
try:
|
|
result = db.table("addresses").select("*").eq("id", address_id).execute()
|
|
|
|
if not result.data:
|
|
raise HTTPException(status_code=404, detail="Address not found")
|
|
|
|
address = result.data[0]
|
|
|
|
# RBAC: verificar que la dirección pertenece al usuario
|
|
if address["user_id"] != current_user["id"]:
|
|
raise HTTPException(status_code=403, detail="No autorizado")
|
|
|
|
return address
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.delete("/{address_id}", summary="Eliminar dirección")
|
|
async def delete_address(
|
|
address_id: int,
|
|
current_user: dict = Depends(get_current_user)
|
|
):
|
|
"""Eliminar dirección del usuario."""
|
|
db = get_db()
|
|
try:
|
|
# Verificar RBAC primero
|
|
result = db.table("addresses").select("user_id").eq("id", address_id).execute()
|
|
|
|
if not result.data:
|
|
raise HTTPException(status_code=404, detail="Address not found")
|
|
|
|
if result.data[0]["user_id"] != current_user["id"]:
|
|
raise HTTPException(status_code=403, detail="No autorizado")
|
|
|
|
# Eliminar
|
|
db.table("addresses").delete().eq("id", address_id).execute()
|
|
|
|
return {"ok": True, "message": "Address deleted"}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|