341de89004
PUT /api/decisions/{listing_id} to set decision,
GET /api/decisions to list all user decisions,
DELETE /api/decisions/{listing_id} to remove a decision.
All 6 API route tests pass.
111 lines
3.4 KiB
Python
111 lines
3.4 KiB
Python
"""API routes for listing decisions (like/dislike)."""
|
|
import logging
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel, Field
|
|
|
|
from api.auth import User, get_current_user
|
|
from database import engine
|
|
from repositories.decision_repository import DecisionRepository
|
|
from repositories.user_repository import UserRepository
|
|
from services import decision_service
|
|
|
|
logger = logging.getLogger("uvicorn")
|
|
|
|
decision_router = APIRouter(prefix="/api/decisions", tags=["decisions"])
|
|
|
|
|
|
class SetDecisionRequest(BaseModel):
|
|
decision: str = Field(description="'liked' or 'disliked'")
|
|
listing_type: str = Field(default="RENT", description="'RENT' or 'BUY'")
|
|
|
|
|
|
class DecisionResponse(BaseModel):
|
|
id: int
|
|
listing_id: int
|
|
listing_type: str
|
|
decision: str
|
|
created_at: str
|
|
updated_at: str
|
|
|
|
|
|
def _get_user_id(user: User) -> int:
|
|
"""Resolve auth User to database user ID."""
|
|
user_repo = UserRepository(engine)
|
|
db_user = user_repo.get_user_by_email(user.email)
|
|
if db_user is None:
|
|
db_user = user_repo.create_user(user.email)
|
|
if db_user.id is None:
|
|
raise HTTPException(status_code=500, detail="Failed to create user")
|
|
return db_user.id
|
|
|
|
|
|
@decision_router.put("/{listing_id}", response_model=DecisionResponse)
|
|
async def set_decision(
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
listing_id: int,
|
|
body: SetDecisionRequest,
|
|
) -> DecisionResponse:
|
|
"""Set a decision (like/dislike) on a listing."""
|
|
user_id = _get_user_id(user)
|
|
repo = DecisionRepository(engine)
|
|
try:
|
|
result = decision_service.set_decision(
|
|
repo,
|
|
user_id=user_id,
|
|
listing_id=listing_id,
|
|
listing_type=body.listing_type,
|
|
decision=body.decision,
|
|
)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
return DecisionResponse(
|
|
id=result.id, # type: ignore[arg-type]
|
|
listing_id=result.listing_id,
|
|
listing_type=result.listing_type,
|
|
decision=result.decision,
|
|
created_at=result.created_at.isoformat(),
|
|
updated_at=result.updated_at.isoformat(),
|
|
)
|
|
|
|
|
|
@decision_router.get("", response_model=list[DecisionResponse])
|
|
async def get_decisions(
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
) -> list[DecisionResponse]:
|
|
"""Get all decisions for the current user."""
|
|
user_id = _get_user_id(user)
|
|
repo = DecisionRepository(engine)
|
|
decisions = decision_service.get_decisions(repo, user_id=user_id)
|
|
return [
|
|
DecisionResponse(
|
|
id=d.id, # type: ignore[arg-type]
|
|
listing_id=d.listing_id,
|
|
listing_type=d.listing_type,
|
|
decision=d.decision,
|
|
created_at=d.created_at.isoformat(),
|
|
updated_at=d.updated_at.isoformat(),
|
|
)
|
|
for d in decisions
|
|
]
|
|
|
|
|
|
@decision_router.delete("/{listing_id}")
|
|
async def delete_decision(
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
listing_id: int,
|
|
listing_type: str = "RENT",
|
|
) -> dict[str, bool]:
|
|
"""Remove a decision (back to neutral)."""
|
|
user_id = _get_user_id(user)
|
|
repo = DecisionRepository(engine)
|
|
deleted = decision_service.clear_decision(
|
|
repo,
|
|
user_id=user_id,
|
|
listing_id=listing_id,
|
|
listing_type=listing_type,
|
|
)
|
|
if not deleted:
|
|
raise HTTPException(status_code=404, detail="Decision not found")
|
|
return {"success": True}
|