SQLAlchemy ORM 框架的使用，涵盖安装配置、核心概念（Engine、Session、Model）、数据库连接、数据模型定义及表创建。内容包含基本 CRUD 操作、复杂查询（过滤、聚合、连接）、关系操作（一对多、多对多）以及事务管理。最后提供了会话管理、异常处理和连接池等最佳实践建议，帮助开发者掌握 Python 数据库交互的核心技能。
SQLAlchemy 是 Python 中最流行的 ORM（对象关系映射）框架之一，它提供了高效且灵活的数据库操作方式。本文将介绍如何使用 SQLAlchemy ORM 进行数据库操作。
bash
pip install sqlalchemy
如果需要连接特定数据库，还需安装相应的驱动程序：
bash
python
from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker
engine = create_engine('sqlite:///example.db', echo=True)
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)
python
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()
python
users = session.query(User).all()
first_user = session.query(User).first()
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()
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="[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}")
python
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()
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="[email protected]") db.add(user)
SQLAlchemy ORM 提供了强大而灵活的数据库操作方式，通过本文的介绍，您应该能够：
SQLAlchemy 还有更多高级特性，如混合属性、事件监听、自定义查询等，值得进一步探索学习。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online
将 HTML 片段转为 GitHub Flavored Markdown，支持标题、列表、链接、代码块与表格等；浏览器内处理，可链接预填。 在线工具，HTML转Markdown在线工具，online
通过删除不必要的空白来缩小和压缩JSON。 在线工具，JSON 压缩在线工具，online
