老程序员博客

编程生涯的最后一舞

当前位置：首页 > 技术 > 正文内容

Python SQLAlchemy ORM 数据库操作指南

访客 技术 2026年6月3日 1

SQLAlchemy是Python生态系统中备受推崇的对象关系映射(ORM)框架，它为开发者提供了与数据库交互的高级抽象层。本指南将深入探讨如何利用SQLAlchemy ORM实现高效的数据库操作。

目录

  1. 环境准备与安装
  2. 核心组件解析
  3. 数据库连接配置
  4. 数据模型设计
  5. 数据库表结构生成
  6. 基本数据操作
  7. 高级查询技巧
  8. 关系型数据处理
  9. 事务控制机制
  10. 开发最佳实践

环境准备与安装

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示例
db_engine = create_engine('sqlite:///database.db', echo=True)

# PostgreSQL示例
# db_engine = create_engine('postgresql://user:password@localhost:5432/mydb')

# MySQL示例
# db_engine = create_engine('mysql+mysqlconnector://user:password@localhost:3306/mydb')

# 创建会话工厂
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=db_engine)

# 实例化会话对象
current_session = SessionFactory()

数据模型设计

python

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, declarative_base

# 定义基础模型类
BaseModel = declarative_base()

class Person(BaseModel):
    __tablename__ = 'persons'
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, index=True)
    
    # 定义一对多关系
    articles = relationship("Article", back_populates="creator")
    
class Article(BaseModel):
    __tablename__ = 'articles'
    
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(100), nullable=False)
    content = Column(String(500))
    creator_id = Column(Integer, ForeignKey('persons.id'))
    
    # 定义多对一关系
    creator = relationship("Person", back_populates="articles")
    
    # 定义多对多关系（通过关联表）
    categories = relationship("Category", secondary="article_categories", back_populates="articles")

class Category(BaseModel):
    __tablename__ = 'categories'
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(30), unique=True, nullable=False)
    
    articles = relationship("Article", secondary="article_categories", back_populates="categories")

# 关联表（用于多对多关系）
class ArticleCategory(BaseModel):
    __tablename__ = 'article_categories'
    
    article_id = Column(Integer, ForeignKey('articles.id'), primary_key=True)
    category_id = Column(Integer, ForeignKey('categories.id'), primary_key=True)

数据库表结构生成

python

# 创建所有定义的表
BaseModel.metadata.create_all(bind=db_engine)

# 删除所有表
# BaseModel.metadata.drop_all(bind=db_engine)

基本数据操作

创建数据

python

# 创建新用户
new_person = Person(username="张明", email="zhangming@example.com")
current_session.add(new_person)
current_session.commit()

# 批量创建
current_session.add_all([
    Person(username="李华", email="lihua@example.com"),
    Person(username="王芳", email="wangfang@example.com")
])
current_session.commit()

读取数据

python

# 获取所有用户
all_persons = current_session.query(Person).all()

# 获取第一个用户
first_person = current_session.query(Person).first()

# 根据ID获取用户
specific_person = current_session.query(Person).get(1)

更新数据

python

# 查询并更新
person = current_session.query(Person).get(1)
person.username = "张明新"
current_session.commit()

# 批量更新
current_session.query(Person).filter(Person.username.like("张%")).update({"username": "张氏"}, synchronize_session=False)
current_session.commit()

删除数据

python

# 查询并删除
person = current_session.query(Person).get(1)
current_session.delete(person)
current_session.commit()

# 批量删除
current_session.query(Person).filter(Person.username == "李华").delete(synchronize_session=False)
current_session.commit()

高级查询技巧

基本查询

python

# 获取所有记录
all_persons = current_session.query(Person).all()

# 获取特定字段
usernames = current_session.query(Person.username).all()

# 排序
persons_desc = current_session.query(Person).order_by(Person.username.desc()).all()

# 限制结果数量
limited_persons = current_session.query(Person).limit(10).all()

# 分页查询
paginated_persons = current_session.query(Person).offset(5).limit(10).all()

过滤查询

python

from sqlalchemy import or_

# 等值过滤
person = current_session.query(Person).filter(Person.username == "张明").first()

# 模糊查询
persons_like = current_session.query(Person).filter(Person.username.like("张%")).all()

# IN查询
persons_in = current_session.query(Person).filter(Person.username.in_(["张明", "李华"])).all()

# 多条件查询
persons_multi = current_session.query(Person).filter(
    Person.username == "张明", 
    Person.email.like("%@example.com")
).all()

# 或条件
persons_or = current_session.query(Person).filter(
    or_(Person.username == "张明", Person.username == "李华")
).all()

# 不等于
persons_ne = current_session.query(Person).filter(Person.username != "张明").all()

聚合查询

python

from sqlalchemy import func

# 计数
total_count = current_session.query(Person).count()

# 分组计数
article_counts = current_session.query(
    Person.username, 
    func.count(Article.id)
).join(Article).group_by(Person.username).all()

# 统计计算
avg_id = current_session.query(func.avg(Person.id)).scalar()

连接查询

python

# 内连接
results = current_session.query(Person, Article).join(Article).filter(Article.title.like("%Python%")).all()

# 左外连接
results = current_session.query(Person, Article).outerjoin(Article).all()

# 指定连接条件
results = current_session.query(Person, Article).join(Article, Person.id == Article.creator_id).all()

关系型数据处理

python

# 创建带关系的对象
new_person = Person(username="赵伟", email="zhaowei@example.com")
new_article = Article(title="Python入门指南", content="学习Python编程的基础知识", creator=new_person)
current_session.add(new_article)
current_session.commit()

