安装配置

执行以下命令安装核心库：

pip install sqlalchemy

数据库驱动按需安装：

# PostgreSQL适配器
pip install psycopg2

# MySQL连接器
pip install mysqlclient

核心组件

• 数据库引擎：管理底层连接池
• 会话对象：操作原子性控制单元
• 模型类：表结构的Python映射

数据库连接

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 初始化SQLite引擎
db_engine = create_engine('sqlite:///data.db', echo=False)

# 配置会话工厂
SessionFactory = sessionmaker(bind=db_engine, autocommit=False)
current_session = SessionFactory()

模型定义

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

ModelBase = declarative_base()

class Employee(ModelBase):
    __tablename__ = 'employees'
    
    employee_id = Column(Integer, primary_key=True)
    full_name = Column(String(50))
    contact_email = Column(String(100), unique=True)
    
    assignments = relationship("Task", back_populates="owner")

class Task(ModelBase):
    __tablename__ = 'tasks'
    
    task_id = Column(Integer, primary_key=True)
    task_title = Column(String(100))
    owner_id = Column(Integer, ForeignKey('employees.employee_id'))
    
    owner = relationship("Employee", back_populates="assignments")
    
    labels = relationship("Tag", secondary="task_labels", back_populates="tasks")

# 多对多关联表
class TaskLabel(ModelBase):
    __tablename__ = 'task_labels'
    
    task_ref = Column(Integer, ForeignKey('tasks.task_id'), primary_key=True)
    tag_ref = Column(Integer, ForeignKey('tags.tag_id'), primary_key=True)

表结构生成

# 创建所有模型对应的表
ModelBase.metadata.create_all(bind=db_engine)

数据操作

新增记录

new_employee = Employee(full_name="陈明", contact_email="chen@domain.com")
current_session.add(new_employee)
current_session.commit()

查询记录

# 条件过滤
employee = current_session.query(Employee).filter_by(full_name="陈明").first()

# 关联查询
results = current_session.query(Employee, Task).join(Task).all()

更新记录

employee.contact_email = "new_email@domain.com"
current_session.commit()

删除记录

current_session.delete(employee)
current_session.commit()

高级查询

from sqlalchemy import func, or_

# 聚合统计
task_count = current_session.query(func.count(Task.task_id)).scalar()

# 多条件筛选
employees = current_session.query(Employee).filter(
    or_(Employee.full_name.like("陈%"), 
        Employee.contact_email.contains("domain"))
).limit(5).all()

关系管理

# 添加关联对象
dev_tag = Tag(name="开发")
new_task = Task(task_title="API优化", owner=new_employee)
new_task.labels.append(dev_tag)
current_session.commit()

事务控制

try:
    emp = Employee(full_name="测试员工", contact_email="test@test.com")
    current_session.add(emp)
    current_session.commit()
except Exception:
    current_session.rollback()
    raise

优化实践

• 使用会话上下文管理资源
• 预加载关联对象避免N+1查询
• 设置连接池超时参数

from contextlib import contextmanager

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