358 lines
12 KiB
Python
358 lines
12 KiB
Python
# app/services/shop_service.py
|
|
"""
|
|
Shop service for managing shop operations and product catalog.
|
|
|
|
This module provides classes and functions for:
|
|
- Shop creation and management
|
|
- Shop access control and validation
|
|
- Shop product catalog operations
|
|
- Shop filtering and search
|
|
"""
|
|
|
|
import logging
|
|
from typing import List, Optional, Tuple
|
|
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.exceptions import (
|
|
ShopNotFoundException,
|
|
ShopAlreadyExistsException,
|
|
UnauthorizedShopAccessException,
|
|
InvalidShopDataException,
|
|
ProductNotFoundException,
|
|
ShopProductAlreadyExistsException,
|
|
MaxShopsReachedException,
|
|
ValidationException,
|
|
)
|
|
from models.schemas.shop import ShopCreate, ShopProductCreate
|
|
from models.database.product import Product
|
|
from models.database.shop import Shop, ShopProduct
|
|
from models.database.user import User
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ShopService:
|
|
"""Service class for shop operations following the application's service pattern."""
|
|
|
|
def create_shop(
|
|
self, db: Session, shop_data: ShopCreate, current_user: User
|
|
) -> Shop:
|
|
"""
|
|
Create a new shop.
|
|
|
|
Args:
|
|
db: Database session
|
|
shop_data: Shop creation data
|
|
current_user: User creating the shop
|
|
|
|
Returns:
|
|
Created shop object
|
|
|
|
Raises:
|
|
ShopAlreadyExistsException: If shop code already exists
|
|
MaxShopsReachedException: If user has reached maximum shops
|
|
InvalidShopDataException: If shop data is invalid
|
|
"""
|
|
try:
|
|
# Validate shop data
|
|
self._validate_shop_data(shop_data)
|
|
|
|
# Check user's shop limit (if applicable)
|
|
self._check_shop_limit(db, current_user)
|
|
|
|
# Normalize shop code to uppercase
|
|
normalized_shop_code = shop_data.shop_code.upper()
|
|
|
|
# Check if shop code already exists (case-insensitive check)
|
|
if self._shop_code_exists(db, normalized_shop_code):
|
|
raise ShopAlreadyExistsException(normalized_shop_code)
|
|
|
|
# Create shop with uppercase code
|
|
shop_dict = shop_data.model_dump()
|
|
shop_dict["shop_code"] = normalized_shop_code # Store as uppercase
|
|
|
|
new_shop = Shop(
|
|
**shop_dict,
|
|
owner_id=current_user.id,
|
|
is_active=True,
|
|
is_verified=(current_user.role == "admin"),
|
|
)
|
|
|
|
db.add(new_shop)
|
|
db.commit()
|
|
db.refresh(new_shop)
|
|
|
|
logger.info(
|
|
f"New shop created: {new_shop.shop_code} by {current_user.username}"
|
|
)
|
|
return new_shop
|
|
|
|
except (ShopAlreadyExistsException, MaxShopsReachedException, InvalidShopDataException):
|
|
db.rollback()
|
|
raise # Re-raise custom exceptions
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"Error creating shop: {str(e)}")
|
|
raise ValidationException("Failed to create shop")
|
|
|
|
def get_shops(
|
|
self,
|
|
db: Session,
|
|
current_user: User,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
active_only: bool = True,
|
|
verified_only: bool = False,
|
|
) -> Tuple[List[Shop], int]:
|
|
"""
|
|
Get shops with filtering.
|
|
|
|
Args:
|
|
db: Database session
|
|
current_user: Current user requesting shops
|
|
skip: Number of records to skip
|
|
limit: Maximum number of records to return
|
|
active_only: Filter for active shops only
|
|
verified_only: Filter for verified shops only
|
|
|
|
Returns:
|
|
Tuple of (shops_list, total_count)
|
|
"""
|
|
try:
|
|
query = db.query(Shop)
|
|
|
|
# Non-admin users can only see active and verified shops, plus their own
|
|
if current_user.role != "admin":
|
|
query = query.filter(
|
|
(Shop.is_active == True)
|
|
& ((Shop.is_verified == True) | (Shop.owner_id == current_user.id))
|
|
)
|
|
else:
|
|
# Admin can apply filters
|
|
if active_only:
|
|
query = query.filter(Shop.is_active == True)
|
|
if verified_only:
|
|
query = query.filter(Shop.is_verified == True)
|
|
|
|
total = query.count()
|
|
shops = query.offset(skip).limit(limit).all()
|
|
|
|
return shops, total
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting shops: {str(e)}")
|
|
raise ValidationException("Failed to retrieve shops")
|
|
|
|
def get_shop_by_code(self, db: Session, shop_code: str, current_user: User) -> Shop:
|
|
"""
|
|
Get shop by shop code with access control.
|
|
|
|
Args:
|
|
db: Database session
|
|
shop_code: Shop code to find
|
|
current_user: Current user requesting the shop
|
|
|
|
Returns:
|
|
Shop object
|
|
|
|
Raises:
|
|
ShopNotFoundException: If shop not found
|
|
UnauthorizedShopAccessException: If access denied
|
|
"""
|
|
try:
|
|
shop = (
|
|
db.query(Shop)
|
|
.filter(func.upper(Shop.shop_code) == shop_code.upper())
|
|
.first()
|
|
)
|
|
|
|
if not shop:
|
|
raise ShopNotFoundException(shop_code)
|
|
|
|
# Check access permissions
|
|
if not self._can_access_shop(shop, current_user):
|
|
raise UnauthorizedShopAccessException(shop_code, current_user.id)
|
|
|
|
return shop
|
|
|
|
except (ShopNotFoundException, UnauthorizedShopAccessException):
|
|
raise # Re-raise custom exceptions
|
|
except Exception as e:
|
|
logger.error(f"Error getting shop {shop_code}: {str(e)}")
|
|
raise ValidationException("Failed to retrieve shop")
|
|
|
|
def add_product_to_shop(
|
|
self, db: Session, shop: Shop, shop_product: ShopProductCreate
|
|
) -> ShopProduct:
|
|
"""
|
|
Add existing product to shop catalog with shop-specific settings.
|
|
|
|
Args:
|
|
db: Database session
|
|
shop: Shop to add product to
|
|
shop_product: Shop product data
|
|
|
|
Returns:
|
|
Created ShopProduct object
|
|
|
|
Raises:
|
|
ProductNotFoundException: If product not found
|
|
ShopProductAlreadyExistsException: If product already in shop
|
|
"""
|
|
try:
|
|
# Check if product exists
|
|
product = self._get_product_by_id_or_raise(db, shop_product.product_id)
|
|
|
|
# Check if product already in shop
|
|
if self._product_in_shop(db, shop.id, product.id):
|
|
raise ShopProductAlreadyExistsException(shop.shop_code, shop_product.product_id)
|
|
|
|
# Create shop-product association
|
|
new_shop_product = ShopProduct(
|
|
shop_id=shop.id,
|
|
product_id=product.id,
|
|
**shop_product.model_dump(exclude={"product_id"}),
|
|
)
|
|
|
|
db.add(new_shop_product)
|
|
db.commit()
|
|
db.refresh(new_shop_product)
|
|
|
|
# Load the product relationship
|
|
db.refresh(new_shop_product)
|
|
|
|
logger.info(f"Product {shop_product.product_id} added to shop {shop.shop_code}")
|
|
return new_shop_product
|
|
|
|
except (ProductNotFoundException, ShopProductAlreadyExistsException):
|
|
db.rollback()
|
|
raise # Re-raise custom exceptions
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"Error adding product to shop: {str(e)}")
|
|
raise ValidationException("Failed to add product to shop")
|
|
|
|
def get_shop_products(
|
|
self,
|
|
db: Session,
|
|
shop: Shop,
|
|
current_user: User,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
active_only: bool = True,
|
|
featured_only: bool = False,
|
|
) -> Tuple[List[ShopProduct], int]:
|
|
"""
|
|
Get products in shop catalog with filtering.
|
|
|
|
Args:
|
|
db: Database session
|
|
shop: Shop to get products from
|
|
current_user: Current user requesting products
|
|
skip: Number of records to skip
|
|
limit: Maximum number of records to return
|
|
active_only: Filter for active products only
|
|
featured_only: Filter for featured products only
|
|
|
|
Returns:
|
|
Tuple of (shop_products_list, total_count)
|
|
|
|
Raises:
|
|
UnauthorizedShopAccessException: If shop access denied
|
|
"""
|
|
try:
|
|
# Check access permissions
|
|
if not self._can_access_shop(shop, current_user):
|
|
raise UnauthorizedShopAccessException(shop.shop_code, current_user.id)
|
|
|
|
# Query shop products
|
|
query = db.query(ShopProduct).filter(ShopProduct.shop_id == shop.id)
|
|
|
|
if active_only:
|
|
query = query.filter(ShopProduct.is_active == True)
|
|
if featured_only:
|
|
query = query.filter(ShopProduct.is_featured == True)
|
|
|
|
total = query.count()
|
|
shop_products = query.offset(skip).limit(limit).all()
|
|
|
|
return shop_products, total
|
|
|
|
except UnauthorizedShopAccessException:
|
|
raise # Re-raise custom exceptions
|
|
except Exception as e:
|
|
logger.error(f"Error getting shop products: {str(e)}")
|
|
raise ValidationException("Failed to retrieve shop products")
|
|
|
|
# Private helper methods
|
|
def _validate_shop_data(self, shop_data: ShopCreate) -> None:
|
|
"""Validate shop creation data."""
|
|
if not shop_data.shop_code or not shop_data.shop_code.strip():
|
|
raise InvalidShopDataException("Shop code is required", field="shop_code")
|
|
|
|
if not shop_data.shop_name or not shop_data.shop_name.strip():
|
|
raise InvalidShopDataException("Shop name is required", field="shop_name")
|
|
|
|
# Validate shop code format (alphanumeric, underscores, hyphens)
|
|
import re
|
|
if not re.match(r'^[A-Za-z0-9_-]+$', shop_data.shop_code):
|
|
raise InvalidShopDataException(
|
|
"Shop code can only contain letters, numbers, underscores, and hyphens",
|
|
field="shop_code"
|
|
)
|
|
|
|
def _check_shop_limit(self, db: Session, user: User) -> None:
|
|
"""Check if user has reached maximum shop limit."""
|
|
if user.role == "admin":
|
|
return # Admins have no limit
|
|
|
|
user_shop_count = db.query(Shop).filter(Shop.owner_id == user.id).count()
|
|
max_shops = 5 # Configure this as needed
|
|
|
|
if user_shop_count >= max_shops:
|
|
raise MaxShopsReachedException(max_shops, user.id)
|
|
|
|
def _shop_code_exists(self, db: Session, shop_code: str) -> bool:
|
|
"""Check if shop code already exists (case-insensitive)."""
|
|
return (
|
|
db.query(Shop)
|
|
.filter(func.upper(Shop.shop_code) == shop_code.upper())
|
|
.first() is not None
|
|
)
|
|
|
|
def _get_product_by_id_or_raise(self, db: Session, product_id: str) -> Product:
|
|
"""Get product by ID or raise exception."""
|
|
product = db.query(Product).filter(Product.product_id == product_id).first()
|
|
if not product:
|
|
raise ProductNotFoundException(product_id)
|
|
return product
|
|
|
|
def _product_in_shop(self, db: Session, shop_id: int, product_id: int) -> bool:
|
|
"""Check if product is already in shop."""
|
|
return (
|
|
db.query(ShopProduct)
|
|
.filter(
|
|
ShopProduct.shop_id == shop_id,
|
|
ShopProduct.product_id == product_id
|
|
)
|
|
.first() is not None
|
|
)
|
|
|
|
def _can_access_shop(self, shop: Shop, user: User) -> bool:
|
|
"""Check if user can access shop."""
|
|
# Admins and owners can always access
|
|
if user.role == "admin" or shop.owner_id == user.id:
|
|
return True
|
|
|
|
# Others can only access active and verified shops
|
|
return shop.is_active and shop.is_verified
|
|
|
|
def _is_shop_owner(self, shop: Shop, user: User) -> bool:
|
|
"""Check if user is shop owner."""
|
|
return shop.owner_id == user.id
|
|
|
|
# Create service instance following the same pattern as other services
|
|
shop_service = ShopService()
|