phone type reference data

app/controllers/reference.py

@@ -12,7 +12,7 @@ from app.schemas.reference import (
     BarrierSizeCreateIn, BarrierSizeUpdateIn, ProductApplicationCreateIn, ProductApplicationUpdateIn,
     FOBCreateIn, FOBUpdateIn, EstShipDateCreateIn, EstShipDateUpdateIn,
     EstFreightCreateIn, EstFreightUpdateIn, StatusInfoCreateIn, StatusInfoUpdateIn,
-    PriorityCreateIn, PriorityUpdateIn
+    PriorityCreateIn, PriorityUpdateIn, PhoneTypeOut, PhoneTypeCreateIn, PhoneTypeUpdateIn
 )
 from app.core.dependencies import get_current_user_optional
 from app.schemas.auth import CurrentUser
@@ -879,4 +879,51 @@ def delete_priority(
     success = service.delete_priority(priority_id)
     if not success:
         raise HTTPException(status_code=404, detail="Priority not found")
+    return Response(status_code=204)
+
+@router.get("/phone-types", response_model=List[PhoneTypeOut])
+def get_phone_types(
+    active_only: bool = Query(False, description="Return only active records"),
+    db: Session = Depends(get_db),
+    current_user: Optional[CurrentUser] = Depends(get_current_user_optional)
+):
+    """Get all phone types"""
+    service = ReferenceDataService(db)
+    return service.get_phone_types(active_only)
+
+@router.post("/phone-types", response_model=PhoneTypeOut, status_code=201)
+def create_phone_type(
+    data: PhoneTypeCreateIn,
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user_optional)
+):
+    """Create a new phone type"""
+    service = ReferenceDataService(db)
+    return service.create_phone_type(data)
+
+@router.put("/phone-types/{phone_type_id}", response_model=PhoneTypeOut)
+def update_phone_type(
+    phone_type_id: int,
+    data: PhoneTypeUpdateIn,
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user_optional)
+):
+    """Update an existing phone type"""
+    service = ReferenceDataService(db)
+    result = service.update_phone_type(phone_type_id, data)
+    if not result:
+        raise HTTPException(status_code=404, detail="Phone type not found")
+    return result
+
+@router.delete("/phone-types/{phone_type_id}", status_code=204)
+def delete_phone_type(
+    phone_type_id: int,
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user_optional)
+):
+    """Delete a phone type"""
+    service = ReferenceDataService(db)
+    success = service.delete_phone_type(phone_type_id)
+    if not success:
+        raise HTTPException(status_code=404, detail="Phone type not found")
     return Response(status_code=204)

app/db/models/reference.py

@@ -117,4 +117,13 @@ class EstFreight(Base):
 
     est_freight_id = Column("Id", Integer, primary_key=True, index=True)
     est_freight_description = Column("EstFreightDescription", String(255), nullable=True)
+
+class PhoneType(Base):
+    __tablename__ = "PhoneType"
+
+    id = Column("Id", Integer, primary_key=True, index=True)
+    name = Column("Name", String(50), nullable=False)
+    enabled = Column("Enabled", Boolean, nullable=True)
+    show_in_contact_page = Column("ShowInContactPage", Boolean, nullable=True)
+    modified_date = Column("ModifiedDate", DateTime, nullable=True)
      

app/db/repositories/reference_repo.py

@@ -3,7 +3,7 @@ from sqlalchemy import text
 from app.db.models.reference import (
     State, Country, CompanyType, LeadGeneratedFrom, PaymentTerm,
     PurchasePrice, RentalPrice, BarrierSize, ProductApplication, Priority,
-    FOB, EstShipDate, EstFreight, StatusInfo
+    FOB, EstShipDate, EstFreight, StatusInfo, PhoneType
 )
 from typing import List, Dict, Any, Optional, Type, Union
 from app.core.exceptions import RepositoryException
@@ -406,6 +406,7 @@ class ReferenceDataRepository:
             'est_freights': self.get_est_freights(active_only),
             'status_info': self.get_status_info(active_only),
             'priorities': self.get_priorities(True, 1),  # Get priorities with CustomerTypeID = 1
+            'phone_types': self.get_phone_types(active_only),
         }
 
     def get_status_info(self, active_only: bool = True) -> List[Dict[str, Any]]:
