使用peewee与peewee_migrate时出现TypeError类型错误

问题场景 #

使用Peewee Migrate工具修改MySQL数据库中testprocess表的f4字段，设置其为VARCHAR(255)并添加非空约束，表中无数据，但执行迁移时触发TypeError: sequence item 5: expected str instance, bytes found错误，日志显示SQL语句中出现字节串b'varchar(255)'。

错误日志 #

(.venv) lukehsiao@LukedeMacBook-Air migrations_playground % python migration_tool.py migrate
Applying all pending migrations...
['SELECT ', '`t1`', '.', '`id`', ', ', '`t1`', '.', '`name`', ', ', '`t1`', '.', '`migrated_at`', ' FROM ', '`migratehistory`', ' AS ', '`t1`', ' ORDER BY ', '`t1`', '.', '`id`'] <peewee.Context object at 0x1038eee50>
['SELECT ', '`t1`', '.', '`id`', ', ', '`t1`', '.', '`name`', ', ', '`t1`', '.', '`migrated_at`', ' FROM ', '`migratehistory`', ' AS ', '`t1`', ' ORDER BY ', '`t1`', '.', '`id`'] <peewee.Context object at 0x1038eee50>
<peewee_migrate.migrator.Migrator object at 0x1039072b0>
['ALTER TABLE ', '`testprocess`', ' MODIFY COLUMN ', '`f4`', ' ', 'VARCHAR(255)'] <peewee.Context object at 0x103981d60>
['ALTER TABLE ', '`testprocess`', ' MODIFY ', '`f4`', ' ', b'varchar(255)', ' ', 'NOT NULL'] <peewee.Context object at 0x103981e00>
Traceback (most recent call last):
  File "/Users/lukehsiao/Workspace/migrations_playground/migration_tool.py", line 150, in <module>
    cli()
  File "/Users/lukehsiao/Workspace/migrations_playground/.venv/lib/python3.9/site-packages/click/core.py", line 1157, in __call__
    return self.main(*args, **kwargs)
  File "/Users/lukehsiao/Workspace/migrations_playground/.venv/lib/python3.9/site-packages/click/core.py", line 1078, in main
    rv = self.invoke(ctx)
  File "/Users/lukehsiao/Workspace/migrations_playground/.venv/lib/python3.9/site-packages/click/core.py", line 1688, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/Users/lukehsiao/Workspace/migrations_playground/.venv/lib/python3.9/site-packages/click/core.py", line 1434, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/Users/lukehsiao/Workspace/migrations_playground/.venv/lib/python3.9/site-packages/click/core.py", line 783, in invoke
    return __callback(*args, **kwargs)
  File "/Users/lukehsiao/Workspace/migrations_playground/migration_tool.py", line 130, in migrate
    migration_controller.run_all()
  File "/Users/lukehsiao/Workspace/migrations_playground/migration_tool.py", line 64, in run_all
    self.router.run()
  File "/Users/lukehsiao/Workspace/migrations_playground/.venv/lib/python3.9/site-packages/peewee_migrate/router.py", line 229, in run
    done.append(self.run_one(mname, migrator, fake=fake, force=fake))
  File "/Users/lukehsiao/Workspace/migrations_playground/.venv/lib/python3.9/site-packages/peewee_migrate/router.py", line 200, in run_one
    migrator()
  File "/Users/lukehsiao/Workspace/migrations_playground/.venv/lib/python3.9/site-packages/peewee_migrate/migrator.py", line 76, in __call__
    op.run()
  File "/Users/lukehsiao/Workspace/migrations_playground/.venv/lib/python3.9/site-packages/playhouse/migrate.py", line 165, in run
    self._handle_result(method(*self.args, **kwargs))
  File "/Users/lukehsiao/Workspace/migrations_playground/.venv/lib/python3.9/site-packages/playhouse/migrate.py", line 159, in _handle_result
    self._handle_result(item)
  File "/Users/lukehsiao/Workspace/migrations_playground/.venv/lib/python3.9/site-packages/playhouse/migrate.py", line 156, in _handle_result
    result.run()
  File "/Users/lukehsiao/Workspace/migrations_playground/.venv/lib/python3.9/site-packages/playhouse/migrate.py", line 165, in run
    self._handle_result(method(*self.args, **kwargs))
  File "/Users/lukehsiao/Workspace/migrations_playground/.venv/lib/python3.9/site-packages/playhouse/migrate.py", line 154, in _handle_result
    self.execute(result)
  File "/Users/lukehsiao/Workspace/migrations_playground/.venv/lib/python3.9/site-packages/playhouse/migrate.py", line 150, in execute
    self.migrator.database.execute(node)
  File "/Users/lukehsiao/Workspace/migrations_playground/.venv/lib/python3.9/site-packages/peewee.py", line 3254, in execute
    sql, params = ctx.sql(query).query()
  File "/Users/lukehsiao/Workspace/migrations_playground/.venv/lib/python3.9/site-packages/peewee.py", line 673, in query
    return ''.join(self._sql), self._values
