SQLAlchemy是Python中最流行的ORM（对象关系映射）框架之一，它提供了高效且灵活的数据库操作方式。本文将介绍如何使用SQLAlchemy ORM进行数据库操作。

目录

安装SQLAlchemy

核心概念

连接数据库

定义数据模型

创建数据库表

基本CRUD操作

查询数据

关系操作

事务管理

最佳实践

安装

bash

pip install sqlalchemy

如果需要连接特定数据库，还需安装相应的驱动程序：

bash

# PostgreSQL

pip install psycopg2-binary

# MySQL

pip install mysql-connector-python

# SQLite (Python标准库已包含，无需额外安装)

核心概念

Engine：数据库连接的引擎，负责与数据库通信

Session：数据库会话，管理所有持久化操作

Model：数据模型类，对应数据库中的表

Query：查询对象，用于构建和执行数据库查询

连接数据库

python

from sqlalchemy import create_engine

from sqlalchemy.orm import sessionmaker

# 创建数据库连接引擎

# SQLite示例

engine = create_engine('sqlite:///example.db', echo=True)

# PostgreSQL示例

# engine = create_engine('postgresql://username:password@localhost:5432/mydatabase')

# MySQL示例

# engine = create_engine('mysql+mysqlconnector://username:password@localhost:3306/mydatabase')

# 创建会话工厂

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 创建会话实例

session = SessionLocal()

定义数据模型

python

from sqlalchemy import Column, Integer, String, ForeignKey

from sqlalchemy.orm import relationship, declarative_base

# 创建基类

Base = declarative_base()

class User(Base):

__tablename__ = 'users'

id = Column(Integer, primary_key=True, index=True)

name = Column(String(50), nullable=False)

email = Column(String(100), unique=True, index=True)

# 定义一对多关系

posts = relationship("Post", back_populates="author")

class Post(Base):

__tablename__ = 'posts'

id = Column(Integer, primary_key=True, index=True)

title = Column(String(100), nullable=False)

content = Column(String(500))

author_id = Column(Integer, ForeignKey('users.id'))

# 定义多对一关系

author = relationship("User", back_populates="posts")

# 定义多对多关系（通过关联表）

tags = relationship("Tag", secondary="post_tags", back_populates="posts")

class Tag(Base):

__tablename__ = 'tags'

id = Column(Integer, primary_key=True, index=True)

name = Column(String(30), unique=True, nullable=False)

posts = relationship("Post", secondary="post_tags", back_populates="tags")

# 关联表（用于多对多关系）

class PostTag(Base):

__tablename__ = 'post_tags'

post_id = Column(Integer, ForeignKey('posts.id'), primary_key=True)

tag_id = Column(Integer, ForeignKey('tags.id'), primary_key=True)

创建数据库表

python

# 创建所有表

Base.metadata.create_all(bind=engine)

# 删除所有表

# Base.metadata.drop_all(bind=engine)

基本CRUD操作

创建数据

python

# 创建新用户

new_user = User(name="张三", email="zhangsan@example.com")

session.add(new_user)

session.commit()

# 批量创建

session.add_all([

User(name="李四", email="lisi@example.com"),

User(name="王五", email="wangwu@example.com")

])

session.commit()

读取数据

python

# 获取所有用户

users = session.query(User).all()

# 获取第一个用户

first_user = session.query(User).first()

# 根据ID获取用户

user = session.query(User).get(1)

更新数据

python

# 查询并更新

user = session.query(User).get(1)

user.name = "张三四"

session.commit()

# 批量更新

session.query(User).filter(User.name.like("张%")).update({"name": "张氏"}, synchronize_session=False)

session.commit()

删除数据

python

# 查询并删除

user = session.query(User).get(1)

session.delete(user)

session.commit()

# 批量删除

session.query(User).filter(User.name == "李四").delete(synchronize_session=False)

session.commit()

查询数据

基本查询

python

# 获取所有记录

users = session.query(User).all()

# 获取特定字段

names = session.query(User.name).all()

# 排序

users = session.query(User).order_by(User.name.desc()).all()