@@ -468,6 +469,37 @@ class ReferenceDataRepository:
         logger.info(f"Retrieved {len(priorities_data)} priorities with CustomerTypeID = 1")
         return priorities_data
 
+    def get_phone_types(self, active_only: bool = False) -> List[Dict[str, Any]]:
+        """Get all phone types"""
+        cache_key = f"phone_types_{active_only}"
+        if cache_key in self._cache:
+            return self._cache[cache_key]
+        
+        try:
+            query = self.db.query(PhoneType)
+            if active_only:
+                query = query.filter((PhoneType.enabled == True) | (PhoneType.enabled.is_(None)))
+            
+            rows = query.order_by(PhoneType.name.asc()).all()
+            phone_types_data = []
+            
+            for row in rows:
+                phone_types_data.append({
+                    'id': row.id,
+                    'name': row.name,
+                    'enabled': row.enabled,
+                    'show_in_contact_page': row.show_in_contact_page,
+                    'modified_date': row.modified_date
+                })
+            
+            self._cache[cache_key] = phone_types_data
+            logger.info(f"Retrieved {len(phone_types_data)} phone types")
+            return phone_types_data
+            
+        except Exception as e:
+            logger.error(f"Error retrieving phone types: {e}", exc_info=True)
+            return []
+
     def create_state(self, data: Dict[str, Any]) -> Dict[str, Any]:
         """Create a new state"""
         state = State(**data)
@@ -905,6 +937,37 @@ class ReferenceDataRepository:
         self.db.commit()
         return True
 
