Python中的ORM
原创
Python中的ORM
原创
软件架构师Michael
发布于 2025-06-03 11:24:19
发布于 2025-06-03 11:24:19
代码可运行
运行总次数：0
代码可运行
在现代Web开发中，对象关系映射（ORM）技术已成为连接应用程序与数据库的关键桥梁。Python生态提供了多种优秀的ORM框架，其中SQLAlchemy以其强大的功能和对多种数据库的支持脱颖而出。本文将深入探讨使用SQLAlchemy操作SQL Server、MySQL和PostgreSQL三大主流数据库的技术实践。
一、ORM核心优势与SQLAlchemy简介
ORM的核心价值在于：
- 将数据库表映射为Python类
- 将表字段映射为类属性
- 将数据库操作转化为面向对象方法
- 提供跨数据库的兼容性
SQLAlchemy作为Python最全面的ORM工具，包含两大核心组件：
- Core：提供SQL表达式语言和数据库连接
- ORM：高级对象映射层
# 安装SQLAlchemy及数据库驱动
pip install sqlalchemy
pip install pymysql # MySQL
pip install psycopg2 # PostgreSQL
pip install pyodbc # SQL Server二、跨数据库连接配置
不同数据库的连接配置差异主要体现在连接字符串上：
from sqlalchemy import create_engine
# MySQL连接配置
mysql_engine = create_engine(
"mysql+pymysql://user:password@localhost/mydb?charset=utf8mb4"
)
# PostgreSQL连接配置
postgres_engine = create_engine(
"postgresql+psycopg2://user:password@localhost/mydb"
)
# SQL Server连接配置
mssql_engine = create_engine(
"mssql+pyodbc://user:password@localhost/mydb?driver=ODBC+Driver+17+for+SQL+Server"
)三、统一数据模型定义
SQLAlchemy的数据模型在不同数据库间保持高度一致性：
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(50), nullable=False)
email = Column(String(100), unique=True)
created_at = Column(DateTime, server_default='CURRENT_TIMESTAMP')
def __repr__(self):
return f"<User(name='{self.name}', email='{self.email}')>"四、数据库操作实例
1. 创建表结构
def create_tables(engine):
Base.metadata.create_all(engine)
# 为所有数据库创建表
create_tables(mysql_engine)
create_tables(postgres_engine)
create_tables(mssql_engine)2. 增删改查操作
# 创建会话工厂
Session = sessionmaker(bind=mysql_engine) # 可替换为其他引擎
def crud_operations():
with Session() as session:
# 创建记录
new_user = User(name="张三", email="zhangsan@example.com")
session.add(new_user)
session.commit()
# 查询记录
user = session.query(User).filter_by(name="张三").first()
print(f"查询结果: {user}")
# 更新记录
user.email = "new_email@example.com"
session.commit()
# 删除记录
session.delete(user)
session.commit()3. 事务管理
def transaction_demo():
try:
with Session() as session:
user1 = User(name="李四", email="lisi@example.com")
session.add(user1)
# 触发唯一约束异常
user2 = User(name="王五", email="lisi@example.com")
session.add(user2)
session.commit()
except Exception as e:
print(f"事务失败: {str(e)}")
session.rollback()五、数据库特定功能处理
虽然SQLAlchemy提供统一接口，但处理数据库差异时需注意：
1. 自增字段处理
# SQL Server需要指定IDENTITY属性
if engine.dialect.name == 'mssql':
id = Column(Integer, primary_key=True, autoincrement=True)
else:
id = Column(Integer, primary_key=True)2. 分页查询差异
# 标准分页查询
users = session.query(User).order_by(User.id).offset(10).limit(5).all()
# SQL Server需要特殊处理
if engine.dialect.name == 'mssql':
users = session.query(User).order_by(User.id).offset(10).fetch(5).all()3. JSON字段支持
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.types import JSON
# PostgreSQL
if engine.dialect.name == 'postgresql':
metadata = Column(JSONB)
# MySQL 5.7+
elif engine.dialect.name == 'mysql':
metadata = Column(JSON)
# SQL Server 2016+
elif engine.dialect.name == 'mssql':
metadata = Column(JSON)六、高级查询技巧
1. 复杂条件查询
# 多条件组合
from sqlalchemy import or_
results = session.query(User).filter(
or_(
User.name.like('张%'),
User.email.contains('example')
)
).order_by(User.created_at.desc()).all()2. 聚合查询
from sqlalchemy import func
# 按域名分组统计
email_domain = func.substring_index(User.email, '@', -1).label('domain')
result = session.query(
email_domain,
func.count(User.id)
).group_by(email_domain).all()3. 关系映射（1对多）
class Address(Base):
__tablename__ = 'addresses'
id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('users.id'))
street = Column(String(100))
user = relationship("User", back_populates="addresses")
User.addresses = relationship("Address", order_by=Address.id, back_populates="user")七、性能优化策略
- 批量操作
# 批量插入
session.bulk_save_objects([
User(name=f"user{i}", email=f"user{i}@domain.com")
for i in range(1000)
])
session.commit()
engine = create_engine(
"postgresql+psycopg2://user:pass@host/db",
pool_size=10,
max_overflow=5,
pool_timeout=30
)2.查询优化
# 使用selectinload避免N+1查询
from sqlalchemy.orm import selectinload
users = session.query(User).options(selectinload(User.addresses)).all()原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
评论
登录后参与评论
推荐阅读
目录
腾讯云开发者
