from sqlalchemy import func from sqlalchemy.orm import Session from fastapi import HTTPException from datetime import datetime import logging from typing import List, Optional, Tuple, Dict, Any from models.database_models import User, Shop, Product, ShopProduct from models.api_models import ShopCreate, ShopProductCreate 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: HTTPException: If shop code already exists """ # Normalize shop code to uppercase normalized_shop_code = shop_data.shop_code.upper() # Check if shop code already exists (case-insensitive check against existing data) existing_shop = db.query(Shop).filter( func.upper(Shop.shop_code) == normalized_shop_code ).first() if existing_shop: raise HTTPException(status_code=400, detail="Shop code already exists") # Create shop with uppercase code shop_dict = shop_data.model_dump() # Fixed deprecated .dict() method 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 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) """ 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 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: HTTPException: If shop not found or access denied """ # Explicit type hint to help type checker shop: Optional[Shop] shop: Optional[Shop] = db.query(Shop).filter(func.upper(Shop.shop_code) == shop_code.upper()).first() if not shop: raise HTTPException(status_code=404, detail="Shop not found") # Non-admin users can only see active verified shops or their own shops if current_user.role != "admin": if not shop.is_active or (not shop.is_verified and shop.owner_id != current_user.id): raise HTTPException(status_code=404, detail="Shop not found") return 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: HTTPException: If product not found or already in shop """ # Check if product exists product = db.query(Product).filter(Product.product_id == shop_product.product_id).first() if not product: raise HTTPException(status_code=404, detail="Product not found in marketplace catalog") # Check if product already in shop existing_shop_product = db.query(ShopProduct).filter( ShopProduct.shop_id == shop.id, ShopProduct.product_id == product.id ).first() if existing_shop_product: raise HTTPException(status_code=400, detail="Product already in shop catalog") # 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 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: HTTPException: If shop access denied """ # Non-owners can only see active verified shops if current_user.role != "admin" and shop.owner_id != current_user.id: if not shop.is_active or not shop.is_verified: raise HTTPException(status_code=404, detail="Shop not found") # 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 def get_shop_by_id(self, db: Session, shop_id: int) -> Optional[Shop]: """Get shop by ID""" return db.query(Shop).filter(Shop.id == shop_id).first() def shop_code_exists(self, db: Session, shop_code: str) -> bool: """Check if shop code already exists""" return db.query(Shop).filter(Shop.shop_code == shop_code).first() is not None def get_product_by_id(self, db: Session, product_id: str) -> Optional[Product]: """Get product by product_id""" return db.query(Product).filter(Product.product_id == product_id).first() 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 is_shop_owner(self, shop: Shop, user: User) -> bool: """Check if user is shop owner""" return shop.owner_id == user.id def can_view_shop(self, shop: Shop, user: User) -> bool: """Check if user can view shop""" if user.role == "admin" or self.is_shop_owner(shop, user): return True return shop.is_active and shop.is_verified # Create service instance following the same pattern as other services shop_service = ShopService()