Python SQLAlchemy ORM 数据库操作指南 | 极客日志

Python

Python SQLAlchemy ORM 数据库操作指南

Python 中流行的 ORM 框架 SQLAlchemy 的使用方法，涵盖安装配置、核心概念（Engine、Session、Model）、数据库连接、数据模型定义及表创建。内容还包括基本 CRUD 操作、复杂查询（过滤、聚合、连接）、关系操作（一对多、多对多）以及事务管理。最后总结了会话管理、异常处理等最佳实践，帮助开发者高效进行数据库交互。

CodeArtist发布于 2026/3/30更新于 2026/4/131 浏览

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

安装

pip install sqlalchemy

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

# PostgreSQL
pip install psycopg2-binary
# MySQL
pip install mysql-connector-python
# SQLite (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()

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)

# 创建所有表
Base.metadata.create_all(bind=engine)
# 删除所有表
# Base.metadata.drop_all(bind=engine)

# 创建新用户
new_user = User(name="张三", email="[email protected]")
session.add(new_user)
session.commit()

# 批量创建
session.add_all([
    User(name="李四", email="[email protected]"),
    User(name="王五", email="[email protected]")
])
session.commit()

# 获取所有用户
users = session.query(User).all()
# 获取第一个用户
first_user = session.query(User).first()
# 根据ID获取用户
user = session.query(User).get(1)

# 查询并更新
user = session.query(User).get(1)
user.name = "张三四"
session.commit()

# 批量更新
session.query(User).filter(User.name.like("张%"))\
    .update({"name": "张氏"}, synchronize_session=False)
session.commit()

# 查询并删除
user = session.query(User).get(1)
session.delete(user)
session.commit()

# 批量删除
session.query(User).filter(User.name == "李四")\
    .delete(synchronize_session=False)
session.commit()

# 获取所有记录
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()

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()

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()

# 内连接
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()

# 创建带关系的对象
user = User(name="赵六", email="[email protected]")
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}")

# 自动提交事务
try:
    user = User(name="测试用户", email="[email protected]")
    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="[email protected]")
    session.add(user)
    # 保存点
    savepoint = session.begin_nested()
    try:
        user = User(name="保存点用户", email="[email protected]")
        session.add(user)
        savepoint.commit()
    except:
        savepoint.rollback()

# 使用上下文管理器管理会话
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="[email protected]")
    db.add(user)

Python SQLAlchemy ORM 数据库操作指南

安装

核心概念

连接数据库

定义数据模型

创建数据库表

基本 CRUD 操作

创建数据

读取数据

更新数据

删除数据

查询数据

基本查询

过滤查询

聚合查询

连接查询

关系操作

事务管理

最佳实践

总结

更多推荐文章

相关免费在线工具

Python SQLAlchemy ORM 数据库操作指南

安装

核心概念

连接数据库

定义数据模型

创建数据库表

基本 CRUD 操作

创建数据

读取数据

更新数据

删除数据

查询数据

基本查询

过滤查询

聚合查询

连接查询

关系操作

事务管理

最佳实践

总结

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具