[Python] sqlalchemy 'instance is not bound to a Session; attribute refresh operation cannot...

I'm working with FastAPI and SQLAlchemy. I've created a logic to collect Article data and then execute a background task.

routes.py

@public_router.get("/process")
def process_clustered_articles(db: DbSession, background_tasks: BackgroundTasks):
try:
articles = fetch_service.fetch_and_process_clustered_articles(db)
background_tasks.add_task(generate_images_for_clustered_articles_task, db=db, clustered_articles=articles[:30])
return {"message": "Clustered articles processed successfully"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))


fetch_service.py

def fetch_and_process_clustered_articles(self, db: DbSession, min_similar_articles: int = None) -> List[Article]:
"""
Fetch clustered articles and process them into the normalized schema.
"""
try:
# fetch data from source
clusters = self.clustered_dao.fetch_clustered_articles()

articles = []
for cluster in clusters:

# Convert and insert normalized article
normalized_article : ArticleCreate = self.convert_article_cluster_to_db_format(cluster)
if normalized_article and isinstance(normalized_article, ArticleCreate):
try:
logging.info(f"Processing article: {normalized_article}"))
article = self.article_dao.create(Article(**normalized_article.model_dump()), db)
if article:
articles.append(article)
logging.info(f"article id: {article.id} inserted successfully.")
except Exception as e:
"""
...
"""


return articles
except Exception as e:
logging.error(f"Error processing clustered articles: {e}")
raise RuntimeError(f"Error processing clustered articles: {e}")


article_dao

def create(self, article: Article, db: DbSession) -> Article:
"""
Create a new article in the database.
"""
db.add(article)
db.commit()
db.refresh(article)
return article


When I call '/process' endpoint, I get the error 'Instance <Article at 0x1468ed5e0> is not bound to a Session; attribute refresh operation cannot proceed (Background on this error at: https://sqlalche.me/e/20/bhk3)' during the background task execution. I couldn't debug it directly, so I created a test endpoint with almost identical logic. Running the test endpoint resulted in no errors and it worked correctly.

@public_router.post("/test_generation")
def test_generation(
db: DbSession,
background_tasks: BackgroundTasks,
article : ArticleCreate,
):
"""
Test image generation for a specific article.
"""
try:
article = article_dao.create(db, article)
background_tasks.add_task(generate_images_for_clustered_articles_task, db=db, clustered_articles=[article])
return {"message": "Image generation task started successfully"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))


What could be the reason why the original endpoint throws an error while the test endpoint doesn't? I believe this error shouldn't occur because the Article instance is inserted into the DB and then fetched again via a new query using the refresh function.

Continue reading...