'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