99import ispyb
1010import pytest
1111from ispyb .sqlalchemy import BLSession , ExperimentType , Person , Proposal , url
12- from sqlalchemy import Engine , RootTransaction , and_ , create_engine , event , select
12+ from sqlalchemy import Engine , and_ , create_engine , event , select as sa_select
1313from sqlalchemy .exc import InterfaceError
14- from sqlalchemy .ext .declarative import DeclarativeMeta
15- from sqlalchemy .orm import Session as SQLAlchemySession , sessionmaker
16- from sqlmodel import Session as SQLModelSession , SQLModel
14+ from sqlalchemy .orm import (
15+ Session as SQLAlchemySession ,
16+ SessionTransaction ,
17+ sessionmaker ,
18+ )
19+ from sqlmodel import Session as SQLModelSession , SQLModel , select as sm_select
1720
1821from murfey .util .db import Session as MurfeySession
1922
@@ -121,33 +124,36 @@ class ISPyBTableValues:
121124 }
122125
123126
124- SQLAlchemyTable = TypeVar ("SQLAlchemyTable" , bound = DeclarativeMeta )
127+ SQLTable = TypeVar ("SQLTable" )
125128
126129
127130def get_or_create_db_entry (
128131 session : SQLAlchemySession | SQLModelSession ,
129- table : Type [SQLAlchemyTable ],
130- lookup_kwargs : dict [str , Any ] = {} ,
131- insert_kwargs : dict [str , Any ] = {} ,
132- ) -> SQLAlchemyTable :
132+ table : Type [SQLTable ],
133+ lookup_kwargs : dict [str , Any ] | None = None ,
134+ insert_kwargs : dict [str , Any ] | None = None ,
135+ ) -> SQLTable :
133136 """
134137 Helper function to facilitate looking up or creating SQLAlchemy table entries.
135138 Returns the entry if a match based on the lookup criteria is found, otherwise
136139 creates and returns a new entry.
137140 """
138141
142+ lookup_kwargs = lookup_kwargs or {}
143+ insert_kwargs = insert_kwargs or {}
144+
139145 # if lookup kwargs are provided, check if entry exists
140146 if lookup_kwargs :
141147 conditions = [
142148 getattr (table , key ) == value for key , value in lookup_kwargs .items ()
143149 ]
144150 # Use 'exec()' for SQLModel sessions
145151 if isinstance (session , SQLModelSession ):
146- entry = session .exec (select (table ).where (and_ (* conditions ))).first ()
152+ entry = session .exec (sm_select (table ).where (and_ (* conditions ))).first ()
147153 # Use 'execute()' for SQLAlchemy sessions
148154 elif isinstance (session , SQLAlchemySession ):
149155 entry = (
150- session .execute (select (table ).where (and_ (* conditions )))
156+ session .execute (sa_select (table ).where (and_ (* conditions )))
151157 .scalars ()
152158 .first ()
153159 )
@@ -166,13 +172,17 @@ def get_or_create_db_entry(
166172
167173
168174def restart_savepoint (
169- session : SQLAlchemySession | SQLModelSession , transaction : RootTransaction
175+ session : SQLAlchemySession | SQLModelSession , transaction : SessionTransaction
170176):
171177 """
172178 Re-establish a SAVEPOINT after a nested transaction is committed or rolled back.
173179 This helps to maintain isolation across different test cases.
174180 """
175- if transaction .nested and not transaction ._parent .nested :
181+ if (
182+ transaction .nested
183+ and transaction ._parent is not None
184+ and not transaction ._parent .nested
185+ ):
176186 session .begin_nested ()
177187
178188
@@ -255,7 +265,7 @@ def seed_ispyb_db(ispyb_db_session_factory):
255265 except InterfaceError :
256266 # If this fails in the GitHub test environment, raise it as a genuine error
257267 if os .getenv ("GITHUB_ACTIONS" ) == "true" :
258- raise InterfaceError
268+ raise
259269 pytest .skip ("ISPyB database has not been set up; skipping test" )
260270
261271
0 commit comments