sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Exception handling enhancement

Browse Source
This commit is contained in:
Samir Boulahtit
2025-09-23 22:42:26 +02:00
parent b1a76cdb57
commit 98285aa8aa
35 changed files with 3283 additions and 1743 deletions
Download Patch File Download Diff File Expand all files Collapse all files

411
app/services/shop_service.py
View File

@@ -1,19 +1,30 @@
# app/services/shop_service.py
"""Summary description ....
"""
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 fastapi import HTTPException
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
@@ -26,7 +37,7 @@ 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
self, db: Session, shop_data: ShopCreate, current_user: User
) -> Shop:
"""
Create a new shop.
@@ -40,49 +51,60 @@ class ShopService:
Created shop object
Raises:
HTTPException: If shop code already exists
ShopAlreadyExistsException: If shop code already exists
MaxShopsReachedException: If user has reached maximum shops
InvalidShopDataException: If shop data is invalid
"""
# Normalize shop code to uppercase
normalized_shop_code = shop_data.shop_code.upper()
try:
# Validate shop data
self._validate_shop_data(shop_data)
# 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()
)
# Check user's shop limit (if applicable)
self._check_shop_limit(db, current_user)
if existing_shop:
raise HTTPException(status_code=400, detail="Shop code already exists")
# Normalize shop code to uppercase
normalized_shop_code = shop_data.shop_code.upper()
# 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
# Check if shop code already exists (case-insensitive check)
if self._shop_code_exists(db, normalized_shop_code):
raise ShopAlreadyExistsException(normalized_shop_code)
new_shop = Shop(
**shop_dict,
owner_id=current_user.id,
is_active=True,
is_verified=(current_user.role == "admin"),
)
# Create shop with uppercase code
shop_dict = shop_data.model_dump()
shop_dict["shop_code"] = normalized_shop_code # Store as uppercase
db.add(new_shop)
db.commit()
db.refresh(new_shop)
new_shop = Shop(
**shop_dict,
owner_id=current_user.id,
is_active=True,
is_verified=(current_user.role == "admin"),
)
logger.info(
f"New shop created: {new_shop.shop_code} by {current_user.username}"
)
return new_shop
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,
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.
@@ -98,25 +120,30 @@ class ShopService:
Returns:
Tuple of (shops_list, total_count)
"""
query = db.query(Shop)
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)
# 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()
total = query.count()
shops = query.offset(skip).limit(limit).all()
return shops, total
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:
"""
@@ -131,28 +158,33 @@ class ShopService:
Shop object
Raises:
HTTPException: If shop not found or access denied
ShopNotFoundException: If shop not found
UnauthorizedShopAccessException: If 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")
try:
shop = (
db.query(Shop)
.filter(func.upper(Shop.shop_code) == shop_code.upper())
.first()
)
# 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")
if not shop:
raise ShopNotFoundException(shop_code)
return shop
# 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
self, db: Session, shop: Shop, shop_product: ShopProductCreate
) -> ShopProduct:
"""
Add existing product to shop catalog with shop-specific settings.
@@ -166,59 +198,51 @@ class ShopService:
Created ShopProduct object
Raises:
HTTPException: If product not found or already in shop
ProductNotFoundException: If product not found
ShopProductAlreadyExistsException: If product 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"
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"}),
)
# Check if product already in shop
existing_shop_product = (
db.query(ShopProduct)
.filter(
ShopProduct.shop_id == shop.id, ShopProduct.product_id == product.id
)
.first()
)
db.add(new_shop_product)
db.commit()
db.refresh(new_shop_product)
if existing_shop_product:
raise HTTPException(
status_code=400, detail="Product already in shop catalog"
)
# Load the product relationship
db.refresh(new_shop_product)
# Create shop-product association
new_shop_product = ShopProduct(
shop_id=shop.id,
product_id=product.id,
**shop_product.model_dump(exclude={"product_id"}),
)
logger.info(f"Product {shop_product.product_id} added to shop {shop.shop_code}")
return new_shop_product
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,
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.
@@ -236,58 +260,137 @@ class ShopService:
Tuple of (shop_products_list, total_count)
Raises:
HTTPException: If shop access denied
UnauthorizedShopAccessException: 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")
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)
# 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)
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()
total = query.count()
shop_products = query.offset(skip).limit(limit).all()
return shop_products, total
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()
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")
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
# 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")
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()
if not shop_data.shop_name or not shop_data.shop_name.strip():
raise InvalidShopDataException("Shop name is required", field="shop_name")
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
# 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"
)
.first()
is not None
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 is_shop_owner(self, shop: Shop, user: User) -> bool:
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
# Legacy methods for backward compatibility (deprecated)
def get_shop_by_id(self, db: Session, shop_id: int) -> Optional[Shop]:
"""Get shop by ID. DEPRECATED: Use proper exception handling."""
logger.warning("get_shop_by_id is deprecated, use proper exception handling")
try:
return db.query(Shop).filter(Shop.id == shop_id).first()
except Exception as e:
logger.error(f"Error getting shop by ID: {str(e)}")
return None
def shop_code_exists(self, db: Session, shop_code: str) -> bool:
"""Check if shop code exists. DEPRECATED: Use proper exception handling."""
logger.warning("shop_code_exists is deprecated, use proper exception handling")
return self._shop_code_exists(db, shop_code)
def get_product_by_id(self, db: Session, product_id: str) -> Optional[Product]:
"""Get product by ID. DEPRECATED: Use proper exception handling."""
logger.warning("get_product_by_id is deprecated, use proper exception handling")
try:
return db.query(Product).filter(Product.product_id == product_id).first()
except Exception as e:
logger.error(f"Error getting product by ID: {str(e)}")
return None
def product_in_shop(self, db: Session, shop_id: int, product_id: int) -> bool:
"""Check if product in shop. DEPRECATED: Use proper exception handling."""
logger.warning("product_in_shop is deprecated, use proper exception handling")
return self._product_in_shop(db, shop_id, product_id)
def is_shop_owner(self, shop: Shop, user: User) -> bool:
"""Check if user is shop owner. DEPRECATED: Use _is_shop_owner."""
logger.warning("is_shop_owner is deprecated, use _is_shop_owner")
return self._is_shop_owner(shop, user)
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
"""Check if user can view shop. DEPRECATED: Use _can_access_shop."""
logger.warning("can_view_shop is deprecated, use _can_access_shop")
return self._can_access_shop(shop, user)
# Create service instance following the same pattern as other services