ORM与SQLAlchemy的重要性
ORM（对象关系映射）是连接面向对象编程和关系数据库的桥梁，SQLAlchemy作为Python最强大的ORM框架，提供了灵活的数据建模和查询能力，是Python数据库开发的核心工具。
SQLAlchemy基础
1. 模型定义
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy import create_engine
from datetime import datetime
Base = declarative_base()
class User(Base):
"""用户模型"""
__tablename__ = 'users'
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(50), nullable=False)
email = Column(String(100), unique=True, nullable=False)
created_at = Column(DateTime, default=datetime.now)
updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
# 关系映射
posts = relationship("Post", back_populates="author")
def __repr__(self):
return f"<User(id={self.id}, name='{self.name}', email='{self.email}')>"
class Post(Base):
"""文章模型"""
__tablename__ = 'posts'
id = Column(Integer, primary_key=True, autoincrement=True)
title = Column(String(200), nullable=False)
content = Column(Text)
author_id = Column(Integer, ForeignKey('users.id'), nullable=False)
created_at = Column(DateTime, default=datetime.now)
# 关系映射
author = relationship("User", back_populates="posts")
def __repr__(self):
return f"<Post(id={self.id}, title='{self.title}', author_id={self.author_id})>"
# 创建数据库引擎
engine = create_engine('sqlite:///blog.db', echo=True)
Base.metadata.create_all(engine)
2. 会话管理
from sqlalchemy.orm import sessionmaker
from contextlib import contextmanager
# 创建会话工厂
Session = sessionmaker(bind=engine)
@contextmanager
def get_session():
"""数据库会话上下文管理器"""
session = Session()
try:
yield session
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
def create_tables():
"""创建数据库表"""
Base.metadata.create_all(engine)
print("数据库表创建成功")
create_tables()
查询操作
1. 基本查询
def basic_queries():
"""基本查询操作"""
print("=== 基本查询操作 ===")
with get_session() as session:
# 创建测试数据
user1 = User(name='张三', email='zhangsan@example.com')
user2 = User(name='李四', email='lisi@example.com')
session.add_all([user1, user2])
post1 = Post(title='Python基础教程', content='Python是一门优秀的编程语言', author_id=1)
post2 = Post(title='SQLAlchemy进阶', content='SQLAlchemy是Python最强大的ORM框架', author_id=1)
session.add_all([post1, post2])
with get_session() as session:
# 查询所有用户
all_users = session.query(User).all()
print(f"所有用户: {all_users}")
# 根据ID查询
user = session.query(User).filter(User.id == 1).first()
print(f"ID为1的用户: {user}")
# 条件查询
users_with_zhang = session.query(User).filter(User.name.like('%张%')).all()
print(f"姓名包含'张'的用户: {users_with_zhang}")
# 排序查询
users_ordered = session.query(User).order_by(User.created_at.desc()).all()
print(f"按创建时间倒序的用户: {users_ordered}")
# 限制查询数量
first_user = session.query(User).first()
print(f"第一个用户: {first_user}")
# 统计查询
user_count = session.query(User).count()
print(f"用户总数: {user_count}")
basic_queries()
2. 高级查询
def advanced_queries():
"""高级查询操作"""
print("=== 高级查询操作 ===")
with get_session() as session:
# 多条件查询
users = session.query(User).filter(
User.name.like('%张%'),
User.email.like('%@example.com')
).all()
print(f"多条件查询结果: {users}")
# 范围查询
recent_users = session.query(User).filter(
User.created_at >= datetime(2018, 12, 1)
).all()
print(f"2018年12月后创建的用户: {recent_users}")
# 分组查询
from sqlalchemy import func
post_counts = session.query(
User.name,
func.count(Post.id).label('post_count')
).join(Post).group_by(User.id).all()
print(f"用户文章数量统计: {post_counts}")
# 子查询
subquery = session.query(Post.author_id).filter(Post.title.like('%Python%')).subquery()
users_with_python_posts = session.query(User).filter(User.id.in_(subquery)).all()
print(f"发表过Python相关文章的用户: {users_with_python_posts}")
# 联合查询
from sqlalchemy import or_
users_or_posts = session.query(User).filter(
or_(User.name.like('%张%'), User.email.like('%@gmail.com'))
).all()
print(f"姓名包含'张'或邮箱为gmail的用户: {users_or_posts}")
advanced_queries()
关系映射
1. 一对多关系
def one_to_many_relationships():
"""一对多关系操作"""
print("=== 一对多关系操作 ===")
with get_session() as session:
# 创建用户和文章
user = User(name='王五', email='wangwu@example.com')
session.add(user)
session.flush() # 获取用户ID
post1 = Post(title='Flask入门', content='Flask是轻量级Web框架', author_id=user.id)
post2 = Post(title='Django进阶', content='Django是功能完整的Web框架', author_id=user.id)
session.add_all([post1, post2])
with get_session() as session:
# 通过用户查询文章
user = session.query(User).filter(User.name == '王五').first()
if user:
print(f"用户 {user.name} 的文章:")
for post in user.posts:
print(f" - {post.title}")
# 通过文章查询作者
post = session.query(Post).filter(Post.title == 'Flask入门').first()
if post:
print(f"文章 '{post.title}' 的作者: {post.author.name}")
# 使用join查询
posts_with_authors = session.query(Post, User).join(User).all()
print("文章和作者信息:")
for post, author in posts_with_authors:
print(f" {post.title} - {author.name}")
one_to_many_relationships()
实际应用案例
1. 博客系统
def blog_system_example():
"""博客系统示例"""
print("=== 博客系统示例 ===")
with get_session() as session:
# 创建博客数据
author = User(name='技术博主', email='blogger@example.com')
session.add(author)
session.flush()
# 创建文章
posts_data = [
{'title': 'Python基础语法', 'content': 'Python是一门简单易学的编程语言'},
{'title': 'Flask Web框架', 'content': 'Flask是Python最流行的Web框架之一'},
{'title': 'SQLAlchemy ORM', 'content': 'SQLAlchemy是Python最强大的ORM框架'}
]
for post_data in posts_data:
post = Post(
title=post_data['title'],
content=post_data['content'],
author_id=author.id
)
session.add(post)
with get_session() as session:
# 查询博客统计信息
from sqlalchemy import func
# 作者文章数量
author_stats = session.query(
User.name,
func.count(Post.id).label('post_count'),
func.max(Post.created_at).label('latest_post')
).join(Post).group_by(User.id).all()
print("作者统计信息:")
for name, count, latest in author_stats:
print(f" {name}: {count}篇文章, 最新文章: {latest}")
blog_system_example()
总结
掌握SQLAlchemy ORM是Python数据库开发的核心：
- 模型定义：理解表结构和关系映射
- 查询操作：掌握基本和高级查询技术
- 关系映射：学会一对一、一对多、多对多关系
- 实际应用：在博客系统、数据迁移等场景中的应用
- 最佳实践：遵循ORM设计的最佳实践
- 性能优化：学会查询优化和缓存策略
通过系统学习这些概念，你将能够构建出高效、可维护的数据库应用。
转载请注明：周志洋的博客 » Python实用技巧-ORM与SQLAlchemy
