Coverage for nexusLIMS/db/session_handler.py: 100%
59 statements
« prev ^ index » next coverage.py v7.11.3, created at 2026-03-24 05:23 +0000
1"""Classes and methods to interact with sessions from the NexusLIMS database."""
3import logging
4from datetime import datetime as dt
5from typing import Tuple
7from sqlalchemy.orm import selectinload
8from sqlmodel import Session as DBSession
9from sqlmodel import select
11from nexusLIMS.db.engine import get_engine
12from nexusLIMS.db.enums import EventType, RecordStatus
13from nexusLIMS.db.models import Instrument, SessionLog
14from nexusLIMS.utils.time import current_system_tz
16_logger = logging.getLogger(__name__)
19class Session:
20 """
21 A representation of a session in the NexusLIMS database.
23 A record of an individual session as read from the Nexus Microscopy
24 facility session database. Created by combining two
25 :py:class:`~nexusLIMS.db.models.SessionLog` objects with status
26 ``"TO_BE_BUILT"``.
28 Parameters
29 ----------
30 session_identifier
31 The unique identifier for an individual session on an instrument
32 instrument
33 An object representing the instrument associated with this session
34 dt_range
35 A tuple of two :py:class:`~datetime.datetime` objects representing the start
36 and end of this session )in that order
37 user : str
38 The username associated with this session (may not be trustworthy)
39 """
41 def __init__(
42 self,
43 session_identifier: str,
44 instrument: Instrument,
45 dt_range: Tuple[dt, dt],
46 user: str,
47 ):
48 self.session_identifier = session_identifier
49 self.instrument = instrument
50 self.dt_from, self.dt_to = dt_range
51 self.user = user
53 def __repr__(self):
54 """Return custom representation of a Session."""
55 return (
56 f"{self.dt_from.isoformat()} to {self.dt_to.isoformat()} on "
57 f"{self.instrument.name}"
58 )
60 def update_session_status(self, status: RecordStatus):
61 """
62 Update the status of this Session in the NexusLIMS database.
64 Specifically, update the ``record_status`` in any session logs for this
65 :py:class:`~nexusLIMS.db.session_handler.Session`.
67 Parameters
68 ----------
69 status : RecordStatus
70 The new status for this session (type-safe enum value)
72 Returns
73 -------
74 success : bool
75 Whether the update operation was successful
76 """
77 with DBSession(get_engine()) as session:
78 statement = select(SessionLog).where(
79 SessionLog.session_identifier == self.session_identifier
80 )
81 logs = session.exec(statement).all()
82 for log in logs:
83 log.record_status = status
84 session.commit()
85 return True
87 def insert_record_generation_event(self) -> dict:
88 """
89 Insert record generation event to session log.
91 Insert a log for this session into the session database with
92 ``event_type`` `"RECORD_GENERATION"` and the current time (with local
93 system timezone) as the timestamp.
95 Returns
96 -------
97 res : dict
98 A dictionary containing the inserted log information
99 """
100 _logger.debug("Logging RECORD_GENERATION for %s", self.session_identifier)
102 log = SessionLog(
103 session_identifier=self.session_identifier,
104 instrument=self.instrument.instrument_pid,
105 timestamp=dt.now(tz=current_system_tz()),
106 event_type=EventType.RECORD_GENERATION,
107 user="nexuslims",
108 record_status=RecordStatus.WAITING_FOR_END,
109 )
111 with DBSession(get_engine()) as session:
112 session.add(log)
113 session.commit()
114 session.refresh(log)
116 _logger.debug(
117 "Confirmed RECORD_GENERATION insertion for %s",
118 self.session_identifier,
119 )
121 return {
122 "id_session_log": log.id_session_log,
123 "event_type": log.event_type.value,
124 "session_identifier": log.session_identifier,
125 "timestamp": log.timestamp,
126 }
129def get_sessions_to_build() -> list[Session]:
130 """
131 Get list of sessions that need to be built from the NexusLIMS database.
133 Query the NexusLIMS database for pairs of logs with status
134 ``TO_BE_BUILT`` and return the information needed to build a record for
135 that session.
137 Returns
138 -------
139 sessions : list[Session]
140 A list of :py:class:`~nexusLIMS.db.session_handler.Session` objects
141 containing the sessions that need their record built. Will be an
142 empty list if there's nothing to do.
143 """
144 sessions = []
146 with DBSession(get_engine()) as db_session:
147 # Query for all TO_BE_BUILT logs with eager loading of instrument relationship
148 statement = (
149 select(SessionLog)
150 .where(SessionLog.record_status == RecordStatus.TO_BE_BUILT)
151 .options(selectinload(SessionLog.instrument_obj))
152 )
153 session_logs = db_session.exec(statement).all()
155 # Separate START and END logs
156 start_logs = [sl for sl in session_logs if sl.event_type == EventType.START]
157 end_logs = [sl for sl in session_logs if sl.event_type == EventType.END]
159 for start_l in start_logs:
160 # for every log that has a 'START', there should be one corresponding
161 # log with 'END' that has the same session identifier. If not,
162 # the database is in an inconsistent state and we should know about it
163 el_list = [
164 el
165 for el in end_logs
166 if el.session_identifier == start_l.session_identifier
167 ]
168 if len(el_list) != 1:
169 msg = (
170 "There was not exactly one 'END' log for this 'START' log; "
171 f"len(el_list) was {len(el_list)}; sl was {start_l}; el_list "
172 f"was {el_list}"
173 )
174 raise ValueError(msg)
176 end_l = el_list[0]
177 # Use relationship to get Instrument object (eagerly loaded above)
178 session = Session(
179 session_identifier=start_l.session_identifier,
180 instrument=start_l.instrument_obj, # Relationship navigation!
181 dt_range=(start_l.timestamp, end_l.timestamp), # No fromisoformat()!
182 user=start_l.user,
183 )
184 sessions.append(session)
186 _logger.info("Found %i new sessions to build", len(sessions))
187 return sessions
190def get_all_session_logs() -> list[SessionLog]:
191 """
192 Fetch all session logs from the database and return SessionLogs.
194 Returns
195 -------
196 session_logs : list[SessionLog]
197 A list of all SessionLog objects from the database, ordered by timestamp.
198 Will be an empty list if there are no session logs.
199 """
200 with DBSession(get_engine()) as db_session:
201 # Query for all session logs, ordered by timestamp
202 statement = select(SessionLog).order_by(SessionLog.timestamp)
203 session_logs = list(db_session.exec(statement).all())
205 _logger.info("Found %i session logs in database", len(session_logs))
206 return session_logs