From 5f66ab45157383ee19dfa7fd9835c1baf675776f Mon Sep 17 00:00:00 2001 From: Samir Boulahtit Date: Wed, 3 Dec 2025 21:34:06 +0100 Subject: [PATCH] refactor(vendor): move DB operations from API to service layer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add new methods to vendor_service.py: - get_vendor_by_id(): Fetch vendor by ID with proper exception - get_vendor_by_identifier(): Fetch by ID or code - toggle_verification/set_verification(): Manage vendor verification - toggle_status/set_status(): Manage vendor active status Refactor vendors.py API endpoints: - Remove _get_vendor_by_identifier helper with direct DB queries - All endpoints now use vendor_service methods - Remove direct db.commit() calls from endpoints This fixes API-002 violations and improves architecture compliance. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- app/api/v1/admin/vendors.py | 114 ++++++++-------------- app/services/vendor_service.py | 168 ++++++++++++++++++++++++++++++++- 2 files changed, 208 insertions(+), 74 deletions(-) diff --git a/app/api/v1/admin/vendors.py b/app/api/v1/admin/vendors.py index bbe8d6d2..1a376e6a 100644 --- a/app/api/v1/admin/vendors.py +++ b/app/api/v1/admin/vendors.py @@ -1,22 +1,25 @@ # app/api/v1/admin/vendors.py """ Vendor management endpoints for admin. + +Architecture Notes: +- All business logic is in vendor_service (no direct DB operations here) +- Uses domain exceptions from app/exceptions/vendor.py +- Exception handler middleware converts domain exceptions to HTTP responses """ import logging from fastapi import APIRouter, Body, Depends, Path, Query -from sqlalchemy import func -from sqlalchemy.orm import Session, joinedload +from sqlalchemy.orm import Session from app.api.deps import get_current_admin_api from app.core.database import get_db -from app.exceptions import ConfirmationRequiredException, VendorNotFoundException +from app.exceptions import ConfirmationRequiredException from app.services.admin_service import admin_service from app.services.stats_service import stats_service -from models.database.company import Company +from app.services.vendor_service import vendor_service from models.database.user import User -from models.database.vendor import Vendor from models.schema.stats import VendorStatsResponse from models.schema.vendor import ( VendorCreate, @@ -30,46 +33,6 @@ router = APIRouter(prefix="/vendors") logger = logging.getLogger(__name__) -def _get_vendor_by_identifier(db: Session, identifier: str) -> Vendor: - """ - Helper to get vendor by ID or vendor_code. - Follows the pattern from admin_service._get_vendor_by_id_or_raise. - - Args: - db: Database session - identifier: Either vendor ID (int as string) or vendor_code (string) - - Returns: - Vendor object - - Raises: - VendorNotFoundException: If vendor not found - """ - # Try as integer ID first - try: - vendor_id = int(identifier) - return admin_service.get_vendor_by_id(db, vendor_id) - except (ValueError, TypeError): - # Not an integer, treat as vendor_code - pass - except VendorNotFoundException: - # ID not found, try as vendor_code - pass - - # Try as vendor_code (case-insensitive) - vendor = ( - db.query(Vendor) - .options(joinedload(Vendor.company).joinedload(Company.owner)) - .filter(func.upper(Vendor.vendor_code) == identifier.upper()) - .first() - ) - - if not vendor: - raise VendorNotFoundException(identifier, identifier_type="code") - - return vendor - - @router.post("", response_model=VendorCreateResponse) def create_vendor( vendor_data: VendorCreate, @@ -165,8 +128,11 @@ def get_vendor_details( Accepts either vendor ID (integer) or vendor_code (string). Returns vendor info with company contact details and owner info. + + Raises: + VendorNotFoundException: If vendor not found (404) """ - vendor = _get_vendor_by_identifier(db, vendor_identifier) + vendor = vendor_service.get_vendor_by_identifier(db, vendor_identifier) return VendorDetailResponse( # Vendor fields @@ -214,8 +180,11 @@ def update_vendor( **Cannot update:** - `vendor_code` (immutable) - Business contact info (use company update endpoints) + + Raises: + VendorNotFoundException: If vendor not found (404) """ - vendor = _get_vendor_by_identifier(db, vendor_identifier) + vendor = vendor_service.get_vendor_by_identifier(db, vendor_identifier) vendor = admin_service.update_vendor(db, vendor.id, vendor_update) return VendorDetailResponse( @@ -256,24 +225,22 @@ def toggle_vendor_verification( current_admin: User = Depends(get_current_admin_api), ): """ - Toggle vendor verification status (Admin only). + Set vendor verification status (Admin only). Accepts either vendor ID (integer) or vendor_code (string). Request body: { "is_verified": true/false } - """ - vendor = _get_vendor_by_identifier(db, vendor_identifier) - # Use admin_service method if available, otherwise update directly + Raises: + VendorNotFoundException: If vendor not found (404) + """ + vendor = vendor_service.get_vendor_by_identifier(db, vendor_identifier) + if "is_verified" in verification_data: - try: - vendor, message = admin_service.verify_vendor(db, vendor.id) - logger.info(f"Vendor verification toggled: {message}") - except AttributeError: - # If verify_vendor method doesn't exist, update directly - vendor.is_verified = verification_data["is_verified"] - db.commit() - db.refresh(vendor) + vendor, message = vendor_service.set_verification( + db, vendor.id, verification_data["is_verified"] + ) + logger.info(f"Vendor verification updated: {message}") return VendorDetailResponse( id=vendor.id, @@ -308,24 +275,22 @@ def toggle_vendor_status( current_admin: User = Depends(get_current_admin_api), ): """ - Toggle vendor active status (Admin only). + Set vendor active status (Admin only). Accepts either vendor ID (integer) or vendor_code (string). Request body: { "is_active": true/false } - """ - vendor = _get_vendor_by_identifier(db, vendor_identifier) - # Use admin_service method if available, otherwise update directly + Raises: + VendorNotFoundException: If vendor not found (404) + """ + vendor = vendor_service.get_vendor_by_identifier(db, vendor_identifier) + if "is_active" in status_data: - try: - vendor, message = admin_service.toggle_vendor_status(db, vendor.id) - logger.info(f"Vendor status toggled: {message}") - except AttributeError: - # If toggle_vendor_status method doesn't exist, update directly - vendor.is_active = status_data["is_active"] - db.commit() - db.refresh(vendor) + vendor, message = vendor_service.set_status( + db, vendor.id, status_data["is_active"] + ) + logger.info(f"Vendor status updated: {message}") return VendorDetailResponse( id=vendor.id, @@ -372,14 +337,17 @@ def delete_vendor( - All team members Requires confirmation parameter: `confirm=true` + + Raises: + ConfirmationRequiredException: If confirm=true not provided (400) + VendorNotFoundException: If vendor not found (404) """ - # Raise custom exception instead of HTTPException if not confirm: raise ConfirmationRequiredException( operation="delete_vendor", message="Deletion requires confirmation parameter: confirm=true", ) - vendor = _get_vendor_by_identifier(db, vendor_identifier) + vendor = vendor_service.get_vendor_by_identifier(db, vendor_identifier) message = admin_service.delete_vendor(db, vendor.id) return {"message": message} diff --git a/app/services/vendor_service.py b/app/services/vendor_service.py index c278a958..48b9daca 100644 --- a/app/services/vendor_service.py +++ b/app/services/vendor_service.py @@ -224,7 +224,173 @@ class VendorService: raise # Re-raise custom exceptions except Exception as e: logger.error(f"Error getting vendor {vendor_code}: {str(e)}") - raise ValidationException("Failed to retrieve vendor ") + raise ValidationException("Failed to retrieve vendor") + + def get_vendor_by_id(self, db: Session, vendor_id: int) -> Vendor: + """ + Get vendor by ID (admin use - no access control). + + Args: + db: Database session + vendor_id: Vendor ID to find + + Returns: + Vendor object with company and owner loaded + + Raises: + VendorNotFoundException: If vendor not found + """ + from sqlalchemy.orm import joinedload + from models.database.company import Company + + vendor = ( + db.query(Vendor) + .options(joinedload(Vendor.company).joinedload(Company.owner)) + .filter(Vendor.id == vendor_id) + .first() + ) + + if not vendor: + raise VendorNotFoundException(str(vendor_id), identifier_type="id") + + return vendor + + def get_vendor_by_identifier(self, db: Session, identifier: str) -> Vendor: + """ + Get vendor by ID or vendor_code (admin use - no access control). + + Args: + db: Database session + identifier: Either vendor ID (int as string) or vendor_code (string) + + Returns: + Vendor object with company and owner loaded + + Raises: + VendorNotFoundException: If vendor not found + """ + from sqlalchemy.orm import joinedload + from models.database.company import Company + + # Try as integer ID first + try: + vendor_id = int(identifier) + return self.get_vendor_by_id(db, vendor_id) + except (ValueError, TypeError): + pass # Not an integer, treat as vendor_code + except VendorNotFoundException: + pass # ID not found, try as vendor_code + + # Try as vendor_code (case-insensitive) + vendor = ( + db.query(Vendor) + .options(joinedload(Vendor.company).joinedload(Company.owner)) + .filter(func.upper(Vendor.vendor_code) == identifier.upper()) + .first() + ) + + if not vendor: + raise VendorNotFoundException(identifier, identifier_type="code") + + return vendor + + def toggle_verification(self, db: Session, vendor_id: int) -> tuple[Vendor, str]: + """ + Toggle vendor verification status. + + Args: + db: Database session + vendor_id: Vendor ID + + Returns: + Tuple of (updated vendor, status message) + + Raises: + VendorNotFoundException: If vendor not found + """ + vendor = self.get_vendor_by_id(db, vendor_id) + vendor.is_verified = not vendor.is_verified + db.commit() + db.refresh(vendor) + + status = "verified" if vendor.is_verified else "unverified" + logger.info(f"Vendor {vendor.vendor_code} {status}") + return vendor, f"Vendor {vendor.vendor_code} is now {status}" + + def set_verification( + self, db: Session, vendor_id: int, is_verified: bool + ) -> tuple[Vendor, str]: + """ + Set vendor verification status to specific value. + + Args: + db: Database session + vendor_id: Vendor ID + is_verified: Target verification status + + Returns: + Tuple of (updated vendor, status message) + + Raises: + VendorNotFoundException: If vendor not found + """ + vendor = self.get_vendor_by_id(db, vendor_id) + vendor.is_verified = is_verified + db.commit() + db.refresh(vendor) + + status = "verified" if is_verified else "unverified" + logger.info(f"Vendor {vendor.vendor_code} set to {status}") + return vendor, f"Vendor {vendor.vendor_code} is now {status}" + + def toggle_status(self, db: Session, vendor_id: int) -> tuple[Vendor, str]: + """ + Toggle vendor active status. + + Args: + db: Database session + vendor_id: Vendor ID + + Returns: + Tuple of (updated vendor, status message) + + Raises: + VendorNotFoundException: If vendor not found + """ + vendor = self.get_vendor_by_id(db, vendor_id) + vendor.is_active = not vendor.is_active + db.commit() + db.refresh(vendor) + + status = "active" if vendor.is_active else "inactive" + logger.info(f"Vendor {vendor.vendor_code} {status}") + return vendor, f"Vendor {vendor.vendor_code} is now {status}" + + def set_status( + self, db: Session, vendor_id: int, is_active: bool + ) -> tuple[Vendor, str]: + """ + Set vendor active status to specific value. + + Args: + db: Database session + vendor_id: Vendor ID + is_active: Target active status + + Returns: + Tuple of (updated vendor, status message) + + Raises: + VendorNotFoundException: If vendor not found + """ + vendor = self.get_vendor_by_id(db, vendor_id) + vendor.is_active = is_active + db.commit() + db.refresh(vendor) + + status = "active" if is_active else "inactive" + logger.info(f"Vendor {vendor.vendor_code} set to {status}") + return vendor, f"Vendor {vendor.vendor_code} is now {status}" def add_product_to_catalog( self, db: Session, vendor: Vendor, product: ProductCreate