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

1from typing import List, TypeVar, Protocol, Any 

2from sqlalchemy.orm import Session 

3from sqlalchemy import Column 

4 

5 

6class HasIdColumn(Protocol): 

7 """Protocol for models with an id column that supports SQLAlchemy operations""" 

8 

9 id: Column 

10 

11 

12T = TypeVar("T") 

13 

14 

15def fetch_related_items( 

16 session: Session, 

17 model_class: Any, 

18 ids: List[int], 

19 error_name: str, 

20 require_all: bool = True, 

21) -> List[T]: 

22 """ 

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

24 

25 Args: 

26 session: SQLAlchemy session 

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

28 ids: List of IDs to fetch 

29 error_name: Name to use in error messages 

30 require_all: Whether to verify all requested IDs were found 

31 

32 Returns: 

33 List of found items 

34 

35 Raises: 

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

37 """ 

38 if not ids: 

39 return [] 

40 

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

42 

43 # Optionally verify all requested items were found 

44 if require_all: 

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

46 missing_ids = set(ids) - found_ids 

47 if missing_ids: 

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

49 

50 return items