SQLAlchemy：Python 数据库开发的瑞士军刀

# 安装核心库
pip install sqlalchemy
# 安装数据库驱动（以 MySQL 为例，根据实际数据库选择）
pip install pymysql  # MySQL 驱动
pip install psycopg2-binary  # PostgreSQL 驱动

dialect+driver://username:password@host:port/database

  from sqlalchemy import create_engine

  engine = create_engine(
      "mysql+pymysql://root:your_password@localhost:3306/test_db",
      pool_size=10,  # 连接池大小
      max_overflow=2  # 连接池最大溢出连接数
  )

  engine = create_engine("sqlite:///example.db", echo=True)  # echo=True 打印执行的 SQL 语句

from sqlalchemy import MetaData, Table, Column, Integer, String, DateTime

# 创建元数据对象
metadata = MetaData()

# 定义 users 表
users = Table(
    "users",
    metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("username", String(50), nullable=False, unique=True),
    Column("email", String(100), nullable=False),
    Column("created_at", DateTime, default=datetime.datetime.now)
)

# 创建表（执行 DDL 语句）
metadata.create_all(engine)

# 单条插入
insert_stmt = users.insert().values(
    username="john_doe",
    email="[email protected]"
)
with engine.connect() as connection:
    result = connection.execute(insert_stmt)
    connection.commit()  # 提交事务
    print(f"插入记录 ID：{result.lastrowid}")  # 输出自增主键值

# 批量插入
users_data = [
    {"username": "alice", "email": "[email protected]"},
    {"username": "bob", "email": "[email protected]"}
]
insert_stmt = users.insert()
with engine.connect() as connection:
    connection.execute(insert_stmt, users_data)
    connection.commit()

from sqlalchemy.sql import select

# 查询所有用户
stmt = select(users)
with engine.connect() as connection:
    result = connection.execute(stmt)
    for row in result:
        print(f"用户 {row.id}: {row.username}, 邮箱: {row.email}")

# 条件查询：查询邮箱包含 @example.com 的用户
stmt = select(users).where(users.c.email.like("%@example.com"))
with engine.connect() as connection:
    result = connection.execute(stmt)
    print(f"符合条件的用户数：{len(result.fetchall())}")

# 排序与限制：查询最新注册的 5 个用户
stmt = select(users).order_by(users.c.created_at.desc()).limit(5)

# 更新操作：将用户名为 john_doe 的邮箱改为新地址
update_stmt = users.update().where(users.c.username == "john_doe").values(
    email="[email protected]"
)
with engine.connect() as connection:
    result = connection.execute(update_stmt)
    connection.commit()
    print(f"更新行数：{result.rowcount}")

# 删除操作：删除用户名为 bob 的记录
delete_stmt = users.delete().where(users.c.username == "bob")
with engine.connect() as connection:
    result = connection.execute(delete_stmt)
    connection.commit()
    print(f"删除行数：{result.rowcount}")

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

# 创建基类
Base = declarative_base()

# 定义 User 模型类，映射到 users 表
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), nullable=False, unique=True)
    email = Column(String(100), nullable=False)
    created_at = Column(DateTime, default=datetime.datetime.now)

    # 可选：定义 __repr__ 方法方便调试
    def __repr__(self):
        return f"<User(id={self.id}, username='{self.username}', email='{self.email}')>"

from sqlalchemy.orm import sessionmaker

# 创建会话工厂，绑定数据库引擎
Session = sessionmaker(bind=engine)

# 创建会话实例
with Session() as session:
    # 创建 User 实例
    new_user = User(
        username="charlie",
        email="[email protected]"
    )
    # 添加到会话（暂不提交到数据库）
    session.add(new_user)
    # 提交事务，执行 INSERT 操作
    session.commit()
    print(f"新用户 ID：{new_user.id}")  # 自动获取自增主键

with Session() as session:
    # 查询所有用户
    all_users = session.query(User).all()
    print(f"用户总数：{len(all_users)}")

    # 按条件查询：查询用户名包含 'alice' 的用户
    user = session.query(User).filter(User.username.like("%alice%")).first()
    if user:
        print(f"找到用户：{user.username}")

    # 排序与分页：查询最新的 3 条记录（offset 偏移量，limit 限制数量）
    recent_users = session.query(User).order_by(User.created_at.desc()).offset(0).limit(3).all()

with Session() as session:
    # 查询需要更新的用户
    user = session.query(User).filter_by(username="john_doe").first()
    if user:
        # 修改属性值
        user.email = "[email protected]"
        # 提交事务，执行 UPDATE 操作（会话自动跟踪脏数据）
        session.commit()
        print("用户邮箱已更新")

with Session() as session:
    # 查询需要删除的用户
    user = session.query(User).filter_by(username="bob").first()
    if user:
        # 删除对象
        session.delete(user)
        session.commit()
        print("用户已删除")

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class Post(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True)
    title = Column(String(100), nullable=False)
    content = Column(String(500))
    created_at = Column(DateTime, default=datetime.datetime.now)
    user_id = Column(Integer, ForeignKey("users.id"))  # 外键关联用户表

    # 定义关系：Post 属于某个 User（反向引用 posts）
    user = relationship("User", back_populates="posts")

# 在 User 类中添加反向关系
class User(Base):
    # ... 原有字段 ...
    posts = relationship("Post", back_populates="user", order_by="Post.created_at.desc()")

metadata.create_all(engine)  # 确保表已创建

with Session() as session:
    # 创建用户
    user = User(username="emma", email="[email protected]")
    # 创建帖子并关联用户
    post = Post(
        title="First Post",
        content="Hello SQLAlchemy!",
        user=user  # 通过关系属性关联
    )
    session.add(user)  # 添加用户时自动关联帖子
    session.commit()
    print(f"帖子 ID：{post.id}, 作者：{post.user.username}")

with Session() as session:
    # 查询用户及其所有帖子（JOIN 操作）
    user = session.query(User).filter_by(username="emma").first()
    if user:
        print(f"用户 {user.username} 的帖子：")
        for post in user.posts:
            print(f"- {post.title} ({post.created_at})")

    # 查询帖子及其作者（通过外键直接关联）
    post = session.query(Post).filter_by(title="First Post").first()
    print(f"帖子作者：{post.user.username}")

with Session() as session:
    # 执行原生 SQL 查询，返回 ORM 对象
    sql = "SELECT * FROM users WHERE username = :username"
    user = session.query(User).from_statement(sql).params(username="emma").first()
    print(f"通过原生 SQL 查询到用户：{user.username}")

    # 执行原生 INSERT 语句（非 ORM 方式）
    with engine.connect() as connection:
        connection.execute(
            text("INSERT INTO posts (title, content, user_id) VALUES (:title, :content, :user_id)"),
            {"title": "Native SQL Post", "content": "This is inserted via raw SQL", "user_id": user.id}
        )
        connection.commit()

class Comment(Base):
    __tablename__ = "comments"

    id = Column(Integer, primary_key=True)
    content = Column(String(300), nullable=False)
    created_at = Column(DateTime, default=datetime.datetime.now)
    post_id = Column(Integer, ForeignKey("posts.id"))
    user_id = Column(Integer, ForeignKey("users.id"))

    # 定义关系
    post = relationship("Post", back_populates="comments")
    user = relationship("User", back_populates="comments")

# 更新 Post 模型，添加 comments 关系
class Post(Base):
    # ... 原有字段 ...
    comments = relationship("Comment", back_populates="post", order_by="Comment.created_at")

# 更新 User 模型，添加 comments 关系
class User(Base):
    # ... 原有字段 ...
    comments = relationship("Comment", back_populates="user", order_by="Comment.created_at")

with Session() as session:
    # 查询用户名为 emma 的用户及其所有帖子和评论
    user = session.query(User).filter_by(username="emma").first()
    if user:
        print(f"用户：{user.username} ({user.email})")
        print(f"发布的帖子数量：{len(user.posts)}")

        for post in user.posts:
            print(f"\n帖子标题：{post.title}")
            print(f"发布时间：{post.created_at}")
            print(f"评论数量：{len(post.comments)}")

            for comment in post.comments:
                print(f"- {comment.user.username} 评论：{comment.content}")

from sqlalchemy.orm import joinedload

with Session() as session:
    # 预加载用户的帖子和评论，减少 SQL 查询次数
    user = session.query(User).options(
        joinedload(User.posts).joinedload(Post.comments)
    ).filter_by(username="emma").first()

    if user:
        print(f"用户：{user.username} 的所有内容（预加载）：")
        for post in user.posts:
            print(f"帖子 {post.id}：{post.title}")
            for comment in post.comments:
                print(f"  评论：{comment.content}")

with Session() as session:
    try:
        # 创建用户和帖子（多步操作）
        user = User(username="frank", email="[email protected]")
        post = Post(title="Transaction Test", content="Atomic operation", user=user)

        session.add(user)
        session.add(post)
        # 提交事务：所有操作成功才写入数据库
        session.commit()
        print("事务提交成功")
    except Exception as e:
        # 发生异常时回滚所有操作
        session.rollback()
        print(f"事务失败，已回滚：{str(e)}")

class User(Base):
    __tablename__ = "users"
    # ... 其他字段 ...
    email = Column(String(100), nullable=False, index=True)  # 为 email 字段创建索引

# 复合索引（多字段组合查询优化）
from sqlalchemy import Index
Index("idx_username_email", User.username, User.email)  # 对 username 和 email 创建复合索引

pip install alembic
alembic init migrations  # 初始化迁移环境

from myapp.models import Base  # 导入你的模型基类
target_metadata = Base.metadata

# 配置数据库连接（或读取环境变量）
config.set_main_option("sqlalchemy.url", "mysql+pymysql://root:password@localhost/test_db")

# 生成迁移脚本（自动对比模型与数据库差异）
alembic revision --autogenerate -m "add user age column"

# 应用迁移（更新数据库表结构）
alembic upgrade head

engine = create_engine(
    "postgresql+psycopg2://user:pass@localhost/db",
    pool_size=10,          # 常驻连接数
    max_overflow=20,       # 临时溢出连接数（超过 pool_size 时）
    pool_recycle=3600,     # 连接超时时间（秒），避免数据库主动断开连接
    pool_pre_ping=True     # 连接使用前检测可用性
)

from sqlalchemy.orm import selectinload

# 加载用户及其帖子（适合一对多关系）
users = session.query(User).options(selectinload(User.posts)).all()

  # 批量插入（比循环 add 高效）
  session.bulk_insert_mappings(
      User,
      [{"username": f"user_{i}", "email": f"user_{i}@example.com"} for i in range(1000)]
  )
  session.commit()

  engine = create_engine("sqlite:///test.db", echo=True)  # 打印执行的 SQL

  query = session.query(User).filter(User.email.like("%@example.com"))
  print(query.explain())  # 输出 SQL 执行计划