feat(reference): add customer_type_id filtering to reference data endpoints

- Add customer_type_id query parameter to get_all_reference_data endpoint
- Add customer_type_id query parameter to get_company_types endpoint
- Add customer_type_id query parameter to get_lead_sources endpoint
- Refactor company_types repository method to use direct SQL queries with parameter binding
- Refactor lead_sources repository method to use direct SQL queries with parameter binding
- Update active_only filter logic to include NULL values for enabled/inactive flags
- Improve cross-version database compatibility by using text() queries instead of ORM
- Add proper field name-based mapping for query result handling
- Update cache key generation to include customer_type_id parameter
- Enhance logging messages for better debugging of query failures

app/controllers/reference.py

@@ -26,6 +26,7 @@ router = APIRouter(prefix="/api/v1/reference", tags=["reference-data"])
 @router.get("/all", response_model=ReferenceDataResponse)
 def get_all_reference_data(
     active_only: bool = Query(True, description="Return only active records"),
+    customer_type_id: Optional[int] = Query(None, description="Filter by customer type ID"),
     db: Session = Depends(get_db),
     current_user: Optional[CurrentUser] = Depends(get_current_user_optional)
 ):
@@ -48,7 +49,7 @@ def get_all_reference_data(
     **Note**: This endpoint is public but may have enhanced data for authenticated users.
     """
     service = ReferenceDataService(db)
-    return service.get_all_reference_data(active_only=active_only)
+    return service.get_all_reference_data(active_only=active_only, customer_type_id=customer_type_id)
 
 @router.get("/states", response_model=List[StateOut])
 def get_states(

app/db/repositories/reference_repo.py

@@ -43,7 +43,7 @@ class ReferenceDataRepository:
             return self._cache[cache_key]
             
         try:
-            # Use direct SQL query instead of stored procedure
+            # Use direct SQL query with proper parameter binding for cross-version compatibility
             query_str = """
             SELECT TOP (1000) [StateID], [CustomerTypeID], [Description], [Enabled]
             FROM [hs-prod3].[dbo].[States]
@@ -52,7 +52,7 @@ class ReferenceDataRepository:
             params = {}
             
             if active_only:
-                query_str += " AND [Enabled] = 1"
+                query_str += " AND ([Enabled] = 1 OR [Enabled] IS NULL)"
             
             if customer_type_id is not None:
                 query_str += " AND [CustomerTypeID] = :customer_type_id"
@@ -61,10 +61,10 @@ class ReferenceDataRepository:
             sql_query = text(query_str)
             result = self.db.execute(sql_query, params)
             
+            states_data = []
             if result.returns_rows:
-                # Map the query results to our expected schema
-                states_data = []
                 for row in result.fetchall():
+                    # Use field name-based mapping for proper structure
                     row_dict = dict(row._mapping)
                     # Map query output to our API format
                     state_data = {
@@ -80,7 +80,7 @@ class ReferenceDataRepository:
                 return states_data
                 
         except Exception as e:
-            logger.warning(f"Direct query failed: {e}", exc_info=True)
+            logger.warning(f"Direct states query failed: {e}", exc_info=True)
             
         return []
 
@@ -118,30 +118,46 @@ class ReferenceDataRepository:
         if cache_key in self._cache:
             return self._cache[cache_key]
             
-        query = self.db.query(CompanyType)
-        if active_only:
-            # The CompanyTypes table stores an "Inactive" flag mapped to CompanyType.enabled.
-            # When active_only is True, return only active records (Inactive == False).
-            query = query.filter(CompanyType.enabled == False)
-        
-        if customer_type_id is not None:
-            query = query.filter(CompanyType.customer_type_id == customer_type_id)
-        
-        company_types = query.all()
-        # Align keys to CompanyTypeOut schema: company_type_id, customer_type_id, description, inactive
-        company_types_data = [
-            {
-                'company_type_id': ct.company_type_id,
-                'customer_type_id': ct.customer_type_id,
-                'description': ct.description,
-                'inactive': ct.enabled,
-            }
-            for ct in company_types
-        ]
-        
-        self._cache[cache_key] = company_types_data
-        logger.info(f"Retrieved {len(company_types_data)} company types")
-        return company_types_data
+        try:
+            # Use direct SQL query for better compatibility across Python versions
+            query_str = """
+            SELECT [CompanyTypeID], [CustomerTypeID], [Description], [Inactive]
+            FROM [CompanyTypes]
+            WHERE 1=1
+            """
+            params = {}
+            
+            if active_only:
+                # When active_only is True, return only active records (Inactive == 0 or NULL)
+                query_str += " AND ([Inactive] = 0 OR [Inactive] IS NULL)"
+            
+            if customer_type_id is not None:
+                query_str += " AND [CustomerTypeID] = :customer_type_id"
+                params["customer_type_id"] = customer_type_id
+            
+            sql_query = text(query_str)
+            result = self.db.execute(sql_query, params)
+            
+            company_types_data = []
+            if result.returns_rows:
+                for row in result.fetchall():
+                    # Use field name-based mapping for proper structure
+                    row_dict = dict(row._mapping)
+                    company_type_data = {
+                        'company_type_id': row_dict.get('CompanyTypeID', None),
+                        'customer_type_id': row_dict.get('CustomerTypeID', None),
+                        'description': row_dict.get('Description', None),
+                        'inactive': row_dict.get('Inactive', False),
+                    }
+                    company_types_data.append(company_type_data)
+            
+            self._cache[cache_key] = company_types_data
+            logger.info(f"Retrieved {len(company_types_data)} company types via direct query")
+            return company_types_data
+            
+        except Exception as e:
+            logger.warning(f"Direct company types query failed: {e}", exc_info=True)
+            return []
 
     def get_lead_sources(self, active_only: bool = True, customer_type_id: Optional[int] = None) -> List[Dict[str, Any]]:
         """Get all lead generation sources, optionally filtered by customer_type_id"""
@@ -149,27 +165,46 @@ class ReferenceDataRepository:
         if cache_key in self._cache:
             return self._cache[cache_key]
             
-        query = self.db.query(LeadGeneratedFrom)
-        if active_only:
-            query = query.filter((LeadGeneratedFrom.enabled == True) | (LeadGeneratedFrom.enabled.is_(None)))
-        
-        if customer_type_id is not None:
-            query = query.filter(LeadGeneratedFrom.customer_type_id == customer_type_id)
-        
-        lead_sources = query.order_by(LeadGeneratedFrom.description.asc()).all()
-        lead_sources_data = [
-            {
-                'lead_generated_from_id': ls.lead_generated_from_id,
-                'customer_type_id': ls.customer_type_id,
-                'description': ls.description,
-                'enabled': ls.enabled
-            }
-            for ls in lead_sources
-        ]
-        
-        self._cache[cache_key] = lead_sources_data
-        logger.info(f"Retrieved {len(lead_sources_data)} lead sources")
-        return lead_sources_data
+        try:
+            # Use direct SQL query for better compatibility across Python versions
+            query_str = """
+            SELECT [LeadGeneratedFromID], [CustomerTypeID], [Description], [Enabled]
+            FROM [LeadGeneratedFroms]
+            WHERE 1=1
+            """
+            params = {}
+            
+            if active_only:
+                query_str += " AND ([Enabled] = 1 OR [Enabled] IS NULL)"
+            
+            if customer_type_id is not None:
+                query_str += " AND [CustomerTypeID] = :customer_type_id"
+                params["customer_type_id"] = customer_type_id
+            
+            
+            sql_query = text(query_str)
+            result = self.db.execute(sql_query, params)
+            
+            lead_sources_data = []
+            if result.returns_rows:
+                for row in result.fetchall():
+                    # Use field name-based mapping for proper structure
+                    row_dict = dict(row._mapping)
+                    lead_source_data = {
+                        'lead_generated_from_id': row_dict.get('LeadGeneratedFromID', None),
+                        'customer_type_id': row_dict.get('CustomerTypeID', None),
+                        'description': row_dict.get('Description', None),
+                        'enabled': row_dict.get('Enabled', True)
+                    }
+                    lead_sources_data.append(lead_source_data)
+            
+            self._cache[cache_key] = lead_sources_data
+            logger.info(f"Retrieved {len(lead_sources_data)} lead sources via direct query")
+            return lead_sources_data
+            
+        except Exception as e:
+            logger.warning(f"Direct lead sources query failed: {e}", exc_info=True)
+            return []
 
     def get_payment_terms(self, active_only: bool = True) -> List[Dict[str, Any]]:
         """Get all payment terms"""
@@ -393,13 +428,16 @@ class ReferenceDataRepository:
         """Get all project statuses - Temporarily disabled due to model issues"""
         return []  # Return empty list for now
 
-    def get_all_reference_data(self, active_only: bool = True) -> Dict[str, List[Dict[str, Any]]]:
+    def get_all_reference_data(self, active_only: bool = True, customer_type_id: Optional[int] = None) -> Dict[str, List[Dict[str, Any]]]:
         """Get all reference data in a single call for efficiency"""
+        # Use customer_type_id = 1 as default if not provided (for backward compatibility)
+        default_customer_type_id = customer_type_id if customer_type_id is not None else 1
+        
         return {
-            'states': self.get_states(active_only),
+            'states': self.get_states(active_only, default_customer_type_id),
             'countries': self.get_countries(active_only),
-            'company_types': self.get_company_types(active_only),
-            'lead_sources': self.get_lead_sources(active_only),
+            'company_types': self.get_company_types(active_only, default_customer_type_id),
+            'lead_sources': self.get_lead_sources(active_only, default_customer_type_id),
             'payment_terms': self.get_payment_terms(active_only),
             'purchase_prices': self.get_purchase_prices(active_only),
             'rental_prices': self.get_rental_prices(active_only),
@@ -409,7 +447,7 @@ class ReferenceDataRepository:
             'est_ship_dates': self.get_est_ship_dates(active_only),
             '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
+            'priorities': self.get_priorities(active_only, default_customer_type_id),
             'phone_types': self.get_phone_types(active_only),
         }
 
@@ -447,31 +485,46 @@ class ReferenceDataRepository:
             return []
 
     def get_priorities(self, active_only: bool = True, customer_type_id: Optional[int] = None) -> List[Dict[str, Any]]:
-        """Get all priorities, filtered by CustomerTypeID = 1"""
+        """Get all priorities, optionally filtered by customer_type_id"""
         cache_key = f"priorities_{active_only}_{customer_type_id}"
         if cache_key in self._cache:
             return self._cache[cache_key]
             
-        # Use direct SQL query as specified: SELECT PriorityID, CustomerTypeID, Description FROM Priorities WHERE CustomerTypeID = 1
-        query = text("""
+        try:
+            # Use direct SQL query with proper parameter binding
+            query_str = """
             SELECT [PriorityID], [CustomerTypeID], [Description]
-            FROM [Priorities] 
-            WHERE [CustomerTypeID] = 1
-        """)
-        
-        result = self.db.execute(query)
-        priorities_data = []
-        
-        for row in result.fetchall():
-            priorities_data.append({
-                'priority_id': row[0],  # PriorityID
-                'customer_type_id': row[1],  # CustomerTypeID  
-                'description': row[2]  # Description
-            })
-        
-        self._cache[cache_key] = priorities_data
-        logger.info(f"Retrieved {len(priorities_data)} priorities with CustomerTypeID = 1")
-        return priorities_data
+            FROM [Priorities]
+            WHERE 1=1
+            """
+            params = {}
+            
+            if customer_type_id is not None:
+                query_str += " AND [CustomerTypeID] = :customer_type_id"
+                params["customer_type_id"] = customer_type_id
+            
+            sql_query = text(query_str)
+            result = self.db.execute(sql_query, params)
+            
+            priorities_data = []
+            if result.returns_rows:
+                for row in result.fetchall():
+                    # Use field name-based mapping for proper structure
+                    row_dict = dict(row._mapping)
+                    priority_data = {
+                        'priority_id': row_dict.get('PriorityID', None),
+                        'customer_type_id': row_dict.get('CustomerTypeID', None),
+                        'description': row_dict.get('Description', None)
+                    }
+                    priorities_data.append(priority_data)
+            
+            self._cache[cache_key] = priorities_data
+            logger.info(f"Retrieved {len(priorities_data)} priorities via direct query")
+            return priorities_data
+            
+        except Exception as e:
+            logger.warning(f"Direct priorities query failed: {e}", exc_info=True)
+            return []
 
     def get_phone_types(self, active_only: bool = False) -> List[Dict[str, Any]]:
         """Get all phone types"""

app/services/reference_service.py

@@ -425,7 +425,7 @@ class ReferenceDataService:
         )
 
  
-    def get_all_reference_data(self, active_only: bool = True) -> ReferenceDataResponse:
+    def get_all_reference_data(self, active_only: bool = True, customer_type_id: Optional[int] = None) -> ReferenceDataResponse:
         """
         Get all reference data in a single response for efficiency.
         This is useful for frontend applications that need to populate
@@ -433,7 +433,7 @@ class ReferenceDataService:
         """
         try:
             # Get all data from repository
-            all_data = self.repo.get_all_reference_data(active_only)
+            all_data = self.repo.get_all_reference_data(active_only, customer_type_id)
             
             # Transform to Pydantic models using keys provided by the repository.
             # The repository returns a dict with a subset of keys; map only those.