博客

总有那么几个古老的项目与众不同，有些政府医院的项目使用oracle的字符集是US7ASCII 这个奇葩的字符集。当python 查询这样的数据库时会出现中文字符乱码，不能正确显示内容。在网上搜索了很多解决方案都是在sql里用rawtohex的utl_raw.cast_to_raw进行字符转换，然后对查询结果使用decode(‘GBK’)。

select rawtohex(utl_raw.cast_to_raw(name)) as name from index

例如上面这个例子，其实是将字段转成了byte.查询出结果后在decode。网上都是这样的解决方法。但是这样就太不灵活了。并且要所有中文字段都加这个函数进行转化。太麻烦不好维护，所以不建议使用。

不但读取，python插入中文到US7ASCII 这种字符集的oracle数据库时也是乱码。为了解决这些问题，查阅了官方文档，总结了下面这个解决方案

该方法是采用的python连接oracle官方库Cx_oracle 进行验证，实际上新出的python-oracledb 的 oracledb 也是可以的使用的。并且也提供了使用sqlalchemy 查询或插入Oracle的字符集为US7ASCII 时中文乱码问题解决方案 废话不多说了直接贴代码，


请注意看注释

1 使用Cx_oracle 或者python-oracledb查询或写入US7ASCII 字符集的oracle数据库

Cx_oracle的cursor 提供了两个重要的可以自定义的方法  outputtypehandler和 inputtypehandler这两个方法可以调用自定义的方法对特定的dbtype 类型进行编码转换等操作，这样我们在读取和写入前进行特定的转换就不是乱码了

#!/usr/bin/env python
# -*- coding:gbk -*-
import cx_Oracle
import os


def out_converter(value):
    if value is None:
        return None

    return value.decode("gbk")


def output_type_handler(
        cursor,
        name,
        default_type,
        size,
        precision,
        scale):
    # 判断字段类型是不是字符串
    if default_type == cx_Oracle.DB_TYPE_VARCHAR:
        # 字段类型如果是字符串 bypass_decode=True 是告诉cursor 将byte 返回。然后调用outconverter方法将byte数据 decode gbk
        return cursor.var(str, arraysize=cursor.arraysize,
                          bypass_decode=True, outconverter=out_converter)


def in_converter(value):
    # if isinstance(value, str):
    #     return value.encode("gbk")
    # return value
    return value.encode("gbk")


def input_type_handler(cursor, value, num_elements):
    if isinstance(value, str):
        return cursor.var(cx_Oracle.DB_TYPE_VARCHAR, arraysize=num_elements, inconverter=in_converter)


if __name__ == '__main__':
    # oracle 配置信息
    user = 'system'
    password = 'xxxx'
    dsn = cx_Oracle.makedsn("localhost", '1521', sid="US7TEST")
    # 这个设置变量可以不用，如果错误信息乱码，出现问题可以设置下
    os.environ["NLS_LANG"] = "AMERICAN_AMERICA.US7ASCII"
    # 建立 oracle 的 connection 重要一点是 encoding='US-ASCII'
    connection = cx_Oracle.connect(user=user, password=password, dsn=dsn, encoding="US-ASCII")
    cursor = connection.cursor()
    # 对cursor增加个自定义的output_type_handler 回调方法用来处理读取时的特殊类型
    cursor.outputtypehandler = output_type_handler
    # 对cursor增加个自定义的output_type_handler 回调方法用来处理写入时的特殊类型
    cursor.inputtypehandler = input_type_handler

    # 测试执行带中文executemany
    cursor.executemany(" INSERT INTO TEST_US7ASCII(ID,TEST_CHINESE,TEST_CC)VALUES(:ID,:TEST_CHINESE,:TEST_CC)  ",
                       [{"ID": 1, "TEST_CHINESE": "1这是第1个中文", "TEST_CC": "1这是第2个中文"},
                        {"ID": 2, "TEST_CHINESE": "2这是第1个中文", "TEST_CC": "2这是第2个中文"}])
    # 测试执行带中文execute插入数据
    cursor.execute(" INSERT INTO TEST_US7ASCII(ID,TEST_CHINESE,TEST_CC)VALUES(:ID,:TEST_CHINESE,:TEST_CC)  ",
                   {"ID": 3, "TEST_CHINESE": "3这是第1个中文", "TEST_CC": "3这是第2个中文"})
    # 测试查询带中文,注意查询语句中不能直接用中文，要想有中文请绑定变量
    for row in cursor.execute("select TEST_CHINESE,:1 from TEST_US7ASCII where TEST_CC =:2 ", ["绑定中文", "3这是第2个中文"]):
        print(row[0], row[1])
    connection.commit()
    cursor.close()
    connection.close()

