Sqlalchemy 可以很好地处理多个附加的 SQLite 数据库文件吗？

Posted 2023-03-22

【中文标题】Sqlalchemy 可以很好地处理多个附加的 SQLite 数据库文件吗？

【英文标题】：Can Sqlalchemy work well with multiple attached SQLite database files?

【发布时间】：2014-05-27 00:50:38

【问题描述】：

可以使用 'ATTACH' 语句将多个 SQLite 数据库连接在一起并共同使用它们。每个 SQLite 文件中的表都可以使用特定于模式/文件的关键字来引用。这应该允许您通过文件范围来同时处理具有相同名称的多个表。我在这里阅读了一个非常好的教程：

http://longweekendmobile.com/2010/05/29/how-to-attach-multiple-sqlite-databases-together/

似乎我应该能够使用 SQLAlchemy 的 Table 'schema' 关键字来区分与多个文件的连接。当我去寻找一种将 SQLAlchemy 与通过 ATTACH 连接的 SQLite 数据库一起使用的方法时，这是我找到的唯一示例。不幸的是，它已经过时并且似乎不适用于当前版本。

https://groups.google.com/forum/#!topic/sqlalchemy/QXqs4M2MjbY

我尝试使用声明性类等更新该示例。这是我的尝试：

from sqlalchemy import *
from sqlalchemy.orm import * 
from sqlalchemy.ext.declarative import *
#from sqlalchemy.pool import SingletonThreadPool

metadata = MetaData(object)
DeclarativeBase = declarative_base(metadata=metadata)

##########################################################################
# Classes
##########################################################################

class A(DeclarativeBase):
    __table__              = Table('A', DeclarativeBase.metadata,
                                   Column('id', Integer, primary_key=True, index=True, autoincrement=True),
                                   Column('col_a', Integer, index=True))

class B(DeclarativeBase):
    __table__              = Table('B', DeclarativeBase.metadata,
                                   Column('id', Integer, primary_key=True, index=True, autoincrement=True),
                                   Column('col_b', Integer, index=True),
                                   schema='database_b')

#engine = create_engine('sqlite:////tmp/database_a.sqlite',echo=True, poolclass=SingletonThreadPool)
engine = create_engine('sqlite:////tmp/database_a.sqlite',echo=True)
db     = engine.connect()
db.execute("ATTACH DATABASE '/tmp/database_b.sqlite' AS database_b")

DeclarativeBase.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
session.commit()

不幸的是，我得到了以下输出：

monster:tmp ladmin$ python sqliteattachtest2.py 
2014-04-12 18:04:58,845 INFO sqlalchemy.engine.base.Engine ATTACH DATABASE '/tmp/database_b.sqlite' AS database_b
2014-04-12 18:04:58,845 INFO sqlalchemy.engine.base.Engine ()
2014-04-12 18:04:58,846 INFO sqlalchemy.engine.base.Engine PRAGMA "database_b".table_info("B")
2014-04-12 18:04:58,846 INFO sqlalchemy.engine.base.Engine ()
2014-04-12 18:04:58,846 INFO sqlalchemy.engine.base.Engine ROLLBACK
Traceback (most recent call last):
  File "sqliteattachtest2.py", line 29, in <module>
    DeclarativeBase.metadata.create_all(engine)
  File "/opt/local/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/sqlalchemy/schema.py", line 2793, in create_all
    tables=tables)
  File "/opt/local/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 1479, in _run_visitor
    conn._run_visitor(visitorcallable, element, **kwargs)
  File "/opt/local/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 1122, in _run_visitor
    **kwargs).traverse_single(element)
  File "/opt/local/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/sqlalchemy/sql/visitors.py", line 111, in traverse_single
    return meth(obj, **kw)
  File "/opt/local/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/sqlalchemy/engine/ddl.py", line 57, in visit_metadata
    if self._can_create_table(t)]
  File "/opt/local/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/sqlalchemy/engine/ddl.py", line 35, in _can_create_table
    table.name, schema=table.schema)
  File "/opt/local/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/sqlalchemy/dialects/sqlite/base.py", line 716, in has_table
    cursor = _pragma_cursor(connection.execute(statement))
  File "/opt/local/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 662, in execute
    params)
  File "/opt/local/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 805, in _execute_text
    statement, parameters
  File "/opt/local/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 874, in _execute_context
    context)
  File "/opt/local/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 1024, in _handle_dbapi_exception
    exc_info
  File "/opt/local/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/sqlalchemy/util/compat.py", line 195, in raise_from_cause
    reraise(type(exception), exception, tb=exc_tb)
  File "/opt/local/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 867, in _execute_context
    context)
  File "/opt/local/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/sqlalchemy/engine/default.py", line 324, in do_execute
    cursor.execute(statement, parameters)
sqlalchemy.exc.OperationalError: (OperationalError) unknown database "database_b" 'PRAGMA "database_b".table_info("B")' ()

我读到这种事情可以使用 Postgres 模式来完成。多个附加的 SQLite 数据库似乎是一个自然的对应物。我怀疑我要么在做一些愚蠢的事情，要么错过了一些重要的点。是否可以使用 SQLAlchemy 同时处理多个 SQLite 文件？如果是这样，最好的方法是什么？还有其他 ORM 比 SQLAlchemy 更容易吗？

谢谢！ 丹

【参考方案1】：

使用内存创建 SQlite 引擎而不是附加不同的数据库文件

from sqlalchemy import create_engine, MetaData, Table,Column,Integer,select
from sqlalchemy.orm import mapper, sessionmaker
from sqlite3 import dbapi2 as sqlite
from sqlalchemy.engine.reflection import Inspector

class Bookmarks(object):
    pass

class BookmarksB(object):
    pass



def loadSession():
    engine = create_engine('sqlite://', echo=True)
    engine.execute("attach database 'database_b' as BB;")
    engine.execute("attach database 'database_a' as AA;")
    metadata = MetaData(engine)


    inspector = Inspector.from_engine(engine)
    print inspector.get_table_names()

    moz_bookmarks = Table('table_a', metadata,Column("id", Integer, primary_key=True),schema='AA', autoload=True)
    mapper(Bookmarks, moz_bookmarks)
    moz_bookmarksB = Table('table_b', metadata,Column("id", Integer, primary_key=True),schema='BB', autoload=True)
    mapper(BookmarksB, moz_bookmarksB)

    Session = sessionmaker(bind=engine)
    session = Session()
    return session

if __name__ == "__main__":
    session = loadSession()
    res = session.query(Bookmarks).all()
    for m in res:
        print m.msisdn,m.id

    #print list(select([moz_bookmarks, moz_bookmarksB], moz_bookmarks.c.b_id == moz_bookmarksB.c.id).execute())

【讨论】：

我测试了loadSession的前6行。不打印表名。