feat: Implement distributor management functionality with CRUD operations and pagination

app/app.py

@@ -11,6 +11,7 @@ from app.controllers.reports import router as reports_router
 from app.controllers.employees import router as employees_router
 from app.controllers.reference import router as reference_router
 from app.controllers.bidders import router as bidders_router
+from app.controllers.distributors import router as distributors_router
 
 setup_logging(settings.LOG_LEVEL)
 

app/controllers/distributors.py

@@ -0,0 +1,300 @@
+from fastapi import APIRouter, Depends, status, Query, HTTPException
+from sqlalchemy.orm import Session
+from app.db.session import get_db
+from app.services.distributor_service import DistributorService
+from app.schemas.distributor import DistributorCreate, DistributorOut
+from app.schemas.paginated_response import PaginatedResponse
+from typing import Optional, List
+import logging
+
+logger = logging.getLogger(__name__)
+
+# Common query parameter descriptions
+PAGE_DESC = "Page number (1-indexed)"
+PAGE_SIZE_DESC = "Number of items per page"
+ORDER_BY_DESC = "Field to order by"
+ORDER_DIR_DESC = "Order direction (asc|desc)"
+TERRITORY_DESC = "Territory/Region name"
+STATUS_DESC = "Distributor status (active|inactive|pending)"
+
+router = APIRouter(prefix="/api/v1/distributors", tags=["distributors"])
+
+@router.get(
+    "/",
+    response_model=PaginatedResponse[DistributorOut],
+    summary="List all distributors",
+    response_description="Paginated list of distributors"
+)
+def list_distributors(
+    page: int = Query(1, ge=1, description=PAGE_DESC),
+    page_size: int = Query(10, ge=1, le=100, description=PAGE_SIZE_DESC),
+    order_by: Optional[str] = Query("Id", description=ORDER_BY_DESC),
+    order_dir: Optional[str] = Query("asc", description=ORDER_DIR_DESC),
+    db: Session = Depends(get_db)
+):
+    """Get all distributors with pagination and ordering"""
+    try:
+        logger.info(f"Listing distributors: page={page}, page_size={page_size}")
+        distributor_service = DistributorService(db)
+        result = distributor_service.list(
+            page=page, 
+            page_size=page_size, 
+            order_by=order_by, 
+            order_direction=order_dir.upper()
+        )
+        logger.info(f"Successfully retrieved {len(result.items)} distributors")
+        return result
+    except Exception as e:
+        logger.error(f"Error listing distributors: {e}")
+        return PaginatedResponse[DistributorOut](
+            items=[],
+            page=page,
+            page_size=page_size,
+            total=0
+        )
+
+@router.get(
+    "/territory/{territory}",
+    response_model=PaginatedResponse[DistributorOut],
+    summary="List distributors by territory",
+    response_description="Paginated list of distributors for a specific territory"
+)
+def list_distributors_by_territory(
+    territory: str,
+    page: int = Query(1, ge=1, description=PAGE_DESC),
+    page_size: int = Query(10, ge=1, le=100, description=PAGE_SIZE_DESC),
+    order_by: Optional[str] = Query("Id", description=ORDER_BY_DESC),
+    order_dir: Optional[str] = Query("asc", description=ORDER_DIR_DESC),
+    db: Session = Depends(get_db)
+):
+    """Get all distributors for a specific territory with pagination and ordering"""
+    try:
+        logger.info(f"Listing distributors for territory {territory}: page={page}, page_size={page_size}")
+        distributor_service = DistributorService(db)
+        result = distributor_service.get_by_territory(
+            territory=territory,
+            page=page, 
+            page_size=page_size, 
+            order_by=order_by, 
+            order_dir=order_dir.upper()
+        )
+        logger.info(f"Successfully retrieved {len(result.items)} distributors for territory {territory}")
+        return result
+    except Exception as e:
+        logger.error(f"Error listing distributors for territory {territory}: {e}")
+        return PaginatedResponse[DistributorOut](
+            items=[],
+            page=page,
+            page_size=page_size,
+            total=0
+        )
+
+@router.get(
+    "/status/{status}",
+    response_model=PaginatedResponse[DistributorOut],
+    summary="List distributors by status",
+    response_description="Paginated list of distributors with a specific status"
+)
+def list_distributors_by_status(
+    status: str,
+    page: int = Query(1, ge=1, description=PAGE_DESC),
+    page_size: int = Query(10, ge=1, le=100, description=PAGE_SIZE_DESC),
+    order_by: Optional[str] = Query("Id", description=ORDER_BY_DESC),
+    order_dir: Optional[str] = Query("asc", description=ORDER_DIR_DESC),
+    db: Session = Depends(get_db)
+):
+    """Get all distributors with a specific status with pagination and ordering"""
+    try:
+        logger.info(f"Listing distributors with status {status}: page={page}, page_size={page_size}")
+        distributor_service = DistributorService(db)
+        result = distributor_service.get_by_status(
+            status=status,
+            page=page, 
+            page_size=page_size, 
+            order_by=order_by, 
+            order_dir=order_dir.upper()
+        )
+        logger.info(f"Successfully retrieved {len(result.items)} distributors with status {status}")
+        return result
+    except Exception as e:
+        logger.error(f"Error listing distributors with status {status}: {e}")
+        return PaginatedResponse[DistributorOut](
+            items=[],
+            page=page,
+            page_size=page_size,
+            total=0
+        )
+
+@router.get(
+    "/territories",
+    response_model=List[str],
+    summary="Get all territories",
+    response_description="List of unique territories"
+)
+def get_territories(db: Session = Depends(get_db)):
+    """Get all unique territories"""
+    try:
+        distributor_service = DistributorService(db)
+        territories = distributor_service.get_territories()
+        logger.info(f"Successfully retrieved {len(territories)} territories")
+        return territories
+    except Exception as e:
+        logger.error(f"Error getting territories: {e}")
+        return []
+
+@router.get(
+    "/{distributor_id}",
+    response_model=DistributorOut,
+    summary="Get a distributor by ID",
+    response_description="Distributor details"
+)
+def get_distributor(distributor_id: int, db: Session = Depends(get_db)):
+    """Get a specific distributor by ID"""
+    try:
+        distributor_service = DistributorService(db)
+        return distributor_service.get(distributor_id)
+    except Exception as e:
+        logger.error(f"Error getting distributor {distributor_id}: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Distributor {distributor_id} not found"
+        )
+
+@router.post(
+    "/",
+    response_model=DistributorOut,
+    status_code=status.HTTP_201_CREATED,
+    summary="Create a new distributor",
+    response_description="Created distributor details"
+)
+def create_distributor(
+    distributor_in: DistributorCreate,
+    db: Session = Depends(get_db)
+):
+    """Create a new distributor"""
+    try:
+        distributor_service = DistributorService(db)
+        logger.info(f"Creating new distributor: {distributor_in.company_name}")
+        
+        result = distributor_service.create(distributor_in.dict())
+        logger.info(f"Successfully created distributor {result.id}: {result.company_name}")
+        return result
+    except ValueError as e:
+        logger.error(f"Validation error creating distributor: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=str(e)
+        )
+    except Exception as e:
+        logger.error(f"Error creating distributor: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to create distributor"
+        )
+
+@router.put(
+    "/{distributor_id}",
+    response_model=DistributorOut,
+    summary="Update a distributor",
+    response_description="Updated distributor details"
+)
+def update_distributor(
+    distributor_id: int,
+    distributor_in: DistributorCreate,
+    db: Session = Depends(get_db)
+):
+    """Update an existing distributor"""
+    try:
+        distributor_service = DistributorService(db)
+        logger.info(f"Updating distributor {distributor_id}: {distributor_in.company_name}")
+        
+        result = distributor_service.update(distributor_id, distributor_in.dict())
+        logger.info(f"Successfully updated distributor {distributor_id}: {result.company_name}")
+        return result
+    except ValueError as e:
+        logger.error(f"Validation error updating distributor {distributor_id}: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=str(e)
+        )
+    except Exception as e:
+        logger.error(f"Error updating distributor {distributor_id}: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to update distributor"
+        )
+
+@router.delete(
+    "/{distributor_id}",
+    status_code=status.HTTP_204_NO_CONTENT,
+    summary="Delete a distributor",
+    response_description="Distributor deleted successfully"
+)
+def delete_distributor(
+    distributor_id: int,
+    db: Session = Depends(get_db)
+):
+    """Delete a distributor"""
+    try:
+        distributor_service = DistributorService(db)
+        logger.info(f"Deleting distributor {distributor_id}")
+        
+        distributor_service.delete(distributor_id)
+        logger.info(f"Successfully deleted distributor {distributor_id}")
+        return None
+    except Exception as e:
+        logger.error(f"Error deleting distributor {distributor_id}: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to delete distributor"
+        )
+
+@router.post(
+    "/{distributor_id}/activate",
+    response_model=DistributorOut,
+    summary="Activate a distributor",
+    response_description="Activated distributor details"
+)
+def activate_distributor(
+    distributor_id: int,
+    db: Session = Depends(get_db)
+):
+    """Activate a distributor (set status to active)"""
+    try:
+        distributor_service = DistributorService(db)
+        logger.info(f"Activating distributor {distributor_id}")
+        
+        result = distributor_service.activate(distributor_id)
+        logger.info(f"Successfully activated distributor {distributor_id}")
+        return result
+    except Exception as e:
+        logger.error(f"Error activating distributor {distributor_id}: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to activate distributor"
+        )
+
+@router.post(
+    "/{distributor_id}/deactivate",
+    response_model=DistributorOut,
+    summary="Deactivate a distributor",
+    response_description="Deactivated distributor details"
+)
+def deactivate_distributor(
+    distributor_id: int,
+    db: Session = Depends(get_db)
+):
+    """Deactivate a distributor (set status to inactive)"""
+    try:
+        distributor_service = DistributorService(db)
+        logger.info(f"Deactivating distributor {distributor_id}")
+        
+        result = distributor_service.deactivate(distributor_id)
+        logger.info(f"Successfully deactivated distributor {distributor_id}")
+        return result
+    except Exception as e:
+        logger.error(f"Error deactivating distributor {distributor_id}: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to deactivate distributor"
+        )