2 使用sqlalchemy 查询或写入US7ASCII 字符集的oracle数据库，解决中文乱码问题

基于上面的outputtypehandler和 inputypehandler 两个方法，我们也可以在sqlalchemy 进行设置。就是给引擎建立个监听，监听类型为 before_cursor_execute ，表示真正执行sql 前 对原始的cursor 进行自定义等操作，这样我们可以将outputtypehandler和 inputtypehandler两个方法 赋值给cursor

import cx_Oracle
import os
from sqlalchemy import create_engine
from sqlalchemy import text
from sqlalchemy import event


def out_converter(value):
    if value is None:
        return None

    return value.decode("gbk")


def output_type_handler(
        cursor,
        name,
        default_type,
        size,
        precision,
        scale):
    # 判断字段类型是不是字符串
    if default_type == cx_Oracle.DB_TYPE_VARCHAR:
        # 字段类型如果是字符串 bypass_decode=True 是告诉cursor 将byte 返回。然后调用outconverter方法将byte数据 decode gbk
        return cursor.var(str, arraysize=cursor.arraysize,
                          bypass_decode=True, outconverter=out_converter)


def in_converter(value):
    # if isinstance(value, str):
    #     return value.encode("gbk")
    # return value
    return value.encode("gbk")


def input_type_handler(cursor, value, num_elements):
    if isinstance(value, str):
        return cursor.var(cx_Oracle.DB_TYPE_VARCHAR, arraysize=num_elements, inconverter=in_converter)


def receive_before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    # 对cursor增加个自定义的output_type_handler 回调方法用来处理读取时的特殊类型
    cursor.outputtypehandler = output_type_handler
    # 对cursor增加个自定义的output_type_handler 回调方法用来处理写入时的特殊类型
    cursor.inputtypehandler = input_type_handler


if __name__ == '__main__':
    
    user = 'system'
    password = 'xxxx'
    dsn = cx_Oracle.makedsn("localhost", '1521', sid="US7TEST")
    # 这个设置变量可以不用，如果出现错误消息乱码问题可以设置下
    os.environ["NLS_LANG"] = "AMERICAN_AMERICA.US7ASCII"

    sqlalchemy_database_uri = f'oracle+cx_oracle://{user}:{password}@{dsn}'
    # connect_args参数很重要必须为US-ASCII
    engine = create_engine(sqlalchemy_database_uri, connect_args={"encoding": "US-ASCII"})
    # 对engine 建立一个监听，在每次cursor前进行拦截然后自定义某些内容 该行代码是关键
    event.listen(engine, 'before_cursor_execute', receive_before_cursor_execute)
    # 以下对engine测试,理论上支持orm,session,connect,欢迎测试反馈
    # 测试执行带中文execute 插入中文到数据库
    engine.execute(text(" INSERT INTO TEST_US7ASCII(ID,TEST_CHINESE,TEST_CC)VALUES(:ID,:TEST_CHINESE,:TEST_CC)  "),
                   [{"ID": 1, "TEST_CHINESE": "1这是第1个中文", "TEST_CC": "1这是第2个中文"},
                    {"ID": 2, "TEST_CHINESE": "2这是第1个中文", "TEST_CC": "2这是第2个中文"}])

    # 测试查询带中文,注意查询语句中不能直接用中文，要想有中文请绑定变量
    for row in engine.execute("select TEST_CHINESE,:1 from TEST_US7ASCII where TEST_CC =:2 ", ["绑定中文", "2这是第2个中文"]):
        print(row[0], row[1])

以上就是使用 python 处理oralce 的US7ASCII 字符集的方法，如果对您有帮助欢迎点赞收藏，如有其他问题，欢迎留言讨论

PS ：20230712 更新，自该文章发布以来好多人发来感谢的私信，荣幸之至。值得一提的是该文章不是用来解决所有乱码问题的，切勿一股脑套用。还有值得一提的地方，就是使用该方法时，

用来查询和插入的sql里如果需要写中文，一定要用绑定变量的方式进行查询，插入。在python 里直接用%s 占位最后拼成查询语句的方式是不行的，必须要绑定变量的方式进行sql操作。

列如 要执行

select TEST_CHINESE,'绑定中文' from TEST_US7ASCII where TEST_CC ='查找中文'

要将语句拆开绑定变量进行执行，改成如下可正常执行，具体怎样绑定变量请自行查阅对应驱动的官方文档，这里只列举了一种类型

cursor.execute("select TEST_CHINESE,:1 from TEST_US7ASCII where TEST_CC =:2 ", ["绑定中文", "查找中文"])