TypeError: sequence item 5: expected str instance, bytes found

相关代码 #

迁移文件 #

from contextlib import suppress

import peewee as pw
from peewee_migrate import Migrator


with suppress(ImportError):
    import playhouse.postgres_ext as pw_pext


def migrate(migrator: Migrator, database: pw.Database, *, fake=False):
    """Write your migrations here."""
    print(migrator)
    
    migrator.change_fields('testprocess', f4=pw.CharField(default='test', max_length=255))

    migrator.add_not_null('testprocess', 'f4')


def rollback(migrator: Migrator, database: pw.Database, *, fake=False):
    """Write your rollback migrations here."""
    
    migrator.drop_not_null('testprocess', 'f4')

模型代码 #

from peewee import *
from typing import Union, Dict
from test_models.fields import JSONField, EnumField, UnsignedIntegerField
from test_models.enums import ProjectDocumentType, KMSCaseType, QACheckType
from test_models import eb1a_models
from core.base import (AttachmentBase, Base, DateTimeBase, IdBase)
proxy = Proxy()

# TO Test

class MigrateHistory(Model):
    """Model representing the migratehistory table created by peewee_migrate."""
    id = AutoField(primary_key=True)
    name = CharField(max_length=255, null=False)
    migrated_at = DateTimeField(null=False)

    class Meta:
        db_table = "migratehistory"
        database = proxy

class User(Model):
    id = AutoField()
    name = CharField()
    age = IntegerField()
    happy = BooleanField()
    soc_number: Union[str, CharField] = CharField(max_length=10, null=True)
    body = JSONField(json_type=Dict[str, Union[int, str]], null=True)
    part_1: Union[eb1a_models.EB1AADEDraftPart1, JSONField] = JSONField(
        json_type=eb1a_models.EB1AADEDraftPart1,
        null=True
    )
    part_2: Union[eb1a_models.EB1AADEDraftPart1, JSONField] = JSONField(
        json_type=eb1a_models.EB1AADEDraftPart1,
        null=True
    )
    # type: Union[ProjectDocumentType, EnumField] = EnumField(enum_type=ProjectDocumentType, default=ProjectDocumentType.HAS_FUNDING)
    # case_type: Union[KMSCaseType, EnumField] = EnumField(enum_type=KMSCaseType, index=True)
    class Meta:
        database = proxy  # Use the proxy when defining your model
        table_name = 'users'  # Define the table name


class Checklist(IdBase, DateTimeBase):
    order_case_id: Union[int, UnsignedIntegerField] = UnsignedIntegerField(index=True)
    # check: Union[QACheckType, EnumField] = EnumField(enum_type=QACheckType, null=True)

    class Meta:
        table_name = 'Checklist'


class TestProcess(Model):
    f1 = CharField(null=True)
    f2 = CharField(null=True)
    f3 = CharField(null=True)
    f4 = CharField(default="test")

迁移调用代码 #

import click
import os
from peewee_migrate import Router, Migrator

from peewee import Proxy
from playhouse.mysql_ext import MySQLConnectorDatabase
from test_models.db_models import User,MigrateHistory

class DatabaseHandler:
    def __init__(self) -> None:
        self.proxy = Proxy()
        self.initialize_database()
        self.user_model = User  # Replace User with your actual model class
    
    def initialize_database(self):
        db = MySQLConnectorDatabase(
            'PLAYGROUND',
            host='localhost',
            port=3307,
            user='root',
            charset='utf8mb4'
        )
        self.proxy.initialize(db)  # Initialize the proxy with the actual database
        User.bind(self.proxy)
        MigrateHistory.bind(self.proxy)

    def query_data(self):
        users = User.select()
        print("Retrieved data:")
        for user in users:
            print(f"ID: {user.id}, Name: {user.name}, Age: {user.age}, Happy: {user.happy}")

    def insert_data(self):
            # Insert data into the table
        user_data = [
            {'name': 'John Doe', 'age': 30, 'happy': True},
            {'name': 'Jane Smith', 'age': 25, 'happy': False},
        ]

        with self.proxy.atomic():
            User.insert_many(user_data).execute()