app/db/models/distributor.py

@@ -0,0 +1,26 @@
+from sqlalchemy import Column, Integer, String, Boolean, Float, DateTime, ForeignKey
+from app.db.base import Base
+
+class Distributor(Base):
+    __tablename__ = "Distributors"
+    
+    Id = Column(Integer, primary_key=True, index=True)
+    CompanyName = Column(String(200), nullable=False)
+    ContactPerson = Column(String(200), nullable=True)
+    EmailAddress = Column(String(255), nullable=True)
+    Phone = Column(String(50), nullable=True)
+    Fax = Column(String(50), nullable=True)
+    Address = Column(String(500), nullable=True)
+    City = Column(String(100), nullable=True)
+    State = Column(String(50), nullable=True)
+    ZipCode = Column(String(20), nullable=True)
+    Country = Column(String(100), nullable=True)
+    Website = Column(String(255), nullable=True)
+    Territory = Column(String(100), nullable=True)
+    CommissionRate = Column(Float, nullable=True, default=0.0)
+    Status = Column(String(20), nullable=False, default='active')
+    Notes = Column(String(1000), nullable=True)
+    DateCreated = Column(DateTime, nullable=True)
+    DateModified = Column(DateTime, nullable=True)
+    Enabled = Column(Boolean, nullable=False, default=True)
+    EmployeeId = Column(Integer, nullable=True)

