如何实现适配MySQL与SQLite的Flask-SQLAlchemy混合属性表达式拼接

问题场景 #

现有如下Flask-SQLAlchemy模型，其中display_name混合属性通过func.concat实现字符串拼接，在MySQL环境中可正常查询，但在SQLite测试环境下执行单元测试时，抛出sqlite3.OperationalError: no such function: concat错误。

原模型代码 #

from sqlalchemy import extract
from sqlalchemy.sql import func
from app import db

class Entry(db.Model):
   date_order = db.Column(db.Date, nullable=False)
   version_number = db.Column(db.Integer, nullable=False, default=1)

   @hybrid_property
   def display_name(self):
       return f"{self.date_order.year} ({self.version_number})"

   @display_name.expression
   def display_name(cls):
       return func.concat(extract("year", cls.date_order), " (", cls.version_number, ")")

测试代码与错误 #

测试代码：

# `create_entry` fixture commits the instance to db
def test_user_display_name_expression(create_entry):
    entry = create_entry(date_order=date(2022, 11, 11), version_number=3)

    filtered = Entry.query.filter_by(display_name="2022 (3)").one()

    assert filtered == entry

错误信息：

sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such function: concat

解决方案 #

以下三种方案均可实现MySQL与SQLite的兼容：

方案1：使用SQLAlchemy内置concat函数（推荐） #

SQLAlchemy提供了跨数据库兼容的concat函数，替代原生func.concat，它会自动根据数据库类型生成对应拼接语法。

修改后的模型代码：

from sqlalchemy import extract
from sqlalchemy.sql.functions import concat  # 替换原func.concat导入
from app import db

class Entry(db.Model):
   date_order = db.Column(db.Date, nullable=False)
   version_number = db.Column(db.Integer, nullable=False, default=1)

   @hybrid_property
   def display_name(self):
       return f"{self.date_order.year} ({self.version_number})"

   @display_name.expression
   def display_name(cls):
       return concat(extract("year", cls.date_order), " (", cls.version_number, ")")

方案2：使用字符串连接运算符 #

通过类型转换后使用+或||运算符，SQLAlchemy会自动将其转换为对应数据库的连接语法：

from sqlalchemy import extract, cast, String
from app import db

class Entry(db.Model):
   date_order = db.Column(db.Date, nullable=False)
   version_number = db.Column(db.Integer, nullable=False, default=1)

   @hybrid_property
   def display_name(self):
       return f"{self.date_order.year} ({self.version_number})"

   @display_name.expression
   def display_name(cls):
       # 将数值类型转为字符串后拼接
       year_str = cast(extract("year", cls.date_order), String)
       version_str = cast(cls.version_number, String)
       return year_str + " (" + version_str + ")"
       # 也可使用||：return year_str || " (" || version_str || ")"

方案3：给SQLite注册concat函数 #

如果不想修改现有模型代码，可在初始化SQLite连接时注册自定义的concat函数，让SQLite支持该语法：

import sqlite3
from app import app  # 假设你的Flask app实例在这里

def sqlite_concat(*args):
    return ''.join(str(arg) for arg in args)

# 在数据库初始化时注册函数
with app.app_context():
    engine = db.engine
    if engine.driver == 'pysqlite':
        with engine.connect() as conn:
            conn.connection.create_function('concat', -1, sqlite_concat)

验证 #

修改后，在MySQL和SQLite环境下执行测试代码，均可正常过滤出目标Entry实例，不再报错。

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