python - 使用 sqlalchemy 和 mysql 插入 unicode 消息

Question

我正在尝试将包含笑脸表情符号的字符串插入 MySQL 数据库。我有以下抛出异常的测试。我该如何克服这个错误？

这是测试:

def test_write_unicode(self):
    db_schema = "testing"
    db_url = sqlalchemy.engine.url.URL(drivername='mysql', host=selah.db_host,
        database=db_schema,
        query={ 'read_default_file' : selah.db_config })
    db_eng = sqlalchemy.create_engine(name_or_url=db_url)
    table_name = 'test_write_unicode'
    db_eng.execute("DROP TABLE IF EXISTS {0}".format(table_name))
    db_eng.execute("CREATE TABLE {0} (message text) CHARACTER SET utf8mb4".format(table_name))

    wierd_message = u"It's all about jimmy from stater bros. 😍😍😍 "
    db_eng.execute("INSERT INTO {} (message) VALUES (%s)".format(table_name), wierd_message)

这是我的 db_config 文件:

[mysqld]
default-storage-engine=MyISAM
[client]
user=selah
password=xxxx

这是错误:

Error
Traceback (most recent call last):
  File "/home/selah/Code/selahwork/test_sqlalchemy.py", line 39, in test_write_unicode
    db_eng.execute("INSERT INTO {} (message) VALUES (%s)".format(table_name), wierd_message)
  File "/usr/lib/python2.7/dist-packages/sqlalchemy/engine/base.py", line 1614, in execute
    return connection.execute(statement, *multiparams, **params)
  File "/usr/lib/python2.7/dist-packages/sqlalchemy/engine/base.py", line 662, in execute
    params)
  File "/usr/lib/python2.7/dist-packages/sqlalchemy/engine/base.py", line 805, in _execute_text
    statement, parameters
  File "/usr/lib/python2.7/dist-packages/sqlalchemy/engine/base.py", line 874, in _execute_context
    context)
  File "/usr/lib/python2.7/dist-packages/sqlalchemy/engine/base.py", line 1027, in _handle_dbapi_exception
    util.reraise(*exc_info)
  File "/usr/lib/python2.7/dist-packages/sqlalchemy/engine/base.py", line 867, in _execute_context
    context)
  File "/usr/lib/python2.7/dist-packages/sqlalchemy/engine/default.py", line 324, in do_execute
    cursor.execute(statement, parameters)
  File "/usr/local/lib/python2.7/dist-packages/MySQLdb/cursors.py", line 187, in execute
    query = query % tuple([db.literal(item) for item in args])
  File "/usr/local/lib/python2.7/dist-packages/MySQLdb/connections.py", line 278, in literal
    return self.escape(o, self.encoders)
  File "/usr/local/lib/python2.7/dist-packages/MySQLdb/connections.py", line 208, in unicode_literal
    return db.literal(u.encode(unicode_literal.charset))
UnicodeEncodeError: 'latin-1' codec can't encode characters in position 39-41: ordinal not in range(256)

编辑: 这是我成功的测试:)

def test_write_unicode(self):
    db_schema = "testing"
    db_url = sqlalchemy.engine.url.URL(drivername='mysql', host=selah.db_host,
        database=db_schema,
        query={ 'read_default_file' : selah.db_config, 'charset': 'utf8mb4' })
    db_eng = sqlalchemy.create_engine(name_or_url=db_url)
    table_name = 'test_write_unicode'
    db_eng.execute("DROP TABLE IF EXISTS {0}".format(table_name))
    db_eng.execute("CREATE TABLE {0} (message text) CHARACTER SET utf8mb4".format(table_name))

    wierd_message = u"It's all about jimmy from stater bros. 😍😍😍 "
    db_eng.execute("INSERT INTO {} (message) VALUES (%s)".format(table_name), wierd_message)
    res = db_eng.execute("SELECT * FROM {}".format(table_name))
    self.assertEqual(res.first()[0], wierd_message)

Answer 1

除了设置存储排序规则之外，您还需要告诉 MySQL DBAPI 使用 UTF-8 作为连接字符集。否则，MySQL 通常默认为 latin1_swedish_ci (ISO-8859-1)。

连接字符集通常使用 connection string 中的 charset=utf8mb4 设置但我想 URL(query={'charset': 'utf8mb4'}) 也可能有效。