Coverage for nexusLIMS/tui/common/db_utils.py: 100%

1""" 

2Database utilities for NexusLIMS TUI applications. 

3 

4Provides common database query patterns. 

5""" 

6 

7from typing import Any 

8 

9from sqlmodel import Session, select 

10 

11from nexusLIMS.db.models import Instrument, SessionLog 

12 

13 

14def check_uniqueness( 

15 session: Session, 

16 model: type, 

17 field_name: str, 

18 value: Any, 

19 exclude_pk: Any | None = None, 

20) -> bool: 

21 """ 

22 Check if a value is unique for a given field. 

23 

24 Parameters 

25 ---------- 

26 session : Session 

27 Active database session 

28 model : type 

29 SQLModel model class (e.g., Instrument) 

30 field_name : str 

31 Field name to check (e.g., "api_url") 

32 value : Any 

33 Value to check for uniqueness 

34 exclude_pk : Any | None 

35 Primary key value to exclude (for edit operations) 

36 

37 Returns 

38 ------- 

39 bool 

40 True if unique, False if duplicate exists 

41 

42 Examples 

43 -------- 

44 >>> from nexusLIMS.db.models import Instrument 

45 >>> with get_db_session() as session: 

46 ... is_unique = check_uniqueness( 

47 ... session, Instrument, "api_url", 

48 ... "https://example.com/api/tools/?id=42" 

49 ... ) 

50 """ 

51 # Skip check if value is None (for optional fields) 

52 if value is None: 

53 return True 

54 

55 # Build query 

56 field = getattr(model, field_name) 

57 statement = select(model).where(field == value) 

58 

59 # Exclude current record if editing 

60 if exclude_pk is not None: 

61 pk_field = getattr(model, model.__table__.primary_key.columns.keys()[0]) 

62 statement = statement.where(pk_field != exclude_pk) 

63 

64 # Check if any records exist 

65 existing = session.exec(statement).first() 

66 return existing is None 

67 

68 

69def get_session_log_count(session: Session, instrument_pid: str) -> int: 

70 """ 

71 Get count of session_log entries for an instrument. 

72 

73 Useful for warning users before deleting an instrument with associated data. 

74 

75 Parameters 

76 ---------- 

77 session : Session 

78 Active database session 

79 instrument_pid : str 

80 Instrument primary key 

81 

82 Returns 

83 ------- 

84 int 

85 Number of session_log entries 

86 

87 Examples 

88 -------- 

89 >>> with get_db_session() as session: 

90 ... count = get_session_log_count(session, "FEI-Titan-TEM") 

91 ... if count > 0: 

92 ... print(f"Warning: {count} session logs will be orphaned") 

93 """ 

94 statement = select(SessionLog).where(SessionLog.instrument == instrument_pid) 

95 results = session.exec(statement).all() 

96 return len(results) 

97 

98 

99def find_conflicting_instrument( 

100 session: Session, 

101 field_name: str, 

102 value: Any, 

103 exclude_pid: str | None = None, 

104) -> Instrument | None: 

105 """ 

106 Find an instrument that conflicts with a unique field value. 

107 

108 Parameters 

109 ---------- 

110 session : Session 

111 Active database session 

112 field_name : str 

113 Unique field name (e.g., api_url) 

114 value : Any 

115 Value to search for 

116 exclude_pid : str | None 

117 Instrument PID to exclude (for edit operations) 

118 

119 Returns 

120 ------- 

121 Instrument | None 

122 Conflicting instrument, or None if no conflict 

123 

124 Examples 

125 -------- 

126 >>> with get_db_session() as session: 

127 ... conflict = find_conflicting_instrument( 

128 ... session, "api_url", "https://example.com/api" 

129 ... ) 

130 ... if conflict: 

131 ... print(f"Already used by {conflict.instrument_pid}") 

132 """ 

133 if value is None: 

134 return None 

135 

136 field = getattr(Instrument, field_name) 

137 statement = select(Instrument).where(field == value) 

138 

139 if exclude_pid is not None: 

140 statement = statement.where(Instrument.instrument_pid != exclude_pid) 

141 

142 return session.exec(statement).first()