djangoadmin-wiki/DjangoAdmin_Tornado_NaiveVue/zh/volume4/tools/db-session.md

数据库会话

:::tip 说明 数据库会话通过 scoped_session 实现线程级隔离，Repository 层的写操作自动 commit，BaseHandler.on_finish() 兜底关闭会话。核心代码位于 src/core/database.py。 :::

核心架构

数据库会话管理位于 src/core/database.py。

scoped_session 隔离

from sqlalchemy.orm import scoped_session, sessionmaker

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
db = scoped_session(SessionLocal)

scoped_session 基于 thread-local 实现线程级隔离，每个线程持有独立的数据库会话，互不干扰。

会话使用

业务代码通过 BaseRepository 的 self.db 属性获取当前会话：

class BaseRepository:
    def __init__(self, db=None):
        self.db = db or _default_db  # 默认使用 scoped_session

会话生命周期

创建

Repository 实例化时自动获取会话（scoped_session 按线程自动创建）。

提交

Repository 层的写操作内部自动 db.commit()：

class BaseRepository:
    def create(self, obj):
        if hasattr(obj, "create_user"):
            obj.create_user = str(self._uid())
        self.db.add(obj)
        self.db.commit()  # 自动提交
        return obj

    def update(self, pk, data):
        data["update_user"] = str(self._uid())
        data["update_time"] = datetime.now()
        result = self.db.query(self.model).filter_by(id=pk).update(data)
        self.db.commit()  # 自动提交
        return bool(result)

关闭

Service 层的 try/finally 块确保 repo.close() 释放会话：

class BaseService:
    async def get_page(self, handler):
        try:
            # ... 业务逻辑 ...
            return R.page(handler, data=result, total=total, current=pageNo, size=pageSize)
        finally:
            self.repo.close()  # 释放会话

BaseHandler.on_finish() 兜底关闭会话，避免长连接泄漏：

class BaseHandler(tornado.web.RequestHandler):
    def on_finish(self):
        try:
            db.close()  # 兜底关闭（Service 层已关闭时为幂等 no-op）
        except Exception:
            pass

总结

数据库会话通过 scoped_session 实现线程级隔离，Repository 层写操作自动 commit()，Service 层 try/finally 确保 close()，BaseHandler.on_finish() 兜底关闭。业务代码无需手动管理事务。