# 限制结果数量

users = session.query(User).limit(10).all()

# 偏移量

users = session.query(User).offset(5).limit(10).all()

过滤查询

python

from sqlalchemy import or_

# 等值过滤

user = session.query(User).filter(User.name == "张三").first()

# 模糊查询

users = session.query(User).filter(User.name.like("张%")).all()

# IN查询

users = session.query(User).filter(User.name.in_(["张三", "李四"])).all()

# 多条件查询

users = session.query(User).filter(

User.name == "张三",

User.email.like("%@example.com")

).all()

# 或条件

users = session.query(User).filter(

or_(User.name == "张三", User.name == "李四")

).all()

# 不等于

users = session.query(User).filter(User.name != "张三").all()

聚合查询

python

from sqlalchemy import func

# 计数

count = session.query(User).count()

# 分组计数

user_post_count = session.query(

User.name,

func.count(Post.id)

).join(Post).group_by(User.name).all()

# 求和、平均值等

avg_id = session.query(func.avg(User.id)).scalar()

连接查询

python

# 内连接

results = session.query(User, Post).join(Post).filter(Post.title.like("%Python%")).all()

# 左外连接

results = session.query(User, Post).outerjoin(Post).all()

# 指定连接条件

results = session.query(User, Post).join(Post, User.id == Post.author_id).all()

关系操作

python

# 创建带关系的对象

user = User(name="赵六", email="zhaoliu@example.com")

post = Post(title="我的第一篇博客", content="Hello World!", author=user)

session.add(post)

session.commit()

# 通过关系访问

print(f"文章 '{post.title}' 的作者是 {post.author.name}")

print(f"用户 {user.name} 的所有文章:")

for p in user.posts:

print(f" - {p.title}")

# 多对多关系操作

python_tag = Tag(name="Python")

sqlalchemy_tag = Tag(name="SQLAlchemy")

post.tags.append(python_tag)

post.tags.append(sqlalchemy_tag)

session.commit()

print(f"文章 '{post.title}' 的标签:")

for tag in post.tags:

print(f" - {tag.name}")

事务管理

python

# 自动提交事务

try:

user = User(name="测试用户", email="test@example.com")

session.add(user)

session.commit()

except Exception as e:

session.rollback()

print(f"发生错误: {e}")

# 使用事务上下文管理器

from sqlalchemy.orm import Session

def create_user(session: Session, name: str, email: str):

try:

user = User(name=name, email=email)

session.add(user)

session.commit()

return user

except:

session.rollback()

raise

# 嵌套事务

with session.begin_nested():

user = User(name="事务用户", email="transaction@example.com")

session.add(user)

# 保存点

savepoint = session.begin_nested()

try:

user = User(name="保存点用户", email="savepoint@example.com")

session.add(user)

savepoint.commit()

except:

savepoint.rollback()

最佳实践

会话管理：为每个请求创建新会话，请求结束后关闭

异常处理：始终处理异常并适当回滚事务

延迟加载：注意N+1查询问题，使用 eager loading 优化

连接池：合理配置连接池大小和超时设置

数据验证：在模型层或应用层验证数据完整性

python

# 使用上下文管理器管理会话

from contextlib import contextmanager

@contextmanager

def get_db():

db = SessionLocal()

try:

yield db

db.commit()

except Exception:

db.rollback()

raise

finally:

db.close()

# 使用示例

with get_db() as db:

user = User(name="上下文用户", email="context@example.com")

db.add(user)

总结

SQLAlchemy ORM提供了强大而灵活的数据库操作方式，通过本文的介绍，您应该能够：

安装和配置SQLAlchemy

定义数据模型和关系

执行基本的CRUD操作

构建复杂查询

管理数据库事务

遵循最佳实践

SQLAlchemy还有更多高级特性，如混合属性、事件监*、自定义查询等，值得进一步探索学习。

这是一个非常经典的Spring面试题，也是理解Spring核心思想的关键。很多人会把它们混为一谈，但实际上它们描述的是同一个概念的两个不同侧面。

让我用一个非常通俗易懂的方式来解释。

