
在现代Web开发中,对象关系映射(ORM)技术已成为连接应用程序与数据库的关键桥梁。Python生态提供了多种优秀的ORM框架,其中SQLAlchemy以其强大的功能和对多种数据库的支持脱颖而出。本文将深入探讨使用SQLAlchemy操作SQL Server、MySQL和PostgreSQL三大主流数据库的技术实践。
ORM的核心价值在于:
SQLAlchemy作为Python最全面的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}')>"def create_tables(engine):
    Base.metadata.create_all(engine)
# 为所有数据库创建表
create_tables(mysql_engine)
create_tables(postgres_engine)
create_tables(mssql_engine)# 创建会话工厂
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()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提供统一接口,但处理数据库差异时需注意:
# SQL Server需要指定IDENTITY属性
if engine.dialect.name == 'mssql':
    id = Column(Integer, primary_key=True, autoincrement=True)
else:
    id = Column(Integer, primary_key=True)# 标准分页查询
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()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)# 多条件组合
from sqlalchemy import or_
results = session.query(User).filter(
    or_(
        User.name.like('张%'),
        User.email.contains('example')
    )
).order_by(User.created_at.desc()).all()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()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 删除。