Python3 ランタイムエラー TypeError: bytes 型のオブジェクトは JSON シリアライズ可能ではありません。

dict 型のデータ (中国語が存在する場合) は python2 では変換可能ですが、python3 ではシリアライズの問題が発生します。
TypeError: bytes 型のオブジェクトは JSON シリアライズ可能ではありません。

下位の解決策は
print(result['タイトル'])
print(result['title'].decode('utf-8'))を実行します。
print(result.decode('utf-8'))を実行します。
print(chardet.detect(result['title']))を実行します。

ソースコードを投稿してください。

#! /usr/bin/python3
# -*- coding:utf-8 -*-
import pymysql, os, configparser
from pymysql.cursors import DictCursor
from DBUtils.PooledDB import PooledDB
import chardet
import json





class Config(object):
    """
    # Config().get_content("user_information")
    Parameters inside the configuration file
    [dbMysql]
    host = 192.168.1.101
    port = 3306
    user = root
    password = python123
    """

    def __init__(self, config_filename="dbMysqlConfig.cnf"):
        file_path = os.path.join(os.path.dirname(__file__), config_filename)
        self.cf = configparser.ConfigParser()
        self.cf.read(file_path)

    def get_sections(self):
        return self.cf.sections()

    def get_options(self, section):
        return self.cf.options(section)

    def get_content(self, section):
        result = {}
        for option in self.get_options(section):
            value = self.cf.get(section, option)
            result[option] = int(value) if value.isdigit() else value
        return result


class BasePymysqlPool(object):
    def __init__(self, host, port, user, password, db_name):
        self.db_host = host
        self.db_port = int(port)
        self.user = user
        self.password = str(password)
        self.db = db_name
        self.conn = None
        self.cursor = None


class MyPymysqlPool(BasePymysqlPool):
    """
    MYSQL database object, responsible for generating database connections , this class uses a connection pool to implement the connection
        Get connection object: conn = Mysql.getConn()
        Release the connection object; conn.close() or del conn
    """
    # Connection pool object
    __pool = None

    def __init__(self, conf_name=None):
        self.conf = Config().get_content(conf_name)
        super(MyPymysqlPool, self). __init__(**self.conf)
        # Database constructor that takes the connection from the connection pool and generates the operation cursor
        self._conn = self.__getConn()
        self._cursor = self._conn.cursor()

    def __getConn(self):
        """
        @summary: Static method to retrieve a connection from the connection pool
        @return MySQLdb.connection
        """
        if MyPymysqlPool.__pool is None:
            __pool = PooledDB(creator=pymysql,
                              mincached=1,
                              maxcached=20,
                              host=self.db_host,
                              port=self.db_port,
                              user=self.user,
                              passwd=self.password,
                              db=self.db,
                              use_unicode=False,
                              charset="utf8mb4",
                              cursorclass=DictCursor)
        return __pool.connection()

    def getAll(self, sql, param=None):
        """
        @summary: Execute the query and retrieve all the result sets
        @param sql: query SQL, specify only a list of conditions if there are any, and pass in the values of the conditions using parameter [param]
        @param param: optional parameter, condition list value (tuple/list)
        @return: result list(dictionary object)/boolean Query the result set
        """
        if param is None:
            count = self._cursor.execute(sql)
        else:
            count = self._cursor.execute(sql, param)
        if count > 0:
            result = self._cursor.fetchall()
        else:
            result = False
        return result

    def getOne(self, sql, param=None):
        """
        @summary: Execute the query and take out the first one
        @param sql: query SQL, if there are query conditions, please specify only the list of conditions and pass in the values of the conditions using parameter [param]
        @param param: optional parameter, condition list value (tuple/list)
        @return: result list/boolean the result set of the query
        """
        if param is None:
            count = self._cursor.execute(sql)
        else:
            count = self._cursor.execute(sql, param)
        if count > 0:
            result = self._cursor.fetchone()
        else:
            result = False
        return result

    def getMany(self, sql, num, param=None):
        """
        @summary: Execute the query and retrieve num results
        @param sql: query SQL, if there are query conditions, please specify only the list of conditions and pass in the values of the conditions using parameter [param]
        @param num: the number of results to get
        @param param: optional parameter, condition list value (tuple/list)
        @return: result list/boolean The result set of the query
        """
        if param is None:
            count = self._cursor.execute(sql)
        else:
            count = self._cursor.execute(sql, param)
        if count > 0:
            result = self._cursor.fetchmany(num)
        else:
            result = False
        return result

    def insertMany(self, sql, values):
        """
        @summary: Insert multiple rows into a data table
        @param sql: the SQL format to be inserted
        @param values:The data to be inserted tuples(tuples)/list[list]
        @return: count the number of rows affected
        """
        count = self._cursor.executemany(sql, values)
        return count

    def __query(self, sql, param=None):
        if param is None:
            count = self._cursor.execute(sql)
        else:
            count = self._cursor.execute(sql, param)
        return count

    def update(self, sql, param=None):
        """
        @summary: Update a data table record
        @param sql: SQL format and conditions, use (%s,%s)
        @param param: the value to update tuple/list
        @return: count the number of rows affected
        """
        return self.__query(sql, param)

    def insert(self, sql, param=None):
        """
        @summary: Update data table records
        @param sql: SQL format and conditions, use (%s,%s)
        @param param: the value to update tuple/list
        @return: count the number of rows affected
        """
        return self.__query(sql, param)

    def delete(self, sql, param=None):
        """
        @summary: Delete a data table record
        @param sql: SQL format and conditions, use (%s,%s)
        @param param: condition to be deleted value tuple/list
        @return: count the number of rows affected
        """
        return self.__query(sql, param)

    def begin(self):
        """
        @summary: open transaction
        """
        self._conn.autocommit(0)

    def end(self, option='commit'):
        """
        @summary: Ending the transaction
        """
        if option == 'commit':
            self._conn.commit()
        else:
            self._conn.rollback()

    def dispose(self, isEnd=1):
        """
        @summary: Release connection pool resources
        """
        if isEnd == 1:
            self.end('commit')
        else:
            self.end('rollback')
        self._cursor.close()
        self._conn.close()



