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"
)

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()

# 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
)

# 使用selectinload避免N+1查询
from sqlalchemy.orm import selectinload

users = session.query(User).options(selectinload(User.addresses)).all()