使用SQLAlchemy操作SQLite插入数据报错：operation parameter must be str

问题场景 #

仅使用SQLite与SQLAlchemy技术栈，需实现：

• 向users表插入员工数据（含姓名、随机日期、从roles表随机选角色）
• 查询users表最新5条记录

但插入roles表时触发ValueError: operation parameter must be str，相关代码与报错信息如下：

原代码 #

# запись работника в таблицу users(ФИО, рандом др, роль(рандом из таблицы roles)
# вывод последних добавленных 5 работников
import json
import sqlalchemy as sa
import sqlite3
from sqlalchemy import Table, MetaData
metadata = sa.MetaData()

roles = sa.Table('roles', metadata,
                 sa.Column('id', sa.Integer, primary_key=True),
                 sa.Column('name', sa.String(50)))

users = sa.Table('users', metadata,
                 sa.Column('id', sa.BIGINT, primary_key=True),
                 sa.Column('fio', sa.String(255)),
                 sa.Column('datar', sa.String(255)),
                 sa.Column('id_role', sa.Integer))

with open("C:\Users\kiril\PycharmProjects\testbotjob_project\config.json", 'r',
          encoding='utf-8') as f:  # открыли файл
    options = json.load(f)  # загнали все из файла в переменную

db = options["data_base"]  # путь к бд

con = sqlite3.connect(db)
con.execute(roles.insert().values(name='Грузчик'))
con.close()

# with con:
#     con.execute("""
#             CREATE TABLE roles (
#                 id INT PRIMARY KEY,
#                 name VARCHAR(50)
#             );
#         """)
#     con.execute("""
#         CREATE TABLE users (
#             id BIGINT PRIMARY KEY,
#             fio VARCHAR(30),
#             datar DATE,
#             id_role INT,
#             FOREIGN KEY (id_role) REFERENCES roles (id)
#         );
#     """)

报错信息 #

Traceback (most recent call last):
  File "C:/Users/kiril/PycharmProjects/testbotjob_project/database/db_con.py", line 26, in <module>
    con.execute(roles.insert().values(name='Грузчик'))
ValueError: operation parameter must be str

问题原因 #

你混用了原生sqlite3连接和SQLAlchemy表达式：sqlite3.connect()返回的原生连接仅支持执行字符串格式的SQL语句，而roles.insert().values(...)是SQLAlchemy构建的Insert对象，并非字符串，因此触发类型错误。

解决办法 #

改用SQLAlchemy的引擎与连接对象来操作，避免混用原生sqlite3接口，同时完善表结构与业务逻辑：

修正后的完整代码 #

import json
import sqlalchemy as sa
from sqlalchemy import Table, MetaData
from datetime import datetime, timedelta
import random

metadata = sa.MetaData()

# 定义表结构，添加外键约束与自增主键
roles = sa.Table('roles', metadata,
                 sa.Column('id', sa.Integer, primary_key=True, autoincrement=True),
                 sa.Column('name', sa.String(50), unique=True))

users = sa.Table('users', metadata,
                 sa.Column('id', sa.BIGINT, primary_key=True, autoincrement=True),
                 sa.Column('fio', sa.String(255)),
                 sa.Column('datar', sa.Date),  # 改用Date类型更贴合业务
                 sa.Column('id_role', sa.Integer, sa.ForeignKey('roles.id')))

# 读取数据库配置并构造SQLAlchemy连接URL
with open("C:\Users\kiril\PycharmProjects\testbotjob_project\config.json", 'r', encoding='utf-8') as f:
    options = json.load(f)
db_url = f"sqlite:///{options['data_base']}"

# 创建引擎并自动生成表（不存在则创建）
engine = sa.create_engine(db_url)
metadata.create_all(engine)

# 插入角色数据（避免重复插入）
with engine.connect() as conn:
    existing_role = conn.execute(sa.select(roles).where(roles.c.name == 'Грузчик')).first()
    if not existing_role:
        conn.execute(roles.insert().values(name='Грузчик'))
        conn.commit()

# 生成随机日期的辅助函数
def generate_random_date():
    start_date = datetime(1980, 1, 1)
    end_date = datetime.now()
    delta = end_date - start_date
    random_days = random.randint(0, delta.days)
    return start_date + timedelta(days=random_days)

# 插入员工数据（含随机日期、随机角色）
with engine.connect() as conn:
    # 获取所有可用角色ID
    role_ids = [row[0] for row in conn.execute(sa.select(roles.c.id)).fetchall()]
    if role_ids:
        # 插入测试数据
        conn.execute(users.insert().values(
            fio='Иванов Иван Иванович',
            datar=generate_random_date(),
            id_role=random.choice(role_ids)
        ))
        conn.commit()

# 查询最新5条员工记录（关联角色表获取角色名称）
with engine.connect() as conn:
    latest_users = conn.execute(
        sa.select(users, roles.c.name.label('role_name'))
          .join(roles, users.c.id_role == roles.c.id)
          .order_by(users.c.id.desc())
          .limit(5)
    ).fetchall()

    print("最新5条员工记录：")
    for user in latest_users:
        print(f"ID: {user.id}, ФИО: {user.fio}, Дата: {user.datar}, Роль: {user.role_name}")

关键优化点 #

• 使用sa.create_engine()替代原生sqlite3.connect()，适配SQLAlchemy表达式执行
• 通过metadata.create_all(engine)自动创建表，无需手动编写CREATE TABLE语句
• 为users表添加外键约束，保证数据关联的完整性
• 将datar字段改为sa.Date类型，更符合日期存储的业务逻辑
• 增加角色重复插入校验、随机日期生成、随机角色选取的逻辑，满足需求
• 查询时关联roles表，直接获取角色名称，无需额外查询

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