+    def create_phone_type(self, data: Dict[str, Any]) -> Dict[str, Any]:
+        """Create a new phone type"""
+        phone_type = PhoneType(**data)
+        self.db.add(phone_type)
+        self.db.commit()
+        self.db.refresh(phone_type)
+        mapper = phone_type.__mapper__
+        return {mapper.get_property_by_column(column).key: getattr(phone_type, mapper.get_property_by_column(column).key) for column in phone_type.__table__.columns}
+
+    def update_phone_type(self, phone_type_id: int, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+        """Update an existing phone type"""
+        phone_type = self.db.query(PhoneType).filter(PhoneType.id == phone_type_id).first()
+        if not phone_type:
+            return None
+        for key, value in data.items():
+            if hasattr(phone_type, key):
+                setattr(phone_type, key, value)
+        self.db.commit()
+        self.db.refresh(phone_type)
+        mapper = phone_type.__mapper__
+        return {mapper.get_property_by_column(column).key: getattr(phone_type, mapper.get_property_by_column(column).key) for column in phone_type.__table__.columns}
+
+    def delete_phone_type(self, phone_type_id: int) -> bool:
+        """Delete a phone type"""
+        phone_type = self.db.query(PhoneType).filter(PhoneType.id == phone_type_id).first()
+        if not phone_type:
+            return False
+        self.db.delete(phone_type)
+        self.db.commit()
+        return True
+
     def clear_cache(self):
         """Clear the internal cache to force fresh data on next request"""
         self._cache.clear()

app/schemas/reference.py

@@ -196,6 +196,26 @@ class FOBUpdateIn(BaseModel):
     fob_description: Optional[str] = Field(None, description="FOB description / name")
 
 
+class PhoneTypeOut(BaseModel):
+    id: int = Field(..., description="Phone type ID")
+    name: str = Field(..., description="Phone type name")
+    enabled: Optional[bool] = Field(None, description="Enabled flag")
+    show_in_contact_page: Optional[bool] = Field(None, description="Show in contact page flag")
+    modified_date: Optional[datetime] = Field(None, description="Last modified date")
+
+
+class PhoneTypeCreateIn(BaseModel):
+    name: str = Field(..., description="Phone type name")
+    enabled: Optional[bool] = Field(None, description="Enabled flag")
+    show_in_contact_page: Optional[bool] = Field(None, description="Show in contact page flag")
+
+
+class PhoneTypeUpdateIn(BaseModel):
+    name: Optional[str] = Field(None, description="Phone type name")
+    enabled: Optional[bool] = Field(None, description="Enabled flag")
+    show_in_contact_page: Optional[bool] = Field(None, description="Show in contact page flag")
+
+
 class EstShipDateOut(BaseModel):
     est_ship_date_id: int = Field(..., description="Estimated ship date ID")
     est_ship_date_description: Optional[str] = Field(None, description="Estimated ship date description")
@@ -271,6 +291,7 @@ class ReferenceDataResponse(BaseModel):
     est_freights: Optional[List[EstFreightOut]] = Field([], description="List of Estimated Freight options")
     status_info: List[StatusInfoOut] = Field(..., description="Project status info")
     priorities: Optional[List[PriorityOut]] = Field([], description="List of priorities")
+    phone_types: Optional[List[PhoneTypeOut]] = Field([], description="List of phone types")
 
 class ProjectReferenceResponse(BaseModel):
     """Grouped reference data used on project creation forms"""
@@ -281,3 +302,4 @@ class ProjectReferenceResponse(BaseModel):
     status_info: List[StatusInfoOut] = Field(..., description="Project status info")
     purchase_prices: List[PurchasePriceOut] = Field(..., description="Purchase price options")
     rental_prices: List[RentalPriceOut] = Field(..., description="Rental price options")
+    phone_types: Optional[List[PhoneTypeOut]] = Field([], description="Phone type options")

app/services/reference_service.py

@@ -10,7 +10,7 @@ from app.schemas.reference import (
     BarrierSizeCreateIn, BarrierSizeUpdateIn, ProductApplicationCreateIn, ProductApplicationUpdateIn,
     FOBCreateIn, FOBUpdateIn, EstShipDateCreateIn, EstShipDateUpdateIn,
     EstFreightCreateIn, EstFreightUpdateIn, StatusInfoCreateIn, StatusInfoUpdateIn,
-    PriorityCreateIn, PriorityUpdateIn
+    PriorityCreateIn, PriorityUpdateIn, PhoneTypeOut, PhoneTypeCreateIn, PhoneTypeUpdateIn
 )
 from typing import List, Dict, Any, Optional
 import logging
@@ -76,6 +76,11 @@ class ReferenceDataService:
         priorities_data = self.repo.get_priorities(active_only, customer_type_id)
         return [PriorityOut(**p) for p in priorities_data]
 
+    def get_phone_types(self, active_only: bool = False) -> List[PhoneTypeOut]:
+        """Get all phone types as Pydantic models"""
+        phone_types_data = self.repo.get_phone_types(active_only)
+        return [PhoneTypeOut(**pt) for pt in phone_types_data]
+
     def create_state(self, data: StateCreateIn) -> StateOut:
         """Create a new state"""
         data_dict = data.model_dump()
@@ -344,6 +349,24 @@ class ReferenceDataService:
         """Delete a priority"""
         return self.repo.delete_priority(priority_id)
 
+    def create_phone_type(self, data: PhoneTypeCreateIn) -> PhoneTypeOut:
+        """Create a new phone type"""
+        data_dict = data.model_dump()
+        result = self.repo.create_phone_type(data_dict)
+        return PhoneTypeOut(**result)
+
+    def update_phone_type(self, phone_type_id: int, data: PhoneTypeUpdateIn) -> Optional[PhoneTypeOut]:
+        """Update an existing phone type"""
+        data_dict = data.model_dump(exclude_unset=True)
+        result = self.repo.update_phone_type(phone_type_id, data_dict)
+        if result:
+            return PhoneTypeOut(**result)
+        return None
+
+    def delete_phone_type(self, phone_type_id: int) -> bool:
+        """Delete a phone type"""
+        return self.repo.delete_phone_type(phone_type_id)
+
     def get_fobs(self, active_only: bool = True):
         """Get all FOBs as Pydantic models"""
         fobs_data = self.repo.get_fobs(active_only)
@@ -451,6 +474,8 @@ class ReferenceDataService:
                 response_kwargs['project_statuses'] = [ProjectStatusOut(**ps) for ps in all_data['project_statuses']]
             if 'priorities' in all_data:
                 response_kwargs['priorities'] = [PriorityOut(**p) for p in all_data['priorities']]
+            if 'phone_types' in all_data:
+                response_kwargs['phone_types'] = [PhoneTypeOut(**pt) for pt in all_data['phone_types']]
 
             return ReferenceDataResponse(**response_kwargs)
         except Exception as e: