Python ORM框架深度解析：SQLAlchemy实战

第一程序员

138人浏览 · 2026-05-08 20:01:09

第一程序员 · 2026-05-08 20:01:09 发布

Python ORM框架深度解析：SQLAlchemy实战

引言

大家好，我是一名正在从Rust转向Python的后端开发者。在数据库操作方面，ORM（Object-Relational Mapping）是一个非常重要的工具。它可以帮助我们将数据库表映射到Python对象，从而避免直接编写SQL语句。作为从Rust过来的开发者，我发现Python的SQLAlchemy是一个功能非常强大的ORM框架。今天，我想和大家分享一下我在使用SQLAlchemy方面的一些经验。

ORM概述

什么是ORM？

ORM是一种编程技术，用于在关系型数据库和面向对象编程语言之间建立映射关系。它允许我们使用面向对象的方式来操作数据库，而不需要直接编写SQL语句。

ORM的优点

代码可读性：使用面向对象的方式操作数据库
类型安全：Python的类型系统可以提供一定的类型安全
数据库无关性：可以轻松切换数据库后端
减少重复代码：ORM会自动生成SQL语句

SQLAlchemy架构

应用层
  ↓
SQLAlchemy Core（SQL表达式语言）
  ↓
SQLAlchemy ORM（对象关系映射）
  ↓
数据库引擎（SQLite、MySQL、PostgreSQL等）

SQLAlchemy基础

安装

pip install sqlalchemy

连接数据库

from sqlalchemy import create_engine

# SQLite
engine = create_engine('sqlite:///example.db')

# MySQL
engine = create_engine('mysql+pymysql://user:password@localhost/dbname')

# PostgreSQL
engine = create_engine('postgresql://user:password@localhost/dbname')

定义模型

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100), unique=True)
    age = Column(Integer)
    
    def __repr__(self):
        return f"<User(name='{self.name}', email='{self.email}')>"

创建表

# 创建所有表
Base.metadata.create_all(engine)

会话管理

创建会话

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)
session = Session()

添加数据

# 添加单条记录
user = User(name='John', email='john@example.com', age=30)
session.add(user)
session.commit()

# 添加多条记录
users = [
    User(name='Alice', email='alice@example.com', age=25),
    User(name='Bob', email='bob@example.com', age=35)
]
session.add_all(users)
session.commit()

查询数据

# 查询所有
users = session.query(User).all()

# 条件查询
user = session.query(User).filter_by(name='John').first()

# 多条件查询
users = session.query(User).filter(User.age > 25, User.name.like('%o%')).all()

# 排序
users = session.query(User).order_by(User.age.desc()).all()

# 限制结果数量
users = session.query(User).limit(10).all()

更新数据

user = session.query(User).filter_by(name='John').first()
user.age = 31
session.commit()

删除数据

user = session.query(User).filter_by(name='John').first()
session.delete(user)
session.commit()

高级查询

连接查询

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class Address(Base):
    __tablename__ = 'addresses'
    
    id = Column(Integer, primary_key=True)
    email = Column(String(100))
    user_id = Column(Integer, ForeignKey('users.id'))
    
    user = relationship("User", back_populates="addresses")

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    addresses = relationship("Address", back_populates="user")

# 查询用户及其地址
users_with_addresses = session.query(User).join(Address).all()

聚合查询

from sqlalchemy import func

# 统计用户数量
count = session.query(func.count(User.id)).scalar()

# 计算平均年龄
avg_age = session.query(func.avg(User.age)).scalar()

# 分组统计
result = session.query(User.name, func.count(Address.id)).join(Address).group_by(User.id).all()

子查询

subquery = session.query(User.id).filter(User.age > 30).subquery()
users = session.query(User).filter(User.id.in_(subquery)).all()

关系映射

一对一关系

class Profile(Base):
    __tablename__ = 'profiles'
    
    id = Column(Integer, primary_key=True)
    bio = Column(String(500))
    user_id = Column(Integer, ForeignKey('users.id'))
    
    user = relationship("User", back_populates="profile", uselist=False)

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    profile = relationship("Profile", back_populates="user", uselist=False)

