Skip to main content

SQLAlchemy 面试

以下是关于 SQLAlchemy 的常见面试问题，涵盖基础概念、ORM 使用、核心 SQL 操作、性能优化和实际应用场景，适合准备 Python 数据库相关的技术面试：

1. SQLAlchemy 基础概念​

Q1. SQLAlchemy 是什么？它的主要组件有哪些？​

  • 答案：
    • SQLAlchemy 是一个 Python 的 SQL 工具包和 ORM（Object-Relational Mapping）框架，用于高效操作数据库。
    • 主要组件：
      • Core：底层 SQL 操作（类似传统 SQL 查询）。
      • ORM（Object-Relational Mapping）：将数据库表映射为 Python 类，对象化操作数据库。
      • Engine：数据库连接驱动（如 sqlite://, postgresql://）。
      • Session：管理数据库会话（事务、缓存）。

2. ORM 基础​

Q2. 如何定义一个 SQLAlchemy ORM 模型？示例​

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

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'  # 表名
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100), unique=True)

# 初始化数据库连接
engine = create_engine('sqlite:///example.db')
Base.metadata.create_all(engine)  # 创建表
Session = sessionmaker(bind=engine)
session = Session()

关键点：

  • declarative_base() 定义模型基类。
  • __tablename__ 指定表名。
  • Column 定义字段类型（如 Integer, String）。

3. CRUD 操作​

Q3. 如何使用 SQLAlchemy 进行增删改查（CRUD）？​

(1) 插入数据（Create）​

new_user = User(name="Alice", email="[email protected]")
session.add(new_user)
session.commit()  # 提交事务

(2) 查询数据（Read）​

# 查询所有用户
users = session.query(User).all()

# 条件查询（filter_by / filter）
user = session.query(User).filter_by(name="Alice").first()
user = session.query(User).filter(User.name == "Alice").first()

(3) 更新数据（Update）​

user = session.query(User).filter_by(name="Alice").first()
user.email = "[email protected]"
session.commit()

(4) 删除数据（Delete）​

user = session.query(User).filter_by(name="Alice").first()
session.delete(user)
session.commit()

4. 高级查询​

Q4. 如何实现复杂查询（JOIN、GROUP BY、子查询）？​

(1) JOIN 查询​

# 假设有另一个模型 Post
class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True)
    title = Column(String(100))
    user_id = Column(Integer, ForeignKey('users.id'))
    user = relationship("User", back_populates="posts")

# 查询用户及其所有文章
users = session.query(User).join(Post).all()

(2) GROUP BY 和聚合函数​

from sqlalchemy import func

# 统计每个用户的文章数量
result = session.query(
    User.name,
    func.count(Post.id).label("post_count")
).join(Post).group_by(User.name).all()

(3) 子查询​

subquery = session.query(Post.user_id).filter(Post.title.like("%Python%")).subquery()
users = session.query(User).filter(User.id.in_(subquery)).all()

5. 性能优化​

Q5. SQLAlchemy 如何优化查询性能？​

(1) 延迟加载 vs 立即加载​

  • 默认延迟加载：访问关联数据时才查询（可能引发 N+1 问题）。

  • 立即加载（joinedload / selectinload）：

    from sqlalchemy.orm import joinedload

    # 一次查询加载用户及其文章
    users = session.query(User).options(joinedload(User.posts)).all()

(2) 批量插入​

# 低效：逐条插入
for user in user_list:
    session.add(user)
    session.commit()

# 高效：批量插入
session.bulk_save_objects(user_list)
session.commit()

(3) 使用 Core 替代 ORM​

对于复杂查询，直接使用 SQLAlchemy Core 可能更快：

from sqlalchemy import text
result = engine.execute(text("SELECT * FROM users WHERE name = :name"), {"name": "Alice"})

6. 事务管理​

Q6. 如何管理事务？如何处理异常回滚？​

try:
    user = User(name="Bob")
    session.add(user)
    session.flush()  # 临时提交到数据库（可捕获主键冲突等错误）

    # 其他操作...
    session.commit()  # 最终提交
except Exception as e:
    session.rollback()  # 回滚事务
    print(f"Error: {e}")

7. 开放性问题​

Q7. 你遇到过哪些 SQLAlchemy 的坑？如何解决的？​

  • N+1 查询问题：使用 joinedload 或 selectinload 优化关联查询。
  • 会话管理不当：确保每个请求使用独立 Session，避免跨请求共享。
  • 性能瓶颈：用 bulk_save_objects 替代单条插入，或直接执行原生 SQL。
  • 循环引用模型：使用 relationship 的 lazy="dynamic" 或手动控制加载策略。

总结​

考察方向关键点
ORM 基础模型定义、CRUD 操作、relationship
高级查询JOIN、GROUP BY、子查询
性能优化批量操作、加载策略、Core 替代 ORM
事务管理commit / rollback、异常处理
实战问题N+1 问题、会话生命周期管理

掌握这些问题可以全面展现你对 SQLAlchemy 的深入理解！