refactor(vendor): move DB operations from API to service layer

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 <noreply@anthropic.com>
This commit is contained in:
2025-12-03 21:34:06 +01:00
parent e116be1a2c
commit 5f66ab4515
2 changed files with 208 additions and 74 deletions

View File

@@ -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