if __name__ == '__main__':
    mysql = MyPymysqlPool("dbMysql")
    sqlAll = "select * from novel limit 1"
    result = mysql.getOne(sqlA

最初に: モジュールのインストール

pip3 install numpy

クラスを追加する。

class MyEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, np.ndarray):
            return obj.tolist()
        elif isinstance(obj, bytes):
            return str(obj, encoding='utf-8');
        return json.JSONEncoder.default(self, obj)

最終的なソースコードです。

#! /usr/bin/python3
# -*- coding:utf-8 -*-
import pymysql, os, configparser
from pymysql.cursors import DictCursor
from DBUtils.PooledDB import PooledDB
import chardet
import json
import numpy as np


class MyEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, np.ndarray):
            return obj.tolist()
        elif isinstance(obj, bytes):
            return str(obj, encoding='utf-8');
        return json.JSONEncoder.default(self, obj)


class Config(object):
    """
    # Config().get_content("user_information")
    Parameters inside the configuration file
    [dbMysql]
    host = 192.168.1.101
    port = 3306
    user = root
    password = python123
    """

    def __init__(self, config_filename="dbMysqlConfig.cnf"):
        file_path = os.path.join(os.path.dirname(__file__), config_filename)
        self.cf = configparser.ConfigParser()
        self.cf.read(file_path)

    def get_sections(self):
        return self.cf.sections()

    def get_options(self, section):
        return self.cf.options(section)

    def get_content(self, section):
        result = {}
        for option in self.get_options(section):
            value = self.cf.get(section, option)
            result[option] = int(value) if value.isdigit() else value
        return result


class BasePymysqlPool(object):
    def __init__(self, host, port, user, password, db_name):
        self.db_host = host
        self.db_port = int(port)
        self.user = user
        self.password = str(password)
        self.db = db_name
        self.conn = None
        self.cursor = None


