SQLAlchemy使用笔记（二）

简介：SQLAlchemy 是 Python 中最流行的 ORM（对象关系映射）工具之一，它允许开发者通过操作Python 类和对象实现对数据库的增删改查，无需直接编写 SQL，同时也保留了底层 SQL 的灵活性。

一、表结构

metadata = MetaData()
orders = Table('orders',metadata,Column('id', Integer, primary_key=True, autoincrement=True, comment='订单ID'),Column('user_id', Integer, nullable=False, comment='用户ID'),Column('total_amount', Float, nullable=False, comment='订单总金额'),Column('payment_method', String(30), comment='支付方式'),Column('shipping_address', String(200), comment='收货地址')
)

二、查询操作

# 查询特定用户的所有订单
def get_user_orders(user_id):query = select(orders).where(orders.c.user_id == user_id)with engine.connect() as conn:result = conn.execute(query)return [dict(row) for row in result]  # 转换为字典列表返回

# 查询金额大于指定值的订单
def get_high_value_orders(min_amount):query = select(orders).where(orders.c.total_amount > min_amount).order_by(orders.c.total_amount.desc())#按订单金额降序排列with engine.connect() as conn:result = conn.execute(query)return [dict(row) for row in result]

• 多条件"and"查询

from sqlalchemy import and_# 查询：用户ID=100 且 订单金额>500 且 支付方式为'credit_card'的订单
query = select(orders).where(and_(orders.c.user_id == 100,orders.c.total_amount > 500,orders.c.payment_method == 'credit_card')
)# 等价写法：链式调用where()（默认就是AND关系）
query = select(orders).where(orders.c.user_id == 100)\.where(orders.c.total_amount > 500)\.where(orders.c.payment_method == 'credit_card')

• 多条件"or"查询

from sqlalchemy import or_# 查询：订单金额>1000 或 支付方式为'wechat'的订单
query = select(orders).where(or_(orders.c.total_amount > 1000,orders.c.payment_method == 'wechat')
)

• 多条件"and"和"or"查询

from sqlalchemy import and_, or_# 查询：(用户ID=100 且 金额>500) 或 (支付方式为'alipay' 且 金额>2000)的订单
query = select(orders).where(or_(and_(orders.c.user_id == 100,orders.c.total_amount > 500),and_(orders.c.payment_method == 'alipay',orders.c.total_amount > 2000))
)

三、更新操作

# 更新特定订单的支付方式
def update_payment_method(order_id, new_method):# 1、构建更新语句update_stmt = update(orders).where(orders.c.id == order_id).values(payment_method=new_method)with engine.connect() as conn:# 2、执行更新result = conn.execute(update_stmt)# 3、提交事务（对于需要事务的数据库引擎）conn.commit()# 4、返回受影响的行数return result.rowcount

# 多字段更新：同时更新支付方式和收货地址
def update_order_details(order_id, new_method, new_address):update_stmt = update(orders).where(orders.c.id == order_id).values(payment_method=new_method,shipping_address=new_address)with engine.connect() as conn:result = conn.execute(update_stmt)conn.commit()return result.rowcount

# 条件更新：批量更新满足条件的订单
def batch_update_orders(min_amount, new_method):# 更新所有金额大于min_amount的订单支付方式update_stmt = update(orders).where(orders.c.total_amount > min_amount).values(payment_method=new_method)with engine.connect() as conn:result = conn.execute(update_stmt)conn.commit()print(f"共更新了 {result.rowcount} 条订单")return result.rowcount

四、删除操作

# 删除单个订单（根据ID）
def delete_single_order(order_id):# 构建删除语句，务必添加where条件delete_stmt = delete(orders).where(orders.c.id == order_id)with engine.connect() as conn:# 执行删除result = conn.execute(delete_stmt)# 提交事务conn.commit()# 返回受影响的行数（确认是否删除成功）return result.rowcount

# 批量删除满足条件的订单
def batch_delete_orders(user_id=None, max_amount=None):# 构建基础删除语句delete_stmt = delete(orders)# 收集删除条件conditions = []if user_id:conditions.append(orders.c.user_id == user_id)if max_amount:conditions.append(orders.c.total_amount < max_amount)# 添加条件（若没有条件则不执行删除，避免误删全表）if conditions:from sqlalchemy import and_delete_stmt = delete_stmt.where(and_(*conditions))else:print("未提供删除条件，取消操作")return 0with engine.connect() as conn:result = conn.execute(delete_stmt)conn.commit()print(f"共删除了 {result.rowcount} 条订单")return result.rowcount

# 删除特定支付方式的订单
def delete_by_payment_method(payment_method):delete_stmt = delete(orders).where(orders.c.payment_method == payment_method)with engine.connect() as conn:# 先查询确认要删除的记录数量from sqlalchemy import selectcount = conn.execute(select(orders).where(orders.c.payment_method == payment_method)).rowcountif count == 0:print(f"没有找到支付方式为 {payment_method} 的订单")return 0# 确认后执行删除result = conn.execute(delete_stmt)conn.commit()print(f"已删除 {result.rowcount} 条支付方式为 {payment_method} 的订单")return result.rowcount