防止 MySQL 在 Flask-SQLAlchemy 中截断值答案

【问题标题】：Prevent MySQL to truncate values in Flask-SQLAlchemy防止 MySQL 在 Flask-SQLAlchemy 中截断值
【发布时间】：2025-11-22 17:45:02
【问题描述】：

我有一个使用 Flask-SQLAlchemy 连接到数据库的 Flask 应用程序。当我想向一个太长或无效（ish）的表中插入一个值时，我通常会收到一个警告，指出该值已被截断，并且 MySQL 会愉快地插入该值的有效部分。这会导致以后出现各种问题。

在我的连接过程中检查sql_mode 变量会产生以下模式：

STRICT_TRANS_TABLES
ERROR_FOR_DIVISION_BY_ZERO
NO_AUTO_CREATE_USER
NO_ENGINE_SUBSTITUTION

为了实现我的目标，我需要将STRICT_ALL_TABLES 添加到这个组合中，我可以通过这个（漂亮的转储）sn-p 轻松做到这一点（db 是我应用中的SQLAlchemy 实例）：

res = list(db.engine.execute('SHOW VARIABLES LIKE \'sql_mode\''))
modes = res[0][1].split(',')
db.engine.execute(f'SET sql_mode = \'{",".join(modes)}\'')

# I tried and failed using the following for some reason
db.engine.execute('SET sql_mode = ?', '.'.join(modes)

在此之后，当我使用不合适的值更新模型时，我得到了我想要的；异常而不是警告：

sqlalchemy.exc.DataError: (pymysql.err.DataError) (1265, "Data truncated for column 'value' at row 1")

问题是，当我每次连接到 MySQL 实例时，如何告诉 SQLAlchemy 执行此操作？

【问题讨论】：

根据docs，省略修饰符意味着SESSION。
错误消息是指value 列，而您的 SQL 文本不包含此类子字符串。您确定错误消息是由显示的 SQL 语句执行产生的吗？
docs.sqlalchemy.org/en/13/core/…

标签： python mysql sqlalchemy flask-sqlalchemy

【解决方案1】：

在 cmets 中 @gord-thompson 的文档链接的帮助下，我设计了以下解决方案：

from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import event
from sqlalchemy.dialects.mysql.base import MySQLDialect


def mysql_on_connect(connection, conn_record):  # pylint: disable=unused-variable
    """Turn on strict mode for all tables
    """

    cursor = connection.cursor()

    # Get the current SQL Mode
    cursor.execute('SHOW VARIABLES LIKE \'sql_mode\'')
    res = list(cursor)
    modes = set(res[0][1].split(','))

    # Add STRICT_ALL_TABLES to the list of modes
    modes.add('STRICT_ALL_TABLES')
    new_modes = ','.join(modes)
    cursor.execute('SET sql_mode = %s', new_modes)

    cursor.close()


class StrictSQLAlchemy(SQLAlchemy):
    def init_app(self, app, *args, **kwargs):
        super().init_app(app, *args, **kwargs)

        with app.app_context():
            if isinstance(self.engine.dialect, MySQLDialect):
                event.listens_for(self.engine, 'connect')(mysql_on_connect)


db = StrictSQLAlchemy()


class User(db.Model):
    id = db.Column(db.Integer(), primary_key=True)
    username = db.Column(db.String(length=20), unique=True)


app = Flask()
db.init_app(app)

user = User(username='a_very_long_username_that_doesnt_fit_in_20_characters')
db.session.add(user)
db.session.commit()

现在，我得到一个适当的异常，而不是警告和截断值：

sqlalchemy.exc.DataError: (pymysql.err.DataError) (1406, "Data too long for column 'username' at row 1")

【讨论】：