一对多关系

class Post(Base):
    __tablename__ = 'posts'
    
    id = Column(Integer, primary_key=True)
    title = Column(String(100))
    content = Column(String(1000))
    author_id = Column(Integer, ForeignKey('users.id'))
    
    author = relationship("User", back_populates="posts")

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    posts = relationship("Post", back_populates="author")

多对多关系

from sqlalchemy import Table

# 关联表
user_role = Table(
    'user_role', Base.metadata,
    Column('user_id', Integer, ForeignKey('users.id')),
    Column('role_id', Integer, ForeignKey('roles.id'))
)

class Role(Base):
    __tablename__ = 'roles'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    
    users = relationship("User", secondary=user_role, back_populates="roles")

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    roles = relationship("Role", secondary=user_role, back_populates="users")

事务管理

基本事务

try:
    # 开始事务
    user1 = User(name='Alice', email='alice@example.com')
    user2 = User(name='Bob', email='bob@example.com')
    
    session.add(user1)
    session.add(user2)
    
    # 提交事务
    session.commit()
except Exception as e:
    # 回滚事务
    session.rollback()
    raise e

嵌套事务

# 保存点
session.begin_nested()
try:
    user = User(name='Test', email='test@example.com')
    session.add(user)
    session.commit()
except Exception as e:
    session.rollback()

性能优化

查询优化

# 延迟加载（默认）
users = session.query(User).all()
for user in users:
    # 每次访问addresses都会触发额外查询
    print(user.addresses)

# 预先加载
users = session.query(User).options(joinedload(User.addresses)).all()
for user in users:
    # 不会触发额外查询
    print(user.addresses)

# 选择加载特定字段
users = session.query(User.name, User.email).all()

批量操作

from sqlalchemy import insert, update, delete

# 批量插入
stmt = insert(User).values([
    {'name': 'Alice', 'email': 'alice@example.com'},
    {'name': 'Bob', 'email': 'bob@example.com'}
])
session.execute(stmt)
session.commit()

# 批量更新
stmt = update(User).where(User.age < 30).values(age=30)
session.execute(stmt)
session.commit()

# 批量删除
stmt = delete(User).where(User.name.like('%Test%'))
session.execute(stmt)
session.commit()

实战项目：完整的用户管理系统

from sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from datetime import datetime

# 创建引擎
engine = create_engine('sqlite:///user_management.db')
Base = declarative_base()

# 定义模型
class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    password_hash = Column(String(255), nullable=False)
    created_at = Column(DateTime, default=datetime.now)
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
    
    profile = relationship("UserProfile", back_populates="user", uselist=False)
    posts = relationship("Post", back_populates="author")
    
    def __repr__(self):
        return f"<User(username='{self.username}', email='{self.email}')>"

class UserProfile(Base):
    __tablename__ = 'user_profiles'
    
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    first_name = Column(String(50))
    last_name = Column(String(50))
    bio = Column(String(500))
    location = Column(String(100))
    
    user = relationship("User", back_populates="profile")

class Post(Base):
    __tablename__ = 'posts'
    
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(String(5000), nullable=False)
    author_id = Column(Integer, ForeignKey('users.id'))
    created_at = Column(DateTime, default=datetime.now)
    
    author = relationship("User", back_populates="posts")

# 创建表
Base.metadata.create_all(engine)

# 创建会话
Session = sessionmaker(bind=engine)
session = Session()

