'How to use Objects outside session when fetched from database using SQLAlchemy?
I am using Flask REST Api server to communicate with my front-end. I am using SQLAlchemy (Not Flask-SQLAlchemy) to communicate with SQLite DB file. My problem is that I get
sqlalchemy.orm.exc.DetachedInstanceError: Instance <x> is not bound to a Session; attribute refresh operation cannot proceed (Background on this error at: https://sqlalche.me/e/14/bhk3)
error when I add records and fetch them from db and read them outside the session. I have no idea how to do that without getting the mentioned error. I am using python version 3.8.10. I want to keep reusing the scoped_session but after every commit call I want that the objects get detached from the session so that I can read the objects outside the session.
I tried to achieve what I want by adding self.session.expire_on_commit = False in DbSession context manager but when I do that I get duplicate TestRecord object. So if I insert 2 TestRecord objects, I get 4 back when I call get_test_records() and those 2 objects that are inserted are fetched again. I just want that I only get the 2 inserted TestRecord objects back.
requirements.txt:
Flask==2.0.3
Flask-RESTful==0.3.9
SQLAlchemy==1.4.31
db.py:
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session, Session
import os
engine = create_engine(
f"sqlite:///./test.db",
connect_args={"check_same_thread": False},
echo=True,
)
SessionLocal = scoped_session(
sessionmaker(autoflush=True, autocommit=False, bind=engine)
)
class DbSession:
def __init__(self):
self.session = None
def __enter__(self) -> Session:
self.session = SessionLocal()
return self.session
def __exit__(self, type, value, traceback):
# commit the changes in the session to the db.
self.session.commit()
TestRecord.py
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Integer, String, Column, DateTime
from datetime import datetime
Base = declarative_base()
class TestRecord(Base):
__tablename__ = "test_records"
id = Column(Integer, primary_key=True, index=True)
first_name = Column(String, nullable=False, index=True)
timestamp = Column(DateTime, nullable=False, default=datetime.utcnow)
last_name = Column(String, nullable=False, index=True)
def __repr__(self):
return f"<TestRecord(id={self.id}, first_name={self.first_name}, timestamp={self.timestamp}, last_name={self.last_name})>"
crud.py:
from db import DbSession
from test_record import TestRecord
from datetime import datetime
def insert_multiple_test_records(test_records) -> bool:
try:
with DbSession() as session:
session.add_all(test_records)
return True
except Exception as e:
print(e)
return False
def get_test_records():
try:
with DbSession() as session:
res = session.query(TestRecord).all()
return res
except Exception as e:
print(e)
return []
main.py:
from test_record import TestRecord, Base
from datetime import datetime, timedelta
from db import create_tables
if __name__ == '__main__':
Base.metadata.create_all(bind=engine, checkfirst=True)
t = datetime.now()
dt = timedelta(minutes=1)
lst = [
TestRecord(first_name="Test", timestamp=t+dt+dt+dt+dt+dt+dt, last_name="last0"),
TestRecord(first_name="Test", timestamp=t+dt+dt+dt+dt+dt, last_name="last1"),
TestRecord(first_name="Test", timestamp=t+dt+dt+dt+dt, last_name="last2"),
TestRecord(first_name="Test2", timestamp=t+dt+dt+dt, last_name="last3"),
TestRecord(first_name="Test2", timestamp=t+dt+dt, last_name="last4"),
TestRecord(first_name="Test3", timestamp=t+dt, last_name="last5"),
TestRecord(first_name="Test4", timestamp=t, last_name="last6"),
]
print(insert_multiple_test_records(lst))
print(get_test_records()) # --> this causes the DetachedInstanceError
Sources
This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.
Source: Stack Overflow
| Solution | Source |
|---|
