Проблема с SCOPED_SESSION sqlalchemy в python 3.11
Я столкнулся с проблемой, которую не могу понять, как решить.
Есть приложение на flask. Для него написано множество тестов. Есть фиксы db_session и db_scoped_session, включая глобальную переменную SCOPED_SESSION. Обработчики реализованы в виде представлений.
Упомянутые фиксы - db_session и db_scoped_session и объект SCOPED_SESSION - это разные объекты, судя по id. Это справедливо как для ранней - python 3.9, так и для новой - 3.11, на которую мы перешли. С ранней версией python проблем не возникло, а с версией python 3.11 тесты падали, основной причиной чего стала ошибка - DetachedInstanceError.
Читайте подробнее об этой ошибке.
Она возникает, когда мы пытаемся получить доступ к объекту, отделенному от его сессии.
Когда объект загружается в сессию, он находится в персистентном состоянии - это означает, что любые изменения отслеживаются сессией. Когда сессия закрывается, объекты, связанные с ней, переходят в состояние detached, что и приводит к вышеуказанной ошибке.
Полагая, что ошибка связана с тем, что мы используем разные сессии для фикстур и для самих представлений, я перенес все фикстуры в одну сессию. Это не помогло:)
Сессия очищается в вызываемом методе teardown_appcontext - объектом __remove_session. Удаление этого метода устранило эту проблему, но вызвало новые проблемы другого характера.
Версия sqlalchemy - 1.3.28.
Подскажите, пожалуйста, в чем может быть проблема.
Пример
# view.py
class SomeView:
def get_one(self, id):
return some_thing
# flask_app.py
from flask_admin import Admin
from view import SomeView
admin = Admin(...)
admin.add_view(SomeView(...), endpoint='/some_endpoint/')
# worker.py
class Worker(...):
def _init_application(self):
self._flask_app = Flask(...)
...
self._flask_app.teardown_appcontext(self._remove_session)
...
def _remove_session(self, response_or_exc):
SCOPED_SESSION.remove()
return response_or_exc
# test_case.py
def test_edit_get(test_object, client:FlaskClient, request_headers) -> None:
response: TestResponse = client.get(
f'/some_endpoint/?id={test_object.id}',
headers=request_headers
)
assert response.status_code == 200
# Error coming here
print(test_object.id) # DetachedInstanceError