Coverage for postrfp / ref / service / helpers.py: 100%

1from typing import TypeVar, Any 

2from sqlalchemy.orm import Session 

3 

4 

5T = TypeVar("T") 

6 

7 

8def fetch_related_items( 

9 session: Session, 

10 model_class: Any, 

11 ids: list[int], 

12 error_name: str, 

13 require_all: bool = True, 

14) -> list[T]: 

15 """ 

16 Fetch related items in a single query and verify all were found 

17 

18 Args: 

19 session: SQLAlchemy session 

20 model_class: SQLAlchemy model class to query (must have an 'id' attribute) 

21 ids: List of IDs to fetch 

22 error_name: Name to use in error messages 

23 require_all: Whether to verify all requested IDs were found 

24 

25 Returns: 

26 List of found items 

27 

28 Raises: 

29 ValueError: If require_all is True and any requested IDs were not found 

30 """ 

31 if not ids: 

32 return [] 

33 

34 items = session.query(model_class).filter(model_class.id.in_(ids)).all() 

35 

36 # Optionally verify all requested items were found 

37 if require_all: 

38 found_ids = {item.id for item in items} 

39 missing_ids = set(ids) - found_ids 

40 if missing_ids: 

41 raise ValueError(f"{error_name} with IDs {missing_ids} not found") 

42 

43 return items