SQLAlchemy查询中多UUID列类型转换问题求助

问题描述 #

从UUID类型的列中获取值时，调用CursorObject的fetchall()方法会触发Python错误：TypeError("a bytes-like object is required, not 'str'")。

报错代码（单列查询场景） #

import sqlalchemy as sa

def getScheduleData(engine: sa.engine, scheduleID: str) -> dict:
    rows = []
    metadata = sa.MetaData()
    table = sa.Table('schedules', metadata, autoload_with=engine)

    stmt = sa.select(table.c['userID']).where(table.c["ID"] == scheduleID)
    print(stmt)

    try:
        with engine.connect() as conn:
            result = conn.execute(stmt)
            rows = result.fetchall()
            print(rows)
    except Exception as e:
        print(e)

更复杂场景：查询整行数据（多UUID列） #

当schedules表包含多个UUID类型列，查询整行时同样触发上述错误：

import sqlalchemy as sa

def getScheduleData(engine: sa.engine, scheduleID: str) -> dict:
    rows = []
    metadata = sa.MetaData()
    table = sa.Table('schedules', metadata, autoload_with=engine)

    # 查询所有列
    stmt = sa.select(table).where(table.c["ID"] == scheduleID)
    print(stmt)

    try:
        with engine.connect() as conn:
            result = conn.execute(stmt)
            rows = result.fetchall()
            print(rows)
    except Exception as e:
        print(e)

已有的单列解决方案 #

针对单个UUID列，可以通过sa.cast()将其转为字符串避免报错，但仅适用于指定列：

import sqlalchemy as sa

def getScheduleData(engine: sa.engine, scheduleID: str) -> dict:
    rows = []
    metadata = sa.MetaData()
    table = sa.Table('schedules', metadata, autoload_with=engine)

    # 将列值转为字符串
    stmt = sa.select(sa.cast(table.c['userID'], sa.String(36))).where(table.c["ID"] == scheduleID)
    print(stmt)

    try:
        with engine.connect() as conn:
            result = conn.execute(stmt)
            rows = result.fetchall()
            print(rows)
    except Exception as e:
        print(e)

当前核心问题：如何在查询整行数据且存在多个UUID类型列的场景下，批量实现类似的类型转换？

环境补充信息 #

• 驱动：mysqlconnector
• 数据库：MariaDB
• 引擎创建代码：

self._engine_str = f"mysql+mysqlconnector://{self._user}:{self._password}@{self._host}/"
engine = sa.create_engine(f"{self._engine_str}{db_name}")

错误栈 #

Traceback (most recent call last):
  File "C:\projects\myWebApp\frontend\fastapi\controllers\schedule.py", line 598, in getScheduleData
    rows = result.fetchall()
           ^^^^^^^^^^^^^^^^^
  File "c:\Users\Luke\.conda\envs\mywebapp-server\Lib\site-packages\sqlalchemy\engine\result.py", line 1317, in fetchall
    return self._allrows()
           ^^^^^^^^^^^^^^^
  File "c:\Users\Luke\.conda\envs\mywebapp-server\Lib\site-packages\sqlalchemy\engine\result.py", line 551, in _allrows
    made_rows = [make_row(row) for row in rows]
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "c:\Users\Luke\.conda\envs\mywebapp-server\Lib\site-packages\sqlalchemy\engine\result.py", line 551, in <listcomp>
    made_rows = [make_row(row) for row in rows]
                 ^^^^^^^^^^^^^
  File "lib\\sqlalchemy\\cyextension\\resultproxy.pyx", line 22, in sqlalchemy.cyextension.resultproxy.BaseRow.__init__
  File "lib\\sqlalchemy\\cyextension\\resultproxy.pyx", line 79, in sqlalchemy.cyextension.resultproxy._apply_processors
  File "c:\Users\Luke\.conda\envs\mywebapp-server\Lib\site-packages\sqlalchemy\sql\sqltypes.py", line 3615, in process
    value = _python_UUID(value)
            ^^^^^^^^^^^^^^^^^^^
  File "c:\Users\Luke\.conda\envs\mywebapp-server\Lib\uuid.py", line 175, in __init__
    hex = hex.replace('urn:', '').replace('uuid:', '')
          ^^^^^^^^^^^^^^^^^^^^^^^
TypeError: a bytes-like object is required, not 'str'

解决方案 #

方法1：批量转换所有UUID列为字符串（整行查询场景） #

遍历表的所有列，对UUID类型的列自动应用sa.cast()转换，其他列保持原样：

import sqlalchemy as sa
from sqlalchemy.sql.sqltypes import UUID

def getScheduleData(engine: sa.engine, scheduleID: str) -> dict:
    rows = []
    metadata = sa.MetaData()
    table = sa.Table('schedules', metadata, autoload_with=engine)

    # 批量处理列：UUID列转字符串，其他列直接选择
    selected_columns = [
        sa.cast(col, sa.String(36)) if isinstance(col.type, UUID) else col
        for col in table.c
    ]

    stmt = sa.select(*selected_columns).where(table.c["ID"] == scheduleID)
    print(stmt)

    try:
        with engine.connect() as conn:
            result = conn.execute(stmt)
            rows = result.fetchall()
            print(rows)
    except Exception as e:
        print(e)

方法2：自定义UUID类型处理器（全局生效） #

通过自定义SQLAlchemy的UUID类型处理器，统一处理从数据库返回的UUID值，避免类型不匹配：

import sqlalchemy as sa
from sqlalchemy import types
import uuid

class StringUUID(types.TypeDecorator):
    impl = types.String
    cache_ok = True

    def process_bind_param(self, value, dialect):
        if value is None:
            return value
        return str(value)

    def process_result_value(self, value, dialect):
        if value is None:
            return value
        # 处理驱动可能返回的bytes类型
        if isinstance(value, bytes):
            value = value.decode('utf-8')
        return uuid.UUID(value)

# 创建引擎时无需额外参数，加载表时指定UUID列使用自定义类型
engine = sa.create_engine(
    f"mysql+mysqlconnector://{self._user}:{self._password}@{self._host}/{db_name}"
)

metadata = sa.MetaData()
table = sa.Table(
    'schedules',
    metadata,
    # 覆盖UUID列的类型定义
    sa.Column('ID', StringUUID),
    sa.Column('userID', StringUUID),
    # 其他列保持自动加载
    autoload_with=engine,
    autoload_replace=False
)

# 后续查询无需额外转换
def getScheduleData(engine: sa.engine, scheduleID: str) -> dict:
    rows = []
    stmt = sa.select(table).where(table.c["ID"] == scheduleID)
    try:
        with engine.connect() as conn:
            result = conn.execute(stmt)
            rows = result.fetchall()
            print(rows)
    except Exception as e:
        print(e)

方法3：调整连接参数（根源解决） #

mysqlconnector驱动对UUID的处理存在兼容性问题，可尝试在创建引擎时添加use_unicode=True参数，确保驱动返回字符串而非bytes类型：

engine = sa.create_engine(
    f"mysql+mysqlconnector://{self._user}:{self._password}@{self._host}/{db_name}",
    connect_args={"use_unicode": True}
)

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