Экземпляр SQLALCHEMY 'не связан с сеансом; ОБРАЩЕНИЕ АТРИБИТА ⇐ Python

[Anonymous]

Я работаю с Fastapi и Sqlalchemy. Я создал логику для сбора данных статьи, а затем выполнить фоновую задачу.

Код: Выделить всё

@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
< /code>
Когда я звоню ' /process' endpoint, я получаю сообщение об ошибке
'Instance is not bound to a Session; attribute refresh operation cannot proceed (Background on this error at: https://sqlalche.me/e/20/bhk3)'< /code>
во время выполнения фоновой задачи. Я не мог отлаживать его напрямую, поэтому я создал конечную точку тестирования с почти идентичной логикой. Запуск конечной точки теста не привел к ошибкам, и она работала правильно. < /P>
@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))
< /code>
Что может быть причиной, по которой исходная конечная точка бросает ошибку, а конечная точка теста - нет? Я полагаю, что эта ошибка не должна происходить, потому что экземпляр статьи вставлен в БД, а затем снова получена через новый запрос, используя функцию обновления.

Подробнее здесь: https://stackoverflow.com/questions/796 ... ation-cann