# 用户服务类
class UserService:
    def __init__(self, session):
        self.session = session
    
    def create_user(self, username, email, password):
        # 检查用户是否已存在
        existing_user = self.session.query(User).filter(
            (User.username == username) | (User.email == email)
        ).first()
        
        if existing_user:
            raise ValueError("用户名或邮箱已存在")
        
        # 创建用户
        user = User(
            username=username,
            email=email,
            password_hash=password  # 实际应用中应该使用bcrypt等加密
        )
        
        # 创建用户资料
        profile = UserProfile(user=user)
        
        self.session.add(user)
        self.session.add(profile)
        self.session.commit()
        
        return user
    
    def get_user_by_id(self, user_id):
        return self.session.query(User).filter_by(id=user_id).first()
    
    def get_user_by_email(self, email):
        return self.session.query(User).filter_by(email=email).first()
    
    def update_user_profile(self, user_id, **kwargs):
        user = self.get_user_by_id(user_id)
        if not user:
            raise ValueError("用户不存在")
        
        if user.profile:
            for key, value in kwargs.items():
                setattr(user.profile, key, value)
        else:
            user.profile = UserProfile(user=user, **kwargs)
        
        self.session.commit()
        return user
    
    def delete_user(self, user_id):
        user = self.get_user_by_id(user_id)
        if not user:
            raise ValueError("用户不存在")
        
        self.session.delete(user)
        self.session.commit()

# 使用示例
if __name__ == '__main__':
    service = UserService(session)
    
    # 创建用户
    user = service.create_user('john_doe', 'john@example.com', 'password123')
    print(f"创建用户: {user}")
    
    # 更新用户资料
    user = service.update_user_profile(user.id, first_name='John', last_name='Doe', bio='Hello World')
    print(f"更新资料: {user.profile.first_name} {user.profile.last_name}")
    
    # 查询用户
    found_user = service.get_user_by_email('john@example.com')
    print(f"查询用户: {found_user}")
    
    # 删除用户
    service.delete_user(user.id)
    print("用户已删除")

与Rust ORM的对比

特性	SQLAlchemy	Diesel (Rust)
类型安全	通过类型提示	编译时保证
查询构建	链式API	宏和类型系统
迁移工具	第三方工具	内置迁移支持
异步支持	SQLAlchemy 2.0+	原生支持
性能	较好	更好

总结

SQLAlchemy是一个功能强大的ORM框架，它提供了：

灵活的查询API：支持复杂的查询操作
强大的关系映射：支持各种关系类型
事务管理：完整的事务支持
性能优化：支持预先加载、批量操作等

通过合理使用SQLAlchemy，我们可以：

提高代码的可读性和可维护性
减少SQL语句的编写
实现数据库无关性

作为从Rust转向Python的开发者，我发现SQLAlchemy虽然没有Rust的Diesel那样的编译时类型安全，但它的灵活性和丰富功能使其成为Python生态中最受欢迎的ORM框架之一。

延伸阅读：

亚马逊云科技技术品牌专区

更多推荐

53.1.智能投喂器-硬件定时-基于STM32嵌入式物联网单片机软硬件毕业生系统设计【硬件+APP+云平台】

亚马逊云科技技术品牌专区

CMU 11-785 深度学习导论笔记（一）

神经网络是人工智能中的一种方法，它教会计算机以受人类大脑启发的方式处理数据。近年来，它已成为各种模式识别、预测和分析问题的主要研究方向之一。神经网络在许多问题上确立了最先进的技术水平，并且常常大幅超越之前的基准。上一节我们介绍了神经网络的基本定义，本节中我们来看看神经网络带来的一些突破性应用。语音助手：例如 Siri、Alexa、Google Assistant。视觉与感知：例如人脸检测、人脸识别

亚马逊云科技技术品牌专区

分布式ID的UUID与自定义时钟

第二，由于ID整体随时间戳递增，数据在存储时具有天然的时间局部性，极大提升了数据库的写入性能与范围查询效率。这种“等待”机制，结合工作节点ID的空间划分，确保了跨节点、跨时间的ID全局唯一且严格递增。反之，如果面对的是海量数据、高并发写入的场景，如电商交易、实时监控、社交网络动态等，那么投入精力构建基于自定义时钟的分布式ID服务，将是保障系统长期稳定与高效运行的关键基础设施投资。在云原生与微服务架