app/db/models/reference.py

@@ -5,7 +5,8 @@ class State(Base):
     __tablename__ = "States"
     
     state_id = Column("StateID", Integer, primary_key=True, index=True)
-    state_name = Column("StateName", String(50), nullable=False)
+    # Renamed from StateName to State as per database schema
+    state_name = Column("State", String(50), nullable=False)
     state_code = Column("StateCode", String(2), nullable=False)
     country = Column("Country", String(50), nullable=True)
 

app/db/repositories/distributor_repo.py

@@ -0,0 +1,255 @@
+from sqlalchemy.orm import Session
+from sqlalchemy.exc import SQLAlchemyError
+from app.db.models.distributor import Distributor
+from app.schemas.distributor import DistributorCreate, DistributorOut
+from app.schemas.paginated_response import PaginatedResponse
+from app.core.exceptions import NotFoundException
+from typing import List, Optional
+from datetime import datetime, timezone
+import logging
+
+logger = logging.getLogger(__name__)
+
+class DistributorRepository:
+    def __init__(self, db: Session):
+        self.db = db
+
+    def get(self, distributor_id: int) -> Optional[DistributorOut]:
+        """Get a single distributor by ID using Distributors table directly."""
+        try:
+            distributor = self.db.query(Distributor).filter(Distributor.Id == distributor_id).first()
+            if distributor:
+                return self._map_distributor_data(distributor.__dict__)
+            return None
+        except SQLAlchemyError as e:
+            logger.error(f"Database error getting distributor {distributor_id}: {e}")
+            raise
+        except Exception as e:
+            logger.error(f"Unexpected error getting distributor {distributor_id}: {e}")
+            raise
+
+    def list(self, page: int = 1, page_size: int = 10, order_by: str = "Id", order_direction: str = "ASC") -> PaginatedResponse[DistributorOut]:
+        """Get paginated list of all distributors using Distributors table directly with order by."""
+        try:
+            query = self.db.query(Distributor)
+            # Validate order_by field
+            if not hasattr(Distributor, order_by):
+                order_by = "Id"
+            order_col = getattr(Distributor, order_by)
+            if order_direction.upper() == "DESC":
+                order_col = order_col.desc()
+            else:
+                order_col = order_col.asc()
+            query = query.order_by(order_col)
+            total_records = query.count()
+            distributors = query.offset((page - 1) * page_size).limit(page_size).all()
+            distributor_out_list = [self._map_distributor_data(distributor.__dict__) for distributor in distributors]
+            return PaginatedResponse[DistributorOut](
+                items=distributor_out_list,
+                page=page,
+                page_size=page_size,
+                total=total_records
+            )
+        except Exception as e:
+            logger.error(f"Error listing distributors: {e}")
+            return PaginatedResponse[DistributorOut](
+                items=[],
+                page=page,
+                page_size=page_size,
+                total=0
+            )
+
+    def get_by_territory(self, territory: str, page: int = 1, page_size: int = 10, 
+                         order_by: str = "Id", order_dir: str = "ASC") -> PaginatedResponse[DistributorOut]:
+        """Get distributors for a specific territory using Distributors table directly."""
+        try:
+            query = self.db.query(Distributor).filter(Distributor.Territory == territory)
+            # Validate order_by field
+            if not hasattr(Distributor, order_by):
+                order_by = "Id"
+            order_col = getattr(Distributor, order_by)
+            if order_dir.upper() == "DESC":
+                order_col = order_col.desc()
+            else:
+                order_col = order_col.asc()
+            query = query.order_by(order_col)
+            total_records = query.count()
+            distributors = query.offset((page - 1) * page_size).limit(page_size).all()
+            distributor_out_list = [self._map_distributor_data(distributor.__dict__) for distributor in distributors]
+            return PaginatedResponse[DistributorOut](
+                items=distributor_out_list,
+                page=page,
+                page_size=page_size,
+                total=total_records
+            )
+        except Exception as e:
+            logger.error(f"Error getting distributors for territory {territory}: {e}")
+            return PaginatedResponse[DistributorOut](
+                items=[],
+                page=page,
+                page_size=page_size,
+                total=0
+            )
+
+    def get_by_status(self, status: str, page: int = 1, page_size: int = 10, 
+                      order_by: str = "Id", order_dir: str = "ASC") -> PaginatedResponse[DistributorOut]:
+        """Get distributors by status using Distributors table directly."""
+        try:
+            query = self.db.query(Distributor).filter(Distributor.Status == status)
+            # Validate order_by field
+            if not hasattr(Distributor, order_by):
+                order_by = "Id"
+            order_col = getattr(Distributor, order_by)
+            if order_dir.upper() == "DESC":
+                order_col = order_col.desc()
+            else:
+                order_col = order_col.asc()
+            query = query.order_by(order_col)
+            total_records = query.count()
+            distributors = query.offset((page - 1) * page_size).limit(page_size).all()
+            distributor_out_list = [self._map_distributor_data(distributor.__dict__) for distributor in distributors]
+            return PaginatedResponse[DistributorOut](
+                items=distributor_out_list,
+                page=page,
+                page_size=page_size,
+                total=total_records
+            )
+        except Exception as e:
+            logger.error(f"Error getting distributors by status {status}: {e}")
+            return PaginatedResponse[DistributorOut](
+                items=[],
+                page=page,
+                page_size=page_size,
+                total=0
+            )
+
+    def create(self, distributor_data: DistributorCreate) -> DistributorOut:
+        """Create a new distributor using Distributors table directly."""
+        try:
+            current_time = datetime.now(timezone.utc)
+            new_distributor = Distributor(
+                CompanyName=distributor_data.company_name,
+                ContactPerson=distributor_data.contact_person,
+                EmailAddress=distributor_data.email_address,
+                Phone=distributor_data.phone,
+                Fax=distributor_data.fax,
+                Address=distributor_data.address,
+                City=distributor_data.city,
+                State=distributor_data.state,
+                ZipCode=distributor_data.zip_code,
+                Country=distributor_data.country,
+                Website=distributor_data.website,
+                Territory=distributor_data.territory,
+                CommissionRate=distributor_data.commission_rate or 0.0,
+                Status=distributor_data.status or 'active',
+                Notes=distributor_data.notes,
+                DateCreated=current_time,
+                DateModified=current_time,
+                Enabled=distributor_data.enabled or True,
+                EmployeeId=distributor_data.employee_id
+            )
+            self.db.add(new_distributor)
+            self.db.commit()
+            self.db.refresh(new_distributor)
+            return self._map_distributor_data(new_distributor.__dict__)
+        except SQLAlchemyError as e:
+            logger.error(f"Database error creating distributor: {e}")
+            self.db.rollback()
+            raise
+        except Exception as e:
+            logger.error(f"Unexpected error creating distributor: {e}")
+            self.db.rollback()
+            raise
+
+    def update(self, distributor_id: int, distributor_data: DistributorCreate) -> DistributorOut:
+        """Update a distributor using Distributors table directly."""
+        try:
+            distributor = self.db.query(Distributor).filter(Distributor.Id == distributor_id).first()
+            if not distributor:
+                raise NotFoundException("Distributor not found for update")
+            
+            distributor.CompanyName = distributor_data.company_name
+            distributor.ContactPerson = distributor_data.contact_person
+            distributor.EmailAddress = distributor_data.email_address
+            distributor.Phone = distributor_data.phone
+            distributor.Fax = distributor_data.fax
+            distributor.Address = distributor_data.address
+            distributor.City = distributor_data.city
+            distributor.State = distributor_data.state
+            distributor.ZipCode = distributor_data.zip_code
+            distributor.Country = distributor_data.country
+            distributor.Website = distributor_data.website
+            distributor.Territory = distributor_data.territory
+            distributor.CommissionRate = distributor_data.commission_rate
+            distributor.Status = distributor_data.status
+            distributor.Notes = distributor_data.notes
+            distributor.DateModified = datetime.now(timezone.utc)
+            distributor.Enabled = distributor_data.enabled
+            distributor.EmployeeId = distributor_data.employee_id
+            
+            self.db.commit()
+            self.db.refresh(distributor)
+            return self._map_distributor_data(distributor.__dict__)
+        except SQLAlchemyError as e:
+            logger.error(f"Database error updating distributor {distributor_id}: {e}")
+            self.db.rollback()
+            raise
+        except Exception as e:
+            logger.error(f"Unexpected error updating distributor {distributor_id}: {e}")
+            self.db.rollback()
+            raise
+
+    def delete(self, distributor_id: int) -> bool:
+        """Delete a distributor using Distributors table directly."""
+        try:
+            distributor = self.db.query(Distributor).filter(Distributor.Id == distributor_id).first()
+            if not distributor:
+                raise NotFoundException("Distributor not found for delete")
+            self.db.delete(distributor)
+            self.db.commit()
+            return True
+        except SQLAlchemyError as e:
+            logger.error(f"Database error deleting distributor {distributor_id}: {e}")
+            self.db.rollback()
+            raise
+        except Exception as e:
+            logger.error(f"Unexpected error deleting distributor {distributor_id}: {e}")
+            self.db.rollback()
+            raise
+
+    def get_territories(self) -> List[str]:
+        """Get all unique territories."""
+        try:
+            territories = self.db.query(Distributor.Territory).filter(
+                Distributor.Territory.isnot(None),
+                Distributor.Territory != ''
+            ).distinct().all()
+            return [territory[0] for territory in territories]
+        except Exception as e:
+            logger.error(f"Error getting territories: {e}")
+            return []
+
+    def _map_distributor_data(self, distributor_data: dict) -> DistributorOut:
+        """Map database distributor data to DistributorOut schema"""
+        return DistributorOut(
+            id=distributor_data.get('Id'),
+            company_name=distributor_data.get('CompanyName'),
+            contact_person=distributor_data.get('ContactPerson'),
+            email_address=distributor_data.get('EmailAddress'),
+            phone=distributor_data.get('Phone'),
+            fax=distributor_data.get('Fax'),
+            address=distributor_data.get('Address'),
+            city=distributor_data.get('City'),
+            state=distributor_data.get('State'),
+            zip_code=distributor_data.get('ZipCode'),
+            country=distributor_data.get('Country'),
+            website=distributor_data.get('Website'),
+            territory=distributor_data.get('Territory'),
+            commission_rate=distributor_data.get('CommissionRate'),
+            status=distributor_data.get('Status'),
+            notes=distributor_data.get('Notes'),
+            date_created=distributor_data.get('DateCreated'),
+            date_modified=distributor_data.get('DateModified'),
+            enabled=distributor_data.get('Enabled'),
+            employee_id=distributor_data.get('EmployeeId')
+        )