核心结论

• 控制反转（IoC）是一种设计原则或思想。

• 依赖注入（DI）是实现控制反转这种思想的一种具体技术模式。

可以把它们的关系理解为：

“战略” vs “战术”

• IoC是战略目标：我们要将程序的控制权反转过来。

• DI是战术手段：我们通过“注入”依赖的方式来实现这个目标。

1. 控制反转（Inversion of Control - IoC）

核心思想： 将对象的创建、组装、生命周期的控制权，从应用程序代码中“反转”给一个外部容器（在Spring中就是IoC容器）。

传统开发模式（正转）：

• 你需要一个对象（比如Car），你会在代码里直接new Car()。

• 你的代码主动去创建和管理它所依赖的对象。

• 控制权在程序本身。

java

// 传统方式：程序员自己控制
public class Car {
    private Engine engine;
    public Car() {
        // 程序员主动创建依赖
        this.engine = new Engine(); // 控制权在Car类内部
    }
}

控制反转模式（反转）：

• 你不再自己new对象，而是由一个“容器”来帮你创建好、组装好，然后“送”给你。

• 你的代码是被动接收它所需要的依赖。

• 控制权转移到了外部容器。

java

// IoC方式：容器控制
public class Car {
    private Engine engine;
    // 容器会负责把创建好的Engine对象“注入”到这里
    public Car(Engine engine) { // 控制权在容器
        this.engine = engine;
    }
}

IoC容器的职责：

• 创建对象

• 弄清楚对象之间的依赖关系（比如Car依赖Engine）

• 将依赖的对象装配到一起

• 管理对象的生命周期（何时创建，何时销毁）

2. 依赖注入（Dependency Injection - DI）

核心思想： 是实现IoC的具体技术。它描述的是“如何”将依赖关系传递给一个组件。具体来说，就是容器通过构造函数、Setter方法或接口等方式，将依赖对象“注入”到需要它的组件中。

依赖注入主要有三种方式：

1. 构造器注入（推荐）

容器通过调用类的构造函数来注入依赖。

java

@Component
public class Car {
    private final Engine engine;
    // 容器会在这里注入一个Engine实例
    @Autowired // Spring 4.3后，如果只有一个构造器，可省略此注解
    public Car(Engine engine) {
        this.engine = engine;
    }
}

2. Setter方法注入

容器通过调用类的Setter方法来注入依赖。

java

@Component
public class Car {
    private Engine engine;
    // 容器会调用这个setter方法来注入Engine
    @Autowired
    public void setEngine(Engine engine) {
        this.engine = engine;
    }
}

3. 字段注入（现在不推荐）

容器通过反射直接给字段赋值。

java

@Component
public class Car {
    @Autowired // 直接注入到字段
    private Engine engine;
}

用一个生活中的比喻来总结

假设你想喝一杯咖啡。

传统模式（正转）：

1. 你自己去种咖啡豆。

2. 你自己去烘焙、研磨。

3. 你自己用咖啡机冲泡。

4. 你喝到咖啡。
   

   -> 你控制了所有环节，非常繁琐，高度耦合。

控制反转模式（IoC）：

• 你走进一家咖啡馆（IoC容器）。

• 你直接向服务员点单：“我要一杯拿铁”。

• 服务员（容器）去后厨（创建并组装对象）让咖啡师（Bean）用咖啡机（另一个Bean）和牛奶（另一个Bean）做好一杯拿铁。

• 服务员把这杯做好的拿铁（组装好的对象）递给你。
  

  -> 你不再关心制作过程，只关心最终结果。控制权反转给了咖啡馆。

依赖注入（DI）在这个比喻中：

• 就是服务员把做好的拿铁“递给你”的这个“递”的动作。

• 它是实现“咖啡馆为你服务”这个“控制反转”思想的具体方式。

总结对比表

在Spring框架中，我们通过使用依赖注入这种最主流的方式，来实现控制反转这个崇高的设计目标，最终达到代码解耦、更易测试、更灵活的目的。所以当我们在谈论Spring的IoC时，通常指的就是它的依赖注入功能。