面试题：Python中使用SQLAlchemy实现PostgreSQL数据库事务处理

Question

在Python中，使用SQLAlchemy库连接到PostgreSQL数据库，创建一个包含两个表（'orders'和'order_items'）的数据库模式，并实现一个事务处理逻辑。事务需要完成以下操作：向'orders'表插入一条订单记录，同时向'order_items'表插入对应订单的若干条订单项记录。如果其中任何一个插入操作失败，整个事务需要回滚。请详细写出实现代码。

Answer 1

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base

# 创建数据库引擎
engine = create_engine('postgresql://user:password@localhost/mydb')
Session = sessionmaker(bind=engine)
session = Session()
Base = declarative_base()

# 定义orders表
class Order(Base):
    __tablename__ = 'orders'
    id = Column(Integer, primary_key=True)
    order_number = Column(String)
    order_items = relationship('OrderItem', backref='order')

# 定义order_items表
class OrderItem(Base):
    __tablename__ = 'order_items'
    id = Column(Integer, primary_key=True)
    order_id = Column(Integer, ForeignKey('orders.id'))
    product_name = Column(String)

try:
    # 创建表结构
    Base.metadata.create_all(engine)

    # 开始事务
    new_order = Order(order_number='12345')
    session.add(new_order)

    item1 = OrderItem(product_name='Product 1', order=new_order)
    item2 = OrderItem(product_name='Product 2', order=new_order)
    session.add_all([item1, item2])

    # 提交事务
    session.commit()
except Exception as e:
    # 回滚事务
    session.rollback()
    print(f"事务处理失败: {e}")
finally:
    session.close()