class MyPymysqlPool(BasePymysqlPool):
    """
    MYSQL database object, responsible for generating database connections , this class uses a connection pool to implement the connection
        Get connection object: conn = Mysql.getConn()
        Release the connection object; conn.close() or del conn
    """
    # Connection pool object
    __pool = None

    def __init__(self, conf_name=None):
        self.conf = Config().get_content(conf_name)
        super(MyPymysqlPool, self). __init__(**self.conf)
        # Database constructor that takes the connection from the connection pool and generates the operation cursor
        self._conn = self.__getConn()
        self._cursor = self._conn.cursor()

    def __getConn(self):
        """
        @summary: Static method to retrieve a connection from the connection pool
        @return MySQLdb.connection
        """
        if MyPymysqlPool.__pool is None:
            __pool = PooledDB(creator=pymysql,
                              mincached=1,
                              maxcached=20,
                              host=self.db_host,
                              port=self.db_port,
                              user=self.user,
                              passwd=self.password,
                              db=self.db,
                              use_unicode=False,
                              charset="utf8mb4",
                              cursorclass=DictCursor)
        return __pool.connection()

    def getAll(self, sql, param=None):
        """
        @summary: Execute the query and retrieve all the result sets
        @param sql: query SQL, specify only a list of conditions if there are any, and pass in the values of the conditions using parameter [param]
        @param param: optional parameter, condition list value (tuple/list)
        @return: result list(dictionary object)/boolean the result set of the query
        """
        if param is None:
            count = self._cursor.execute(sql)
        else:
            count = self._cursor.execute(sql, param)
        if count > 0:
            result = self._cursor.fetchall()
        else:
            result = False
        return result

    def getOne(self, sql, param=None):
        """
        @summary: Execute the query and take out the first one
        @param sql: query SQL, if there are query conditions, please specify only the list of conditions and pass in the values of the conditions using parameter [param]
        @param param: optional parameter, condition list value (tuple/list)
        @return: result list/boolean the result set of the query
        """
        if param is None:
            count = self._cursor.execute(sql)
        else:
            count = self._cursor.execute(sql, param)
        if count > 0:
            result = self._cursor.fetchone()
        else:
            result = False
        return result

    def getMany(self, sql, num, param=None):
        """
        @summary: Execute the query and retrieve num results
        @param sql: query SQL, if there are query conditions, please specify only the list of conditions and pass in the values of the conditions using parameter [param]
        @param num: the number of results to get
        @param param: optional parameter, condition list value (tuple/list)
        @return: result list/boolean The result set of the query
        """
        if param is None:
            count = self._cursor.execute(sql)
        else:
            count = self._cursor.execute(sql, param)
        if count > 0:
            result = self._cursor.fetchmany(num)
        else:
            result = False
        return result

    def insertMany(self, sql, values):
        """
        @summary: Insert multiple rows into a data table
        @param sql: the SQL format to be inserted
        @param values:The data to be inserted tuples(tuples)/list[list]
        @return: count the number of rows affected
        """
        count = self._cursor.executemany(sql, values)
        return count

    def __query(self, sql, param=None):
        if param is None:
            count = self._cursor.execute(sql)
        else:
            count = self._cursor.execute(sql, param)
        return count

    def update(self, sql, param=None):
        """
        @summary: Update a data table record
        @param sql: SQL format and conditions, use (%s,%s)
        @param param: the value to update tuple/list
        @return: count the number of rows affected
        """
        return self.__query(sql, param)

    def insert(self, sql, param=None):
        """
        @summary: Update data table records
        @param sql: SQL format and conditions, use (%s,%s)
        @param param: the value to update tuple/list
        @return: count the number of rows affected
        """
        return self.__query(sql, param)

    def delete(self, sql, param=None):
        """
        @summary: Delete a data table record
        @param sql: SQL format and conditions, use (%s,%s)
        @param param: condition to be deleted value tuple/list
        @return: count the number of rows affected
        """
        return self.__query(sql, param)

    def begin(self):
        """
        @summary: open transaction
        """
        self._conn.autocommit(0)

    def end(self, option='commit'):
        """
        @summary: Ending the transaction
        """
        if option == 'commit':
            self._conn.commit()
        else:
            self._conn.rollback()

    def dispose(self, isEnd=1):
        """
        @summary: Release connection pool resources
        """
        if isEnd == 1:
            self.end('commit')
        else:
            self.end('rollback')
        self._cursor.close()
        self._conn.close()



if __name__ == '__main__':
    mysql = MyPymysqlPool("dbMysql")
    sqlAll = "select * from novel limit 1"
    result = mysql.getOne(sqlAll)
    print(result)
    # print(result['title'])