Samir Boulahtit
508e121a0e
Change Product/ProductTranslation from "override/inheritance" pattern (NULL = inherit from marketplace) to "independent copy" pattern (all fields populated at creation). Key changes: - Remove OVERRIDABLE_FIELDS, effective_* properties, reset_* methods - Rename get_override_info() → get_source_comparison_info() - Update copy_to_vendor_catalog() to copy ALL fields + translations - Replace effective_* with direct field access in services - Remove *_overridden fields from schema, keep *_source for comparison - Add migration to populate NULL fields from marketplace products The marketplace_product_id FK is kept for "view original source" feature. Rollback tag: v1.0.0-pre-product-independence 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
293 lines
9.8 KiB
Python
293 lines
9.8 KiB
Python
# app/services/vendor_product_service.py
|
|
"""
|
|
Vendor product service for managing vendor-specific product catalogs.
|
|
|
|
This module provides:
|
|
- Vendor product catalog browsing
|
|
- Product search and filtering
|
|
- Product statistics
|
|
- Product removal from catalogs
|
|
"""
|
|
|
|
import logging
|
|
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
from app.exceptions import ProductNotFoundException
|
|
from models.database.product import Product
|
|
from models.database.vendor import Vendor
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class VendorProductService:
|
|
"""Service for vendor product catalog operations."""
|
|
|
|
def get_products(
|
|
self,
|
|
db: Session,
|
|
skip: int = 0,
|
|
limit: int = 50,
|
|
search: str | None = None,
|
|
vendor_id: int | None = None,
|
|
is_active: bool | None = None,
|
|
is_featured: bool | None = None,
|
|
language: str = "en",
|
|
) -> tuple[list[dict], int]:
|
|
"""
|
|
Get vendor products with search and filtering.
|
|
|
|
Returns:
|
|
Tuple of (products list as dicts, total count)
|
|
"""
|
|
query = (
|
|
db.query(Product)
|
|
.join(Vendor, Product.vendor_id == Vendor.id)
|
|
.options(
|
|
joinedload(Product.vendor),
|
|
joinedload(Product.marketplace_product),
|
|
)
|
|
)
|
|
|
|
if search:
|
|
search_term = f"%{search}%"
|
|
query = query.filter(Product.vendor_sku.ilike(search_term))
|
|
|
|
if vendor_id:
|
|
query = query.filter(Product.vendor_id == vendor_id)
|
|
|
|
if is_active is not None:
|
|
query = query.filter(Product.is_active == is_active)
|
|
|
|
if is_featured is not None:
|
|
query = query.filter(Product.is_featured == is_featured)
|
|
|
|
total = query.count()
|
|
|
|
products = (
|
|
query.order_by(Product.updated_at.desc()).offset(skip).limit(limit).all()
|
|
)
|
|
|
|
result = []
|
|
for product in products:
|
|
result.append(self._build_product_list_item(product, language))
|
|
|
|
return result, total
|
|
|
|
def get_product_stats(self, db: Session, vendor_id: int | None = None) -> dict:
|
|
"""Get vendor product statistics for admin dashboard.
|
|
|
|
Args:
|
|
db: Database session
|
|
vendor_id: Optional vendor ID to filter stats
|
|
|
|
Returns:
|
|
Dict with product counts (total, active, inactive, etc.)
|
|
"""
|
|
# Base query filter
|
|
base_filter = Product.vendor_id == vendor_id if vendor_id else True
|
|
|
|
total = db.query(func.count(Product.id)).filter(base_filter).scalar() or 0
|
|
|
|
active = (
|
|
db.query(func.count(Product.id))
|
|
.filter(base_filter)
|
|
.filter(Product.is_active == True) # noqa: E712
|
|
.scalar()
|
|
or 0
|
|
)
|
|
inactive = total - active
|
|
|
|
featured = (
|
|
db.query(func.count(Product.id))
|
|
.filter(base_filter)
|
|
.filter(Product.is_featured == True) # noqa: E712
|
|
.scalar()
|
|
or 0
|
|
)
|
|
|
|
# Digital/physical counts
|
|
digital = (
|
|
db.query(func.count(Product.id))
|
|
.filter(base_filter)
|
|
.join(Product.marketplace_product)
|
|
.filter(Product.marketplace_product.has(is_digital=True))
|
|
.scalar()
|
|
or 0
|
|
)
|
|
physical = total - digital
|
|
|
|
# Count by vendor (only when not filtered by vendor_id)
|
|
by_vendor = {}
|
|
if not vendor_id:
|
|
vendor_counts = (
|
|
db.query(
|
|
Vendor.name,
|
|
func.count(Product.id),
|
|
)
|
|
.join(Vendor, Product.vendor_id == Vendor.id)
|
|
.group_by(Vendor.name)
|
|
.all()
|
|
)
|
|
by_vendor = {name or "unknown": count for name, count in vendor_counts}
|
|
|
|
return {
|
|
"total": total,
|
|
"active": active,
|
|
"inactive": inactive,
|
|
"featured": featured,
|
|
"digital": digital,
|
|
"physical": physical,
|
|
"by_vendor": by_vendor,
|
|
}
|
|
|
|
def get_catalog_vendors(self, db: Session) -> list[dict]:
|
|
"""Get list of vendors with products in their catalogs."""
|
|
vendors = (
|
|
db.query(Vendor.id, Vendor.name, Vendor.vendor_code)
|
|
.join(Product, Vendor.id == Product.vendor_id)
|
|
.distinct()
|
|
.all()
|
|
)
|
|
return [
|
|
{"id": v.id, "name": v.name, "vendor_code": v.vendor_code} for v in vendors
|
|
]
|
|
|
|
def get_product_detail(self, db: Session, product_id: int) -> dict:
|
|
"""Get detailed vendor product information including override info."""
|
|
product = (
|
|
db.query(Product)
|
|
.options(
|
|
joinedload(Product.vendor),
|
|
joinedload(Product.marketplace_product),
|
|
joinedload(Product.translations),
|
|
)
|
|
.filter(Product.id == product_id)
|
|
.first()
|
|
)
|
|
|
|
if not product:
|
|
raise ProductNotFoundException(product_id)
|
|
|
|
mp = product.marketplace_product
|
|
source_comparison_info = product.get_source_comparison_info()
|
|
|
|
# Get marketplace product translations (for "view original source")
|
|
mp_translations = {}
|
|
if mp:
|
|
for t in mp.translations:
|
|
mp_translations[t.language] = {
|
|
"title": t.title,
|
|
"description": t.description,
|
|
"short_description": t.short_description,
|
|
}
|
|
|
|
# Get vendor translations
|
|
vendor_translations = {}
|
|
for t in product.translations:
|
|
vendor_translations[t.language] = {
|
|
"title": t.title,
|
|
"description": t.description,
|
|
}
|
|
|
|
return {
|
|
"id": product.id,
|
|
"vendor_id": product.vendor_id,
|
|
"vendor_name": product.vendor.name if product.vendor else None,
|
|
"vendor_code": product.vendor.vendor_code if product.vendor else None,
|
|
"marketplace_product_id": product.marketplace_product_id,
|
|
"vendor_sku": product.vendor_sku,
|
|
# Product fields with source comparison info
|
|
**source_comparison_info,
|
|
# Vendor-specific fields
|
|
"is_featured": product.is_featured,
|
|
"is_active": product.is_active,
|
|
"display_order": product.display_order,
|
|
"min_quantity": product.min_quantity,
|
|
"max_quantity": product.max_quantity,
|
|
# Supplier tracking
|
|
"supplier": product.supplier,
|
|
"supplier_product_id": product.supplier_product_id,
|
|
"cost": product.cost,
|
|
"margin_percent": product.margin_percent,
|
|
# Tax/profit info
|
|
"tax_rate_percent": product.tax_rate_percent,
|
|
"net_price": product.net_price,
|
|
"vat_amount": product.vat_amount,
|
|
"profit": product.profit,
|
|
"profit_margin_percent": product.profit_margin_percent,
|
|
# Digital fulfillment
|
|
"download_url": product.download_url,
|
|
"license_type": product.license_type,
|
|
"fulfillment_email_template": product.fulfillment_email_template,
|
|
# Source info from marketplace product
|
|
"source_marketplace": mp.marketplace if mp else None,
|
|
"source_vendor": mp.vendor_name if mp else None,
|
|
"source_gtin": mp.gtin if mp else None,
|
|
"source_sku": mp.sku if mp else None,
|
|
# Translations
|
|
"marketplace_translations": mp_translations,
|
|
"vendor_translations": vendor_translations,
|
|
# Timestamps
|
|
"created_at": product.created_at.isoformat()
|
|
if product.created_at
|
|
else None,
|
|
"updated_at": product.updated_at.isoformat()
|
|
if product.updated_at
|
|
else None,
|
|
}
|
|
|
|
def remove_product(self, db: Session, product_id: int) -> dict:
|
|
"""Remove a product from vendor catalog."""
|
|
product = db.query(Product).filter(Product.id == product_id).first()
|
|
|
|
if not product:
|
|
raise ProductNotFoundException(product_id)
|
|
|
|
vendor_name = product.vendor.name if product.vendor else "Unknown"
|
|
db.delete(product)
|
|
db.flush()
|
|
|
|
logger.info(f"Removed product {product_id} from vendor {vendor_name} catalog")
|
|
|
|
return {"message": f"Product removed from {vendor_name}'s catalog"}
|
|
|
|
def _build_product_list_item(self, product: Product, language: str) -> dict:
|
|
"""Build a product list item dict."""
|
|
mp = product.marketplace_product
|
|
|
|
# Get title from marketplace product translations
|
|
title = None
|
|
if mp:
|
|
title = mp.get_title(language)
|
|
|
|
return {
|
|
"id": product.id,
|
|
"vendor_id": product.vendor_id,
|
|
"vendor_name": product.vendor.name if product.vendor else None,
|
|
"vendor_code": product.vendor.vendor_code if product.vendor else None,
|
|
"marketplace_product_id": product.marketplace_product_id,
|
|
"vendor_sku": product.vendor_sku,
|
|
"title": title,
|
|
"brand": product.brand,
|
|
"price": product.price,
|
|
"currency": product.currency,
|
|
"is_active": product.is_active,
|
|
"is_featured": product.is_featured,
|
|
"is_digital": product.is_digital,
|
|
"image_url": product.primary_image_url,
|
|
"source_marketplace": mp.marketplace if mp else None,
|
|
"source_vendor": mp.vendor_name if mp else None,
|
|
"created_at": product.created_at.isoformat()
|
|
if product.created_at
|
|
else None,
|
|
"updated_at": product.updated_at.isoformat()
|
|
if product.updated_at
|
|
else None,
|
|
}
|
|
|
|
|
|
# Create service instance
|
|
vendor_product_service = VendorProductService()
|