跳到主要内容Python SQLAlchemy ORM 数据库操作指南 | 极客日志Python
Python SQLAlchemy ORM 数据库操作指南
Python 中流行的 ORM 框架 SQLAlchemy 的使用方法,涵盖安装配置、核心概念(Engine、Session、Model)、数据库连接、数据模型定义及表创建。内容还包括基本 CRUD 操作、复杂查询(过滤、聚合、连接)、关系操作(一对多、多对多)以及事务管理。最后总结了会话管理、异常处理等最佳实践,帮助开发者高效进行数据库交互。
CodeArtist1 浏览 SQLAlchemy是Python中最流行的ORM(对象关系映射)框架之一,它提供了高效且灵活的数据库操作方式。本文将介绍如何使用SQLAlchemy ORM进行数据库操作。
安装
pip install sqlalchemy
如果需要连接特定数据库,还需安装相应的驱动程序:
pip install psycopg2-binary
pip install mysql-connector-python
核心概念
- Engine:数据库连接的引擎,负责与数据库通信
- Session:数据库会话,管理所有持久化操作
- Model:数据模型类,对应数据库中的表
- Query:查询对象,用于构建和执行数据库查询
连接数据库
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()
定义数据模型
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)
基本 CRUD 操作
创建数据
读取数据
users = session.query(User).all()
first_user = session.query(User).first()
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()
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()
最佳实践
- 会话管理:为每个请求创建新会话,请求结束后关闭
- 异常处理:始终处理异常并适当回滚事务
- 延迟加载:注意 N+1 查询问题,使用 eager loading 优化
- 连接池:合理配置连接池大小和超时设置
- 数据验证:在模型层或应用层验证数据完整性
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
- 定义数据模型和关系
- 执行基本的 CRUD 操作
- 构建复杂查询
- 管理数据库事务
- 遵循最佳实践
SQLAlchemy 还有更多高级特性,如混合属性、事件监听、自定义查询等,值得进一步探索学习。
微信扫一扫,关注极客日志
微信公众号「极客日志」,在微信中扫描左侧二维码关注。展示文案:极客日志 zeeklog
相关免费在线工具
- curl 转代码
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
- Base64 字符串编码/解码
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
- Base64 文件转换器
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online
- Markdown 转 HTML
将 Markdown(GFM)转为 HTML 片段,浏览器内 marked 解析;与 HTML 转 Markdown 互为补充。 在线工具,Markdown 转 HTML在线工具,online
- HTML 转 Markdown
将 HTML 片段转为 GitHub Flavored Markdown,支持标题、列表、链接、代码块与表格等;浏览器内处理,可链接预填。 在线工具,HTML 转 Markdown在线工具,online
- JSON 压缩
通过删除不必要的空白来缩小和压缩JSON。 在线工具,JSON 压缩在线工具,online