473 lines
16 KiB
Python
473 lines
16 KiB
Python
"""
|
|
Supabase database service for Odoo Directory.
|
|
Handles all database operations.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Optional, List, Dict, Any
|
|
from uuid import UUID
|
|
|
|
from supabase import create_client, Client
|
|
|
|
from ..config import get_settings
|
|
from ..models.schemas import (
|
|
Region,
|
|
Country,
|
|
City,
|
|
Tag,
|
|
Company,
|
|
CompanyTag,
|
|
CityTag,
|
|
CountryTag,
|
|
ExecutionPlan,
|
|
PlanStep,
|
|
TagCategory,
|
|
StepStatus,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class SupabaseService:
|
|
"""Service for Supabase database operations."""
|
|
|
|
def __init__(self):
|
|
"""Initialize Supabase client."""
|
|
settings = get_settings()
|
|
self.client: Client = create_client(
|
|
settings.supabase_url, settings.supabase_key
|
|
)
|
|
|
|
# ==================== Region Operations ====================
|
|
|
|
def create_region(self, region: Region) -> Region:
|
|
"""Create a new region."""
|
|
data = {"name": region.name, "slug": region.slug}
|
|
result = self.client.table("regions").insert(data).execute()
|
|
return Region(**result.data[0])
|
|
|
|
def get_region_by_slug(self, slug: str) -> Optional[Region]:
|
|
"""Get region by slug."""
|
|
result = (
|
|
self.client.table("regions").select("*").eq("slug", slug).execute()
|
|
)
|
|
if result.data:
|
|
return Region(**result.data[0])
|
|
return None
|
|
|
|
def list_regions(self) -> List[Region]:
|
|
"""List all regions."""
|
|
result = self.client.table("regions").select("*").execute()
|
|
return [Region(**r) for r in result.data]
|
|
|
|
# ==================== Country Operations ====================
|
|
|
|
def create_country(self, country: Country) -> Country:
|
|
"""Create a new country."""
|
|
data = {
|
|
"region_id": str(country.region_id) if country.region_id else None,
|
|
"name": country.name,
|
|
"name_local": country.name_local,
|
|
"code": country.code,
|
|
"slug": country.slug,
|
|
"language": country.language,
|
|
}
|
|
result = self.client.table("countries").insert(data).execute()
|
|
return Country(**result.data[0])
|
|
|
|
def get_country_by_code(self, code: str) -> Optional[Country]:
|
|
"""Get country by ISO code."""
|
|
result = (
|
|
self.client.table("countries")
|
|
.select("*")
|
|
.eq("code", code.upper())
|
|
.execute()
|
|
)
|
|
if result.data:
|
|
return Country(**result.data[0])
|
|
return None
|
|
|
|
def get_country_by_slug(self, slug: str) -> Optional[Country]:
|
|
"""Get country by slug."""
|
|
result = (
|
|
self.client.table("countries").select("*").eq("slug", slug).execute()
|
|
)
|
|
if result.data:
|
|
return Country(**result.data[0])
|
|
return None
|
|
|
|
def list_countries(self, region_id: Optional[UUID] = None) -> List[Country]:
|
|
"""List countries, optionally filtered by region."""
|
|
query = self.client.table("countries").select("*")
|
|
if region_id:
|
|
query = query.eq("region_id", str(region_id))
|
|
result = query.execute()
|
|
return [Country(**c) for c in result.data]
|
|
|
|
# ==================== City Operations ====================
|
|
|
|
def create_city(self, city: City) -> City:
|
|
"""Create a new city."""
|
|
data = {
|
|
"country_id": str(city.country_id) if city.country_id else None,
|
|
"name": city.name,
|
|
"name_local": city.name_local,
|
|
"slug": city.slug,
|
|
"state": city.state,
|
|
"population": city.population,
|
|
"latitude": city.latitude,
|
|
"longitude": city.longitude,
|
|
}
|
|
result = self.client.table("cities").insert(data).execute()
|
|
return City(**result.data[0])
|
|
|
|
def get_city_by_slug(
|
|
self, slug: str, country_id: Optional[UUID] = None
|
|
) -> Optional[City]:
|
|
"""Get city by slug, optionally filtered by country."""
|
|
query = self.client.table("cities").select("*").eq("slug", slug)
|
|
if country_id:
|
|
query = query.eq("country_id", str(country_id))
|
|
result = query.execute()
|
|
if result.data:
|
|
return City(**result.data[0])
|
|
return None
|
|
|
|
def list_cities(self, country_id: Optional[UUID] = None) -> List[City]:
|
|
"""List cities, optionally filtered by country."""
|
|
query = self.client.table("cities").select("*")
|
|
if country_id:
|
|
query = query.eq("country_id", str(country_id))
|
|
result = query.execute()
|
|
return [City(**c) for c in result.data]
|
|
|
|
# ==================== Tag Operations ====================
|
|
|
|
def create_tag(self, tag: Tag) -> Tag:
|
|
"""Create a new tag."""
|
|
data = {
|
|
"category": tag.category.value,
|
|
"name": tag.name,
|
|
"slug": tag.slug,
|
|
"description": tag.description,
|
|
}
|
|
result = self.client.table("tags").insert(data).execute()
|
|
return Tag(**result.data[0])
|
|
|
|
def get_tag_by_slug(self, slug: str) -> Optional[Tag]:
|
|
"""Get tag by slug."""
|
|
result = self.client.table("tags").select("*").eq("slug", slug).execute()
|
|
if result.data:
|
|
return Tag(**result.data[0])
|
|
return None
|
|
|
|
def list_tags(self, category: Optional[TagCategory] = None) -> List[Tag]:
|
|
"""List tags, optionally filtered by category."""
|
|
query = self.client.table("tags").select("*")
|
|
if category:
|
|
query = query.eq("category", category.value)
|
|
result = query.execute()
|
|
return [Tag(**t) for t in result.data]
|
|
|
|
def get_or_create_tag(
|
|
self, name: str, category: TagCategory, description: Optional[str] = None
|
|
) -> Tag:
|
|
"""Get existing tag or create new one."""
|
|
slug = name.lower().replace(" ", "-").replace("&", "and")
|
|
existing = self.get_tag_by_slug(slug)
|
|
if existing:
|
|
return existing
|
|
return self.create_tag(
|
|
Tag(category=category, name=name, slug=slug, description=description)
|
|
)
|
|
|
|
# ==================== Company Operations ====================
|
|
|
|
def create_company(self, company: Company) -> Company:
|
|
"""Create a new company."""
|
|
data = {
|
|
"city_id": str(company.city_id) if company.city_id else None,
|
|
"name": company.name,
|
|
"slug": company.slug,
|
|
"google_place_id": company.google_place_id,
|
|
"address": company.address,
|
|
"phone": company.phone,
|
|
"website": company.website,
|
|
"email": company.email,
|
|
"rating": company.rating,
|
|
"review_count": company.review_count,
|
|
"latitude": company.latitude,
|
|
"longitude": company.longitude,
|
|
"status": company.status,
|
|
}
|
|
result = self.client.table("companies").insert(data).execute()
|
|
return Company(**result.data[0])
|
|
|
|
def update_company(self, company_id: UUID, updates: Dict[str, Any]) -> Company:
|
|
"""Update a company."""
|
|
result = (
|
|
self.client.table("companies")
|
|
.update(updates)
|
|
.eq("id", str(company_id))
|
|
.execute()
|
|
)
|
|
return Company(**result.data[0])
|
|
|
|
def get_company_by_id(self, company_id: UUID) -> Optional[Company]:
|
|
"""Get company by ID."""
|
|
result = (
|
|
self.client.table("companies")
|
|
.select("*")
|
|
.eq("id", str(company_id))
|
|
.execute()
|
|
)
|
|
if result.data:
|
|
return Company(**result.data[0])
|
|
return None
|
|
|
|
def get_company_by_place_id(self, place_id: str) -> Optional[Company]:
|
|
"""Get company by Google Place ID (for deduplication)."""
|
|
result = (
|
|
self.client.table("companies")
|
|
.select("*")
|
|
.eq("google_place_id", place_id)
|
|
.execute()
|
|
)
|
|
if result.data:
|
|
return Company(**result.data[0])
|
|
return None
|
|
|
|
def list_companies(
|
|
self,
|
|
city_id: Optional[UUID] = None,
|
|
status: Optional[str] = None,
|
|
limit: int = 100,
|
|
offset: int = 0,
|
|
) -> List[Company]:
|
|
"""List companies with optional filters."""
|
|
query = self.client.table("companies").select("*")
|
|
if city_id:
|
|
query = query.eq("city_id", str(city_id))
|
|
if status:
|
|
query = query.eq("status", status)
|
|
query = query.range(offset, offset + limit - 1)
|
|
result = query.execute()
|
|
return [Company(**c) for c in result.data]
|
|
|
|
def count_companies(
|
|
self, city_id: Optional[UUID] = None, status: Optional[str] = None
|
|
) -> int:
|
|
"""Count companies with optional filters."""
|
|
query = self.client.table("companies").select("id", count="exact")
|
|
if city_id:
|
|
query = query.eq("city_id", str(city_id))
|
|
if status:
|
|
query = query.eq("status", status)
|
|
result = query.execute()
|
|
return result.count or 0
|
|
|
|
# ==================== Company Tag Operations ====================
|
|
|
|
def add_company_tag(
|
|
self, company_id: UUID, tag_id: UUID, confidence: float = 1.0
|
|
) -> CompanyTag:
|
|
"""Add a tag to a company."""
|
|
data = {
|
|
"company_id": str(company_id),
|
|
"tag_id": str(tag_id),
|
|
"confidence": confidence,
|
|
}
|
|
result = self.client.table("company_tags").insert(data).execute()
|
|
return CompanyTag(**result.data[0])
|
|
|
|
def get_company_tags(self, company_id: UUID) -> List[Tag]:
|
|
"""Get all tags for a company."""
|
|
result = (
|
|
self.client.table("company_tags")
|
|
.select("tag_id, tags(*)")
|
|
.eq("company_id", str(company_id))
|
|
.execute()
|
|
)
|
|
tags = []
|
|
for item in result.data:
|
|
if item.get("tags"):
|
|
tags.append(Tag(**item["tags"]))
|
|
return tags
|
|
|
|
def remove_company_tags(self, company_id: UUID) -> None:
|
|
"""Remove all tags from a company."""
|
|
self.client.table("company_tags").delete().eq(
|
|
"company_id", str(company_id)
|
|
).execute()
|
|
|
|
# ==================== City Tag Aggregation ====================
|
|
|
|
def upsert_city_tag(self, city_id: UUID, tag_id: UUID, count: int) -> CityTag:
|
|
"""Upsert city tag aggregation."""
|
|
data = {
|
|
"city_id": str(city_id),
|
|
"tag_id": str(tag_id),
|
|
"company_count": count,
|
|
}
|
|
result = (
|
|
self.client.table("city_tags")
|
|
.upsert(data, on_conflict="city_id,tag_id")
|
|
.execute()
|
|
)
|
|
return CityTag(**result.data[0])
|
|
|
|
def get_city_tags(self, city_id: UUID) -> List[Dict[str, Any]]:
|
|
"""Get all tag aggregations for a city with tag details."""
|
|
result = (
|
|
self.client.table("city_tags")
|
|
.select("*, tags(*)")
|
|
.eq("city_id", str(city_id))
|
|
.execute()
|
|
)
|
|
return result.data
|
|
|
|
# ==================== Country Tag Aggregation ====================
|
|
|
|
def upsert_country_tag(
|
|
self, country_id: UUID, tag_id: UUID, company_count: int, city_count: int
|
|
) -> CountryTag:
|
|
"""Upsert country tag aggregation."""
|
|
data = {
|
|
"country_id": str(country_id),
|
|
"tag_id": str(tag_id),
|
|
"company_count": company_count,
|
|
"city_count": city_count,
|
|
}
|
|
result = (
|
|
self.client.table("country_tags")
|
|
.upsert(data, on_conflict="country_id,tag_id")
|
|
.execute()
|
|
)
|
|
return CountryTag(**result.data[0])
|
|
|
|
def get_country_tags(self, country_id: UUID) -> List[Dict[str, Any]]:
|
|
"""Get all tag aggregations for a country with tag details."""
|
|
result = (
|
|
self.client.table("country_tags")
|
|
.select("*, tags(*)")
|
|
.eq("country_id", str(country_id))
|
|
.execute()
|
|
)
|
|
return result.data
|
|
|
|
# ==================== Execution Plan Operations ====================
|
|
|
|
def create_execution_plan(self, plan: ExecutionPlan) -> ExecutionPlan:
|
|
"""Create a new execution plan."""
|
|
data = {
|
|
"name": plan.name,
|
|
"country_code": plan.country_code,
|
|
"cities": plan.cities,
|
|
"status": plan.status.value,
|
|
}
|
|
result = self.client.table("execution_plans").insert(data).execute()
|
|
return ExecutionPlan(**result.data[0])
|
|
|
|
def update_execution_plan(
|
|
self, plan_id: UUID, updates: Dict[str, Any]
|
|
) -> ExecutionPlan:
|
|
"""Update an execution plan."""
|
|
result = (
|
|
self.client.table("execution_plans")
|
|
.update(updates)
|
|
.eq("id", str(plan_id))
|
|
.execute()
|
|
)
|
|
return ExecutionPlan(**result.data[0])
|
|
|
|
def get_execution_plan(self, plan_id: UUID) -> Optional[ExecutionPlan]:
|
|
"""Get execution plan by ID."""
|
|
result = (
|
|
self.client.table("execution_plans")
|
|
.select("*")
|
|
.eq("id", str(plan_id))
|
|
.execute()
|
|
)
|
|
if result.data:
|
|
plan = ExecutionPlan(**result.data[0])
|
|
# Load steps
|
|
steps_result = (
|
|
self.client.table("plan_steps")
|
|
.select("*")
|
|
.eq("plan_id", str(plan_id))
|
|
.execute()
|
|
)
|
|
plan.steps = [PlanStep(**s) for s in steps_result.data]
|
|
return plan
|
|
return None
|
|
|
|
def list_execution_plans(
|
|
self, status: Optional[StepStatus] = None
|
|
) -> List[ExecutionPlan]:
|
|
"""List execution plans."""
|
|
query = self.client.table("execution_plans").select("*")
|
|
if status:
|
|
query = query.eq("status", status.value)
|
|
result = query.execute()
|
|
return [ExecutionPlan(**p) for p in result.data]
|
|
|
|
# ==================== Plan Step Operations ====================
|
|
|
|
def create_plan_step(self, step: PlanStep) -> PlanStep:
|
|
"""Create a new plan step."""
|
|
data = {
|
|
"plan_id": str(step.plan_id) if step.plan_id else None,
|
|
"step_name": step.step_name,
|
|
"city_id": str(step.city_id) if step.city_id else None,
|
|
"status": step.status.value,
|
|
"items_processed": step.items_processed,
|
|
"items_total": step.items_total,
|
|
}
|
|
result = self.client.table("plan_steps").insert(data).execute()
|
|
return PlanStep(**result.data[0])
|
|
|
|
def update_plan_step(self, step_id: UUID, updates: Dict[str, Any]) -> PlanStep:
|
|
"""Update a plan step."""
|
|
result = (
|
|
self.client.table("plan_steps")
|
|
.update(updates)
|
|
.eq("id", str(step_id))
|
|
.execute()
|
|
)
|
|
return PlanStep(**result.data[0])
|
|
|
|
def get_plan_steps(
|
|
self, plan_id: UUID, step_name: Optional[str] = None
|
|
) -> List[PlanStep]:
|
|
"""Get steps for a plan."""
|
|
query = (
|
|
self.client.table("plan_steps").select("*").eq("plan_id", str(plan_id))
|
|
)
|
|
if step_name:
|
|
query = query.eq("step_name", step_name)
|
|
result = query.execute()
|
|
return [PlanStep(**s) for s in result.data]
|
|
|
|
# ==================== Error Logging ====================
|
|
|
|
def log_error(
|
|
self,
|
|
entity_type: str,
|
|
entity_id: Optional[UUID],
|
|
operation: str,
|
|
error_message: str,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
) -> None:
|
|
"""Log an error to the database."""
|
|
data = {
|
|
"entity_type": entity_type,
|
|
"entity_id": str(entity_id) if entity_id else None,
|
|
"operation": operation,
|
|
"error_message": error_message,
|
|
"details": details or {},
|
|
}
|
|
try:
|
|
self.client.table("error_logs").insert(data).execute()
|
|
except Exception as e:
|
|
logger.error(f"Failed to log error to database: {e}")
|