时间戳

首页 > 解决方案 > 使用 SQLAlchemy 动态附加 SQLite 数据库

python - 使用 SQLAlchemy 动态附加 SQLite 数据库

问题描述

我已经能够成功地让 SQLAlchemy 和 SQLite 一起工作以达到垂直分区的目的（将某些表分区到不同的数据库中）。

我遇到的问题是应用程序的某些部分不需要分区表，如果我在那些实际上不需要分区表的会话中通过“附加”命令加载它们，那么附加的数据库将被有效锁定，从而消除分区的全部好处。

我已经尝试使用以下代码，如果 SQLALchemy 对象继承自某个基类，那么我会为会话附加数据库，但这似乎不像我预期的那样工作：

class LazyAttachDatabaseSession(orm.Session):
    def __init__(self, theEngine, dbpath, **kw):
        super().__init__(**kw)
        self._engine = theEngine
        self._dbpath = dbpath
        self._logger = logging.getLogger(__name__)

    def get_bind(self, mapper=None, clause=None):
        if mapper is not None and issubclass(mapper.class_, _Realtime):
            self._engine.execute(f"ATTACH DATABASE '{self._dbpath}' as 'realtime'")
        return self._engine

我将 Session 对象配置如下：

    _Session = orm.sessionmaker(theEngine=engine,
                                class_=LazyAttachDatabaseSession,
                                dbpath=realtimeDBPath)

但由于各种原因，这不起作用。例如，我收到“数据库已附加”错误

标签： pythonsqlitesqlalchemy

解决方案

如果存在这样的事情，“问题”是 SQLite 连接没有被池化，因此 attach 命令的效果是短暂的/瞬态的。

假设您只希望每个进程/连接/线程设置一次，修改代码如下：

class LazyAttachDatabaseSession(orm.Session):
    isAttached = False #NOTE: New class-level variable

    def __init__(self, theEngine, dbpath, **kw):
        super().__init__(**kw)
        self._engine = theEngine
        self._dbpath = dbpath
        self._logger = logging.getLogger(__name__)

    def get_bind(self, mapper=None, clause=None):
        if not LazyAttachDatabaseSession.isAttached and mapper is not None and issubclass(mapper.class_, _Realtime):
            self._logger.info("Attaching %s database as 'realtime'",self._dbpath)
            self._engine.execute(f"ATTACH DATABASE '{self._dbpath}' as 'realtime'")
            LazyAttachDatabaseSession.isAttached = True
        return self._engine

并配置引擎如下：

    engine: sa.engine.Engine = create_engine(url,
                                             echo=debug,
                                             poolclass=sa.pool.SingletonThreadPool,
                                             connect_args={'timeout': 30})

注意“poolclass”参数。如果实际需要，这应该只附加“实时”数据库。

推荐阅读