# 通过关系访问
print(f"文章 '{new_article.title}' 的作者是 {new_article.creator.username}")
print(f"用户 {new_person.username} 的所有文章:")
for article in new_person.articles:
    print(f"  - {article.title}")

# 多对多关系操作
tech_category = Category(name="技术")
python_category = Category(name="Python")

new_article.categories.append(tech_category)
new_article.categories.append(python_category)
current_session.commit()

print(f"文章 '{new_article.title}' 的分类:")
for category in new_article.categories:
    print(f"  - {category.name}")

事务控制机制

python

# 手动事务控制
try:
    person = Person(username="事务测试用户", email="transaction_test@example.com")
    current_session.add(person)
    current_session.commit()
except Exception as e:
    current_session.rollback()
    print(f"发生错误: {e}")

# 使用事务上下文管理器
from sqlalchemy.orm import Session

def create_person(session: Session, username: str, email: str):
    try:
        person = Person(username=username, email=email)
        session.add(person)
        session.commit()
        return person
    except:
        session.rollback()
        raise

# 嵌套事务
with current_session.begin_nested():
    person = Person(username="嵌套事务用户", email="nested@example.com")
    current_session.add(person)

# 保存点
savepoint = current_session.begin_nested()
try:
    person = Person(username="保存点测试用户", email="savepoint_test@example.com")
    current_session.add(person)
    savepoint.commit()
except:
    savepoint.rollback()

开发最佳实践

  1. 会话生命周期管理：为每个请求创建独立的会话，请求结束后立即关闭
  2. 异常处理机制：始终捕获异常并适当回滚事务
  3. 查询优化：注意N+1查询问题，使用预加载(eager loading)提升性能
  4. 连接池配置：合理设置连接池大小和超时参数
  5. 数据验证：在模型层或应用层实施数据完整性校验

python

# 使用上下文管理器管理会话
from contextlib import contextmanager

@contextmanager
def get_database_session():
    db_session = SessionFactory()
    try:
        yield db_session
        db_session.commit()
    except Exception:
        db_session.rollback()
        raise
    finally:
        db_session.close()

# 使用示例
with get_database_session() as db:
    person = Person(username="上下文用户", email="context_user@example.com")
    db.add(person)

总结

SQLAlchemy ORM为Python开发者提供了强大而灵活的数据库操作能力，通过本指南的介绍，您应该能够：

  1. 正确安装和配置SQLAlchemy环境
  2. 设计并实现数据模型及其关系
  3. 执行完整的CRUD操作
  4. 构建复杂高效的查询语句
  5. 管理数据库事务确保数据一致性
  6. 遵循行业最佳实践提升应用质量

SQLAlchemy还包含众多高级特性，如混合属性、事件监听、自定义查询扩展等，值得进一步深入学习探索。

标签: SQLAlchemy
返回列表

上一篇：Maven 中 Jetty 与 Tomcat 插件的配置与使用

下一篇：如何将文本输入解析为可交互的语法树结构

相关文章

Linux crontab 详解

1) crontab 是什么cron 是 Linux 的定时任务守护进程；crontab 是用来编辑/查看“按时间周期执行命令”的表（cron table）。常见两类：用户 crontab：每个用户一份（crontab -e 编辑）系统级 crontab / cron.d：可指定执行用户（/etc/crontab、/etc/cron.d/*）2) crontab 时间...

富文本里可以允许的 HTML 属性

一、所有标签默认允许的安全属性（极少）class        （可选）id           （通常建议禁用）title️ 注意：id 容易被滥用做锚点注入，很多系统直接禁用class 允许的话最好只允许固定前缀（如 editor-*）二、a 标签允许属性<a href="" t...

Mac 安装 Node.js 指南

方法一：通过官网安装包（最简单，适合初学者）如果你只是想快速安装并开始使用，这是最直接的方法。访问 Node.js 官网。页面会显示两个版本：LTS (Recommended For Most Users)：长期支持版，最稳定。建议选这个。Current：最新特性版，包含最新功能但可能不够稳定。下载 .pkg 安装包并运行。按照安装向导点击“下一步”即可完成。方法二：使用 Homebrew 安装（...

Dom\HTML_NO_DEFAULT_NS 的副作用：自动加闭合标签

在使用Dom\HTMLDocument时，Dom\HTML_NO_DEFAULT_NS 将禁止在解析过程中设置元素的命名空间, 此设置是为了与DOMDocument向后兼容而存在的。当使用它时，已知的一个副作用就是：自动加闭合标签例如 </img> 为什么会这样？当你使用：Dom\HTML_NO_DEFAULT_NS文档会变成 无命名空间模式，此时内部更接近 XML...

Laravel 事件和监听器创建

在 Laravel 中，使用 Artisan 命令创建 Events（事件） 和 Listeners（监听器） 是非常高效的。你可以通过以下几种方式来实现：1. 手动创建单个 Event如果你只想创建一个事件类，可以使用 make:event 命令：Bashphp artisan make:event UserRegistered执行后，文件将生成在 app/Even...

自定义域名解析神器 dnsmasq

什么是 dnsmasq？dnsmasq 是一个轻量级、功能强大的网络服务工具，专为小型和中等规模网络设计。它是一个综合的网络基础设施解决方案[1]。dnsmasq 能做什么？功能说明应用场景DNS 转发与缓存将 DNS 查询转发到上游服务器（ISP、Google DNS 等），并在本地缓存结果加快 DNS 查询速度，减少外部 DNS 流量本地 DNS解析本地网络设备的主机名，无需编辑&n...

发表评论

访客

◎欢迎参与讨论，请在这里发表您的看法和观点。

Copyright © agingcoder.cn

Powered By Z-BlogPHP. Theme by TOYEAN.