mirror of
http://88.130.71.182:3000/BlitTech/deals24togo_be.git
synced 2026-06-12 23:33:21 +00:00
80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
"""Favorites / wishlist service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from app.core.exceptions import ConflictException, NotFoundException
|
|
from app.core.supabase import get_supabase_admin
|
|
|
|
|
|
class FavoriteService:
|
|
def __init__(self):
|
|
self.db = get_supabase_admin()
|
|
|
|
def add_favorite(self, user_id: str, listing_id: str) -> dict:
|
|
# Check listing exists
|
|
listing = self.db.table("listings").select("id").eq("id", listing_id).execute()
|
|
if not listing.data:
|
|
raise NotFoundException("Listing not found")
|
|
|
|
# Check if already favorited
|
|
existing = (
|
|
self.db.table("favorites")
|
|
.select("id")
|
|
.eq("user_id", user_id)
|
|
.eq("listing_id", listing_id)
|
|
.execute()
|
|
)
|
|
if existing.data:
|
|
raise ConflictException("Listing already in favorites")
|
|
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
result = (
|
|
self.db.table("favorites")
|
|
.insert({"user_id": user_id, "listing_id": listing_id, "created_at": now})
|
|
.execute()
|
|
)
|
|
if not result.data:
|
|
raise Exception("Failed to add favorite")
|
|
return result.data[0]
|
|
|
|
def remove_favorite(self, user_id: str, listing_id: str) -> dict:
|
|
result = (
|
|
self.db.table("favorites")
|
|
.delete()
|
|
.eq("user_id", user_id)
|
|
.eq("listing_id", listing_id)
|
|
.execute()
|
|
)
|
|
if not result.data:
|
|
raise NotFoundException("Favorite not found")
|
|
return {"message": "Removed from favorites"}
|
|
|
|
def list_favorites(self, user_id: str) -> dict:
|
|
result = (
|
|
self.db.table("favorites")
|
|
.select(
|
|
"*, listings("
|
|
"id, title, description, price, images, location, status, "
|
|
"agency_id, category_id, listing_type, condition, negotiable, "
|
|
"views_count, created_at, updated_at"
|
|
")",
|
|
count="exact",
|
|
)
|
|
.eq("user_id", user_id)
|
|
.order("created_at", desc=True)
|
|
.execute()
|
|
)
|
|
return {"favorites": result.data, "total": result.count or 0}
|
|
|
|
def is_favorited(self, user_id: str, listing_id: str) -> bool:
|
|
result = (
|
|
self.db.table("favorites")
|
|
.select("id")
|
|
.eq("user_id", user_id)
|
|
.eq("listing_id", listing_id)
|
|
.execute()
|
|
)
|
|
return len(result.data) > 0
|