app/db/repositories/reference_repo.py

@@ -48,13 +48,26 @@ class ReferenceDataRepository:
             result = self.db.execute(sp_query)
             
             if result.returns_rows:
-                states_data = [dict(row._mapping) for row in result.fetchall()]
+                # Map the stored procedure results to our expected schema
+                states_data = []
+                for row in result.fetchall():
+                    row_dict = dict(row._mapping)
+                    # Adjust for potential naming differences in stored procedure output
+                    # Map the SP's column names to our expected schema
+                    state_data = {
+                        'state_id': row_dict.get('StateID', None),
+                        'state_name': row_dict.get('State', row_dict.get('StateName', None)),
+                        'state_code': row_dict.get('StateCode', None),
+                        'country': row_dict.get('Country', None)
+                    }
+                    states_data.append(state_data)
+                    
                 self._cache[cache_key] = states_data
                 logger.info(f"Retrieved {len(states_data)} states via stored procedure")
                 return states_data
                 
         except Exception as e:
-            logger.warning(f"Stored procedure failed, using fallback: {e}")
+            logger.warning(f"Stored procedure failed, using fallback: {e}", exc_info=True)
             
         # Fallback to direct query
         query = self.db.query(State)
@@ -66,7 +79,7 @@ class ReferenceDataRepository:
         states_data = [
             {
                 'state_id': state.state_id,
-                'state_name': state.state_name,
+                'state_name': state.state_name,  # This will now correctly map from the 'State' column
                 'state_code': state.state_code,
                 'country': state.country
             }

app/schemas/distributor.py

@@ -0,0 +1,47 @@
+from pydantic import BaseModel, Field
+from typing import Optional
+from datetime import datetime
+
+class DistributorCreate(BaseModel):
+    company_name: str = Field(..., description="Company name")
+    contact_person: Optional[str] = Field(None, description="Contact person name")
+    email_address: Optional[str] = Field(None, description="Email address")
+    phone: Optional[str] = Field(None, description="Phone number")
+    fax: Optional[str] = Field(None, description="Fax number")
+    address: Optional[str] = Field(None, description="Street address")
+    city: Optional[str] = Field(None, description="City")
+    state: Optional[str] = Field(None, description="State/Province")
+    zip_code: Optional[str] = Field(None, description="ZIP/Postal code")
+    country: Optional[str] = Field(None, description="Country")
+    website: Optional[str] = Field(None, description="Website URL")
+    territory: Optional[str] = Field(None, description="Territory/Region")
+    commission_rate: Optional[float] = Field(0.0, description="Commission rate")
+    status: Optional[str] = Field("active", description="Distributor status")
+    notes: Optional[str] = Field(None, description="Additional notes")
+    enabled: Optional[bool] = Field(True, description="Enabled status")
+    employee_id: Optional[int] = Field(None, description="Employee ID")
+
+class DistributorOut(BaseModel):
+    id: int = Field(..., description="Distributor ID")
+    company_name: str = Field(..., description="Company name")
+    contact_person: Optional[str] = Field(None, description="Contact person name")
+    email_address: Optional[str] = Field(None, description="Email address")
+    phone: Optional[str] = Field(None, description="Phone number")
+    fax: Optional[str] = Field(None, description="Fax number")
+    address: Optional[str] = Field(None, description="Street address")
+    city: Optional[str] = Field(None, description="City")
+    state: Optional[str] = Field(None, description="State/Province")
+    zip_code: Optional[str] = Field(None, description="ZIP/Postal code")
+    country: Optional[str] = Field(None, description="Country")
+    website: Optional[str] = Field(None, description="Website URL")
+    territory: Optional[str] = Field(None, description="Territory/Region")
+    commission_rate: Optional[float] = Field(None, description="Commission rate")
+    status: Optional[str] = Field(None, description="Distributor status")
+    notes: Optional[str] = Field(None, description="Additional notes")
+    date_created: Optional[datetime] = Field(None, description="Date created")
+    date_modified: Optional[datetime] = Field(None, description="Date modified")
+    enabled: Optional[bool] = Field(None, description="Enabled status")
+    employee_id: Optional[int] = Field(None, description="Employee ID")
+
+    class Config:
+        from_attributes = True  # Updated for Pydantic v2

app/services/distributor_service.py

@@ -0,0 +1,161 @@
+from sqlalchemy.orm import Session
+from app.db.repositories.distributor_repo import DistributorRepository
+from app.schemas.distributor import DistributorCreate, DistributorOut
+from app.schemas.paginated_response import PaginatedResponse
+from app.core.exceptions import NotFoundException
+from typing import List, Optional
+
+# Constants
+DISTRIBUTOR_NOT_FOUND_MSG = "Distributor not found"
+
+class DistributorService:
+    def __init__(self, db: Session):
+        self.repo = DistributorRepository(db)
+
+    def get(self, distributor_id: int) -> DistributorOut:
+        """Get a single distributor by ID"""
+        distributor = self.repo.get(distributor_id)
+        if not distributor:
+            raise NotFoundException(DISTRIBUTOR_NOT_FOUND_MSG)
+        return distributor
+
+    def list(self, page: int = 1, page_size: int = 10, order_by: str = "Id", 
+             order_direction: str = "ASC") -> PaginatedResponse[DistributorOut]:
+        """Get a paginated list of all distributors"""
+        return self.repo.list(page, page_size, order_by, order_direction)
+
+    def get_by_territory(self, territory: str, page: int = 1, page_size: int = 10, 
+                        order_by: str = "Id", order_dir: str = "ASC") -> PaginatedResponse[DistributorOut]:
+        """Get all distributors for a specific territory"""
+        return self.repo.get_by_territory(territory, page, page_size, order_by, order_dir)
+
+    def get_by_status(self, status: str, page: int = 1, page_size: int = 10, 
+                     order_by: str = "Id", order_dir: str = "ASC") -> PaginatedResponse[DistributorOut]:
+        """Get all distributors by status"""
+        return self.repo.get_by_status(status, page, page_size, order_by, order_dir)
+
+    def create(self, data: dict) -> DistributorOut:
+        """Create a new distributor"""
+        # Validate and clean data
+        validated_data = self._validate_distributor_data(data)
+        
+        # Convert dict to DistributorCreate for validation
+        distributor_data = DistributorCreate(**validated_data)
+        
+        return self.repo.create(distributor_data)
+
+    def update(self, distributor_id: int, data: dict) -> DistributorOut:
+        """Update an existing distributor"""
+        # Check if distributor exists
+        existing_distributor = self.repo.get(distributor_id)
+        if not existing_distributor:
+            raise NotFoundException(DISTRIBUTOR_NOT_FOUND_MSG)
+        
+        # Validate and clean data
+        validated_data = self._validate_distributor_data(data)
+        
+        # Convert dict to DistributorCreate for validation
+        distributor_data = DistributorCreate(**validated_data)
+        
+        return self.repo.update(distributor_id, distributor_data)
+
+    def delete(self, distributor_id: int):
+        """Delete a distributor"""
+        # Check if distributor exists
+        existing_distributor = self.repo.get(distributor_id)
+        if not existing_distributor:
+            raise NotFoundException(DISTRIBUTOR_NOT_FOUND_MSG)
+        
+        self.repo.delete(distributor_id)
+
+    def get_territories(self) -> List[str]:
+        """Get all unique territories"""
+        return self.repo.get_territories()
+
+    def activate(self, distributor_id: int) -> DistributorOut:
+        """Activate a distributor"""
+        # Check if distributor exists
+        existing_distributor = self.repo.get(distributor_id)
+        if not existing_distributor:
+            raise NotFoundException(DISTRIBUTOR_NOT_FOUND_MSG)
+        
+        # Update status to active
+        distributor_data = DistributorCreate(**{
+            **existing_distributor.dict(),
+            'status': 'active'
+        })
+        
+        return self.repo.update(distributor_id, distributor_data)
+
+    def deactivate(self, distributor_id: int) -> DistributorOut:
+        """Deactivate a distributor"""
+        # Check if distributor exists
+        existing_distributor = self.repo.get(distributor_id)
+        if not existing_distributor:
+            raise NotFoundException(DISTRIBUTOR_NOT_FOUND_MSG)
+        
+        # Update status to inactive
+        distributor_data = DistributorCreate(**{
+            **existing_distributor.dict(),
+            'status': 'inactive'
+        })
+        
+        return self.repo.update(distributor_id, distributor_data)
+
+    def _validate_distributor_data(self, data: dict) -> dict:
+        """Internal method to validate and clean distributor data"""
+        # Ensure company name is provided (required)
+        if not data.get('company_name'):
+            raise ValueError("Company name is required")
+            
+        # Clean and validate email
+        self._validate_email(data)
+        
+        # Clean phone numbers
+        self._clean_phone_fields(data)
+        
+        # Validate commission rate
+        self._validate_commission_rate(data)
+        
+        # Validate status
+        self._validate_status(data)
+        
+        # Clean website URL
+        self._clean_website_url(data)
+            
+        return data
+
+    def _validate_email(self, data: dict):
+        """Validate and clean email address"""
+        if data.get('email_address'):
+            email = data['email_address'].strip().lower()
+            if '@' not in email:
+                raise ValueError("Invalid email address format")
+            data['email_address'] = email
+
+    def _clean_phone_fields(self, data: dict):
+        """Clean phone number fields"""
+        phone_fields = ['phone', 'fax']
+        for field in phone_fields:
+            if data.get(field):
+                cleaned_phone = ''.join(c for c in data[field] if c.isdigit() or c in '+()-. ')
+                data[field] = cleaned_phone.strip()
+
+    def _validate_commission_rate(self, data: dict):
+        """Validate commission rate"""
+        if data.get('commission_rate') is not None and data.get('commission_rate') < 0:
+            raise ValueError("Commission rate must be a positive number")
+
+    def _validate_status(self, data: dict):
+        """Validate status field"""
+        valid_statuses = ['active', 'inactive', 'pending']
+        if data.get('status') and data.get('status') not in valid_statuses:
+            raise ValueError(f"Status must be one of: {', '.join(valid_statuses)}")
+
+    def _clean_website_url(self, data: dict):
+        """Clean and format website URL"""
+        if data.get('website'):
+            website = data['website'].strip()
+            if website and not website.startswith(('http://', 'https://')):
+                website = 'https://' + website
+            data['website'] = website