安装配置

安装核心包与数据库驱动：

pip install sqlalchemy
# PostgreSQL驱动
pip install psycopg2
# MySQL驱动
pip install mysqlclient

核心组件

• Engine：数据库连接引擎
• Session：持久化操作管理器
• Declarative Base：模型定义基类
• Query：数据库查询构造器

数据库连接

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

db_engine = create_engine('postgresql://user:pass@localhost:5432/mydb')
SessionFactory = sessionmaker(bind=db_engine)
db_session = SessionFactory()

数据模型定义

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, declarative_base

ModelBase = declarative_base()

class Customer(ModelBase):
    __tablename__ = 'customers'
    customer_id = Column(Integer, primary_key=True)
    full_name = Column(String(50))
    orders = relationship("PurchaseOrder", backref="buyer")

class PurchaseOrder(ModelBase):
    __tablename__ = 'orders'
    order_id = Column(Integer, primary_key=True)
    product = Column(String(100))
    customer_ref = Column(Integer, ForeignKey('customers.customer_id'))

# 创建表结构
ModelBase.metadata.create_all(bind=db_engine)

数据操作

创建记录

new_customer = Customer(full_name="李晓明")
db_session.add(new_customer)
db_session.commit()

查询操作

# 条件查询
customers = db_session.query(Customer).filter(Customer.full_name.like("李%")).all()

# 关联查询
results = db_session.query(Customer, PurchaseOrder).join(PurchaseOrder).all()

更新删除

# 更新记录
cust = db_session.query(Customer).get(1)
cust.full_name = "张伟"
db_session.commit()

# 删除记录
db_session.query(Customer).filter(Customer.customer_id == 2).delete()
db_session.commit()

关系管理

cust = Customer(full_name="王芳")
order1 = PurchaseOrder(product="笔记本电脑", buyer=cust)
order2 = PurchaseOrder(product="智能手机", buyer=cust)
db_session.add_all([cust, order1, order2])
db_session.commit()

# 访问关联对象
for order in cust.orders:
    print(f"{cust.full_name} 订购了 {order.product}")

事务控制

try:
    db_session.begin_nested()
    # 数据库操作
    db_session.commit()
except Exception:
    db_session.rollback()

最佳实践

• 使用上下文管理器管理会话生命周期
• 采用预加载优化关联查询性能
• 合理配置数据库连接池参数

from contextlib import contextmanager

@contextmanager
def db_context():
    session = SessionFactory()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()