mirror of
http://88.130.71.182:3000/BlitTech/deals24togo_be.git
synced 2026-06-12 23:33:21 +00:00
135 lines
4.7 KiB
Python
135 lines
4.7 KiB
Python
"""Agency CRUD service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
|
|
from app.core.exceptions import ForbiddenException, NotFoundException
|
|
from app.core.supabase import get_supabase_admin
|
|
|
|
|
|
class AgencyService:
|
|
def __init__(self):
|
|
self.db = get_supabase_admin()
|
|
|
|
def get_agency(self, agency_id: str) -> dict:
|
|
result = self.db.table("agencies").select("*").eq("id", agency_id).execute()
|
|
if not result.data:
|
|
raise NotFoundException("Agency not found")
|
|
return result.data[0]
|
|
|
|
def get_agency_by_user(self, user_id: str) -> dict:
|
|
result = self.db.table("agencies").select("*").eq("user_id", user_id).execute()
|
|
if not result.data:
|
|
raise NotFoundException("Agency not found for this user")
|
|
return result.data[0]
|
|
|
|
def create_agency(self, user_id: str, data: dict) -> dict:
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
agency_data = {
|
|
"user_id": user_id,
|
|
**data,
|
|
"verified": False,
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
}
|
|
result = self.db.table("agencies").insert(agency_data).execute()
|
|
if not result.data:
|
|
raise Exception("Failed to create agency")
|
|
return result.data[0]
|
|
|
|
def update_agency(self, agency_id: str, user_id: str, user_role: str, data: dict) -> dict:
|
|
# Check ownership or admin
|
|
agency = self.get_agency(agency_id)
|
|
if user_role != "admin" and agency["user_id"] != user_id:
|
|
raise ForbiddenException("Not authorized to update this agency")
|
|
|
|
update_data = {k: v for k, v in data.items() if v is not None}
|
|
if not update_data:
|
|
return agency
|
|
|
|
update_data["updated_at"] = datetime.now(timezone.utc).isoformat()
|
|
result = (
|
|
self.db.table("agencies")
|
|
.update(update_data)
|
|
.eq("id", agency_id)
|
|
.execute()
|
|
)
|
|
if not result.data:
|
|
raise NotFoundException("Agency not found")
|
|
return result.data[0]
|
|
|
|
def list_agencies(
|
|
self,
|
|
page: int = 1,
|
|
page_size: int = 20,
|
|
verified_only: bool = False,
|
|
) -> dict:
|
|
query = self.db.table("agencies").select("*", count="exact")
|
|
if verified_only:
|
|
query = query.eq("verified", True)
|
|
|
|
offset = (page - 1) * page_size
|
|
result = (
|
|
query.order("created_at", desc=True)
|
|
.range(offset, offset + page_size - 1)
|
|
.execute()
|
|
)
|
|
|
|
return {
|
|
"agencies": result.data,
|
|
"total": result.count or 0,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
}
|
|
|
|
def verify_agency(self, agency_id: str, requester_role: str) -> dict:
|
|
if requester_role != "admin":
|
|
raise ForbiddenException("Only admins can verify agencies")
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
result = (
|
|
self.db.table("agencies")
|
|
.update({"verified": True, "updated_at": now})
|
|
.eq("id", agency_id)
|
|
.execute()
|
|
)
|
|
if not result.data:
|
|
raise NotFoundException("Agency not found")
|
|
|
|
# Also verify the associated user
|
|
agency = result.data[0]
|
|
self.db.table("users").update(
|
|
{"verified": True, "updated_at": now}
|
|
).eq("id", agency["user_id"]).execute()
|
|
|
|
return result.data[0]
|
|
|
|
def revoke_verification(self, agency_id: str, requester_role: str) -> dict:
|
|
if requester_role != "admin":
|
|
raise ForbiddenException("Only admins can revoke agency verification")
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
result = (
|
|
self.db.table("agencies")
|
|
.update({"verified": False, "updated_at": now})
|
|
.eq("id", agency_id)
|
|
.execute()
|
|
)
|
|
if not result.data:
|
|
raise NotFoundException("Agency not found")
|
|
return result.data[0]
|
|
|
|
def delete_agency(self, agency_id: str) -> dict:
|
|
# Fetch first to get user_id for cascade deletion
|
|
agency_result = self.db.table("agencies").select("user_id").eq("id", agency_id).execute()
|
|
if not agency_result.data:
|
|
raise NotFoundException("Agency not found")
|
|
user_id = agency_result.data[0]["user_id"]
|
|
|
|
# Delete messages belonging to this agency (avoids orphaned rows)
|
|
self.db.table("messages").delete().eq("agency_id", agency_id).execute()
|
|
self.db.table("agencies").delete().eq("id", agency_id).execute()
|
|
# Delete the associated user account to avoid orphaned records
|
|
self.db.table("users").delete().eq("id", user_id).execute()
|
|
return {"message": "Agency deleted"}
|