如何将已创建的SQLite FTS5虚拟表映射为SQLAlchemy对象？

我想用SQLAlchemy访问SQLite3中的FTS5虚拟表，确保每次插入主表时虚拟表自动更新。参考相关方案写了代码后，通过SQLAlchemy的echo功能确认主表、FTS虚拟表和触发器都已正确创建，但不知道怎么把已存在的虚拟表映射成SQLAlchemy对象，也疑惑为什么用CreateFtsTable创建的表还要重新声明？

已编写的代码 #

class CreateFtsTable(DDLElement):
    """Represents a CREATE VIRTUAL TABLE ... USING fts5 statement, for indexing
    a given table.
    """

    def __init__(self, table, version=5):
        self.table = table
        self.version = version


@compiles(CreateFtsTable)
def compile_create_fts_table(element, compiler, **kw):
    tbl = element.table
    version = element.version
    preparer = compiler.preparer

    vtbl_name = preparer.quote(tbl.__table__.name + "_idx")

    columns = [x.name for x in tbl.__mapper__.columns]
    columns.append('tokenize="porter unicode61"')
    columns = ', '.join(columns)

    return f"CREATE VIRTUAL TABLE IF NOT EXISTS {vtbl_name} USING FTS{version} ({columns})"

class WorkItem(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    type = db.Column(db.String, nullable=False)
    state = db.Column(db.String, nullable=False)
    title = db.Column(db.String, nullable=False)
    description = db.Column(db.String, nullable=False)

update_fts = DDL('''CREATE TRIGGER work_item_update AFTER INSERT ON work_item
  BEGIN
    INSERT INTO work_item_idx (id, type, state, title, description) 
    VALUES (new.id, new.type, new.state, new.title, new.description);
  END;''')
db.event.listen(WorkItem.__table__, 'after_create', CreateFtsTable(WorkItem))
db.event.listen(WorkItem.__table__, 'after_create', update_fts)

为什么需要重新映射？ #

你用CreateFtsTable和DDL事件只是在数据库层面创建了虚拟表，但SQLAlchemy的ORM本身不知道这个表的存在——它需要对应的模型类或表对象作为操作入口，才能通过ORM进行查询、修改等交互。数据库里的物理表和SQLAlchemy的映射对象是两个独立层面的东西，前者是存储载体，后者是ORM的操作接口。

两种映射已存在FTS虚拟表的方法 #

方法1：手动声明ORM模型类 #

直接创建对应work_item_idx虚拟表的模型类，结构和FTS表完全匹配：

class WorkItemIdx(db.Model):
    __tablename__ = 'work_item_idx'
    # FTS表默认自带rowid主键，这里映射原表的id用于关联
    id = db.Column(db.Integer)
    type = db.Column(db.String)
    state = db.Column(db.String)
    title = db.Column(db.String)
    description = db.Column(db.String)

    # 标记为SQLite虚拟表，启用自动增长特性
    __table_args__ = {'sqlite_autoincrement': True}

之后就能直接用这个模型类做FTS查询：

# 搜索标题或描述包含"bug"的记录
results = db.session.query(WorkItemIdx).filter(WorkItemIdx.match('bug')).all()

方法2：反射（Reflect）已存在的表 #

如果不想手动维护模型类，可以用SQLAlchemy的反射功能自动读取数据库表结构，生成Table对象：

from sqlalchemy import MetaData, Table

metadata = MetaData()
# 自动从数据库加载work_item_idx表的结构
work_item_idx = Table('work_item_idx', metadata, autoload_with=db.engine)

# 使用反射得到的表执行FTS查询
results = db.session.query(work_item_idx).filter(work_item_idx.c.match('bug')).all()

这种方法适合表结构复杂或不想重复编写模型的场景，但反射得到的是Core层面的Table对象，而非ORM模型类，操作方式更偏向于原生SQL风格。

额外建议：完善触发器同步逻辑 #

目前你的触发器只处理了INSERT操作，为了保证FTS表和主表数据完全一致，建议补充UPDATE和DELETE的触发器：

CREATE TRIGGER work_item_update_trigger AFTER UPDATE ON work_item
BEGIN
  DELETE FROM work_item_idx WHERE id = old.id;
  INSERT INTO work_item_idx (id, type, state, title, description) 
  VALUES (new.id, new.type, new.state, new.title, new.description);
END;

CREATE TRIGGER work_item_delete_trigger AFTER DELETE ON work_item
BEGIN
  DELETE FROM work_item_idx WHERE id = old.id;
END;

把这两个触发器也通过db.event.listen绑定到WorkItem.__table__的after_create事件即可。

内容的提问来源于stack exchange，提问作者chronos