class MigrationController:
    def __init__(self, db) -> None:
        self.db = db
        self.ignore_tables = ['attachmentbase', 'base', 'datetimebase', 'idbase', 'migratehistory']
        self.scan_dir = 'test_models'
        self.router = Router(db.proxy, ignore=self.ignore_tables)
        self.migrator = Migrator(self.db.proxy)
        
    
    def make_migrations(self, name: str):
        migration_name = self.router.create(name=name, auto=self.scan_dir)
        click.echo(f"Created migration: {migration_name}")

    def run(self, name: str):
        if name in [os.path.splitext(filename)[0] for filename in os.listdir(self.router.migrate_dir)]:
            self.router.run(name)
        else:
            click.echo(f"{name} doesn't exist")

    def run_all(self):
        self.router.run()
    
    def roll_back(self):
        self.router.rollback()

    def show_migrations(self):
        files = sorted(os.listdir(self.router.migrate_dir))
        migrations = self.get_migration_history()

        migration_dict = {name: time for name, time in migrations}

        for file in files:
            name_without_extension = os.path.splitext(file)[0]
            
            if name_without_extension in migration_dict:
                timestamp = migration_dict[name_without_extension].strftime('%Y-%m-%d %H:%M:%S')
                click.echo(f"[X] {file} => {timestamp}")
            else:
                click.echo(f"[ ] {file}")

    
    def get_migration_history(self):
        """Retrieve the list of applied migrations."""
        try:
            migrations = (MigrateHistory
                        .select()
                        .order_by(MigrateHistory.migrated_at.asc()))

            return [(migration.name, migration.migrated_at) for migration in migrations]

        except Exception as e:
            print(f"Error while fetching migration history: {e}")
            return []

@click.group()
def cli():
    pass


@cli.command()
@click.option('--name', required=True, help="Name of the migration to create.")
def makemigrations(name):
    click.echo("Generating migration files...")
    db = DatabaseHandler()
    migration_controller = MigrationController(db)
    migration_controller.make_migrations(name=name)


@cli.command()
def showmigrations():
    click.echo("Showing migrations...")
    db = DatabaseHandler()
    migration_controller = MigrationController(db)
    migration_controller.show_migrations()


@cli.command()
@click.option('--name', help="Name of the specific migration to apply.")
def migrate(name):
    db = DatabaseHandler()
    migration_controller = MigrationController(db)
    if name:
        click.echo(f"Applying migration: {name}")
        migration_controller.run(name)
    else:
        click.echo("Applying all pending migrations...")
        migration_controller.run_all()


@cli.command()
@click.option('--name', help="Name of the specific migration to rollback.")
@click.option('--count', type=int, help="Number of migrations to rollback.")
def rollback(name, count):
    db = DatabaseHandler()
    migration_controller = MigrationController(db)
    if name:
        click.echo(f"Rolling back migration: {name}")
    elif count:
        click.echo(f"Rolling back {count} migrations...")
        for _ in range(count):
            migration_controller.roll_back()
    else:
        click.echo("Please provide either --name or --count option.")


if __name__ == '__main__':
    cli()

根源分析 #

问题出在playhouse.migrate模块处理MySQL字段类型时，执行add_not_null操作时从数据库读取的字段类型为字节串（b'varchar(255)'），而非字符串，导致SQL语句拼接时类型不匹配。这是Peewee/playhouse与MySQL驱动交互时的编码处理bug，在单独修改字段约束时容易触发。

解决办法 #

方案1：合并字段修改与非空约束为一次操作 #

修改迁移文件，在change_fields中直接设置null=False，避免分两次操作：

def migrate(migrator: Migrator, database: pw.Database, *, fake=False):
    print(migrator)
    # 合并字段定义与非空约束
    migrator.change_fields('testprocess', f4=pw.CharField(default='test', max_length=255, null=False))

def rollback(migrator: Migrator, database: pw.Database, *, fake=False):
    migrator.change_fields('testprocess', f4=pw.CharField(default='test', null=True))

方案2：手动执行SQL语句 #

绕过Peewee Migrate的封装方法，直接执行ALTER TABLE语句：

def migrate(migrator: Migrator, database: pw.Database, *, fake=False):
    print(migrator)
    database.execute_sql("ALTER TABLE testprocess MODIFY COLUMN f4 VARCHAR(255) NOT NULL DEFAULT 'test';")

def rollback(migrator: Migrator, database: pw.Database, *, fake=False):
    database.execute_sql("ALTER TABLE testprocess MODIFY COLUMN f4 VARCHAR(255) DEFAULT 'test';")

方案3：升级Peewee及相关库 #

检查当前Peewee、peewee-migrate版本，升级到最新稳定版：

pip install --upgrade peewee peewee-migrate playhouse

方案4：临时补丁（无法升级时） #

如果无法升级库，可以在迁移文件中手动构造SQL避免字节串问题：

# 仅作临时应急，不推荐长期使用
def migrate(migrator: Migrator, database: pw.Database, *, fake=False):
    print(migrator)
    migrator.change_fields('testprocess', f4=pw.CharField(default='test', max_length=255))
    
    # 手动执行非空约束修改
    migrator.database.execute_sql("ALTER TABLE testprocess MODIFY f4 VARCHAR(255) NOT NULL;")

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