SQLAlchemy2 从入门到精通（超详细实战教程）

# 安装核心库
pip install sqlalchemy
# 查看已安装的 sqlalchemy 包的详细信息
pip show sqlalchemy

name: sqlalchemy
version: 2.0.48
summary: database abstraction library
home-page: https://www.sqlalchemy.org
author: mike bayer
author-email: mike_mp@zzzcomputing.com
license: mit
location: d:\workspaces\python\test\test01\.venv\lib\site-packages
requires: greenlet, typing-extensions
required-by:

# 导入核心模块（2.0+ 推荐的导入方式）
from typing import generator, optional
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarativebase, session
from sqlalchemy.pool import queuepool  # 连接池配置
# ====================== 1. 数据库引擎优化（核心性能优化） ======================
# 针对 sqlite 和 sqlalchemy 2.0+ 的引擎配置优化
# 关键优化点：
# - 关闭 echo（生产环境必须关闭，调试时可开启）
# - 配置连接池（sqlite 虽无真正连接池，但配置可兼容其他数据库）
# - 设置 sqlite 专属优化参数（提升读写性能）
sync_engine = create_engine(
    "sqlite:///sync_test.db",
    # 基础配置
    echo=true,  # 生产环境关闭 sql 打印，提升性能
    echo_pool=false,  # 关闭连接池日志
    # 连接池配置（适配 mysql/postgresql 等数据库时更重要）
    poolclass=queuepool,
    pool_size=5,  # 核心连接数
    max_overflow=10,  # 最大溢出连接数
    pool_recycle=3600,  # 1小时回收连接，避免失效
    pool_pre_ping=true,  # 获取连接前检测是否有效，防止死连接
    # sqlite 专属优化参数（2.0+ 推荐）
    connect_args={
        "check_same_thread": false,  # 允许多线程访问（sqlite 必要配置）
        "timeout": 30,  # 数据库锁定超时时间（默认5秒，提升到30秒避免锁死）
    },
)
# ====================== 2. 模型基类优化（2.0+ 规范） ======================
# sqlalchemy 2.0+ 推荐直接继承 declarativebase（替代旧的 declarative_base()）
class base(declarativebase):
    """所有数据模型的基类（2.0+ 规范写法）"""
    __abstract__ = true  # 标记为抽象类，不会生成数据库表
# ====================== 3. 会话工厂优化（安全性+规范性） ======================
# 2.0+ 推荐显式指定类型，同时保留核心配置
sessionlocal = sessionmaker(
    autocommit=false,
    autoflush=false,
    bind=sync_engine,
    expire_on_commit=false,  # 优化：提交后不自动过期对象，提升查询性能
    class_=session,  # 显式指定会话类（2.0+ 规范）
)
# ====================== 4. 会话获取函数优化（类型安全+容错性） ======================
def get_db() -> generator[session, none, none]:
    """
    获取数据库会话生成器（2.0+ 类型安全版本）
    特性：
    - 类型注解明确，支持 ide 提示
    - 异常捕获并友好提示
    - 确保会话最终关闭
    """
    db_session: optional[session] = none  # 初始化变量
    try:
        db_session = sessionlocal()
        yield db_session
    except exception as e:
        # 可选：记录日志（推荐生产环境添加）
        # import logging; logging.error(f"数据库会话异常: {str(e)}")
        db_session.rollback()  # 异常时回滚未提交的操作
        raise  # 重新抛出异常，让上层处理
    finally:
        if db_session is not none:
            db_session.close()  # 确保会话无论是否异常都关闭
if __name__ == "__main__":
    # 测试示例
    db = next(get_db())
    print(db)  # <sqlalchemy.orm.session.session object at 0x00000297efe323c0>

import asyncio
from typing import asyncgenerator, optional
from sqlalchemy.ext.asyncio import (
    create_async_engine,
    async_sessionmaker,
    asyncsession,
    asyncengine  # 显式导入类型
)
from sqlalchemy.orm import declarativebase  # 2.0+ 推荐的基类
from sqlalchemy.pool import asyncadaptedqueuepool  # 异步连接池
# ====================== 1. 配置项抽离（提升可维护性） ======================
# 数据库连接配置（统一管理，方便环境切换）
database_url = "sqlite+aiosqlite:///async_test.db"
# 调试开关（建议通过环境变量控制，如 os.getenv("debug", "false") == "true"）
debug_mode = true
# ====================== 2. 异步引擎优化（核心性能+兼容性） ======================
async_engine: asyncengine = create_async_engine(
    url=database_url,
    # 基础配置
    echo=debug_mode,  # 生产环境关闭，调试时开启
    echo_pool=false,  # 关闭连接池日志（减少io开销）
    # 异步连接池配置（2.0+ 推荐 asyncadaptedqueuepool）
    poolclass=asyncadaptedqueuepool,
    pool_size=5,  # 核心连接数（sqlite 无实际连接，适配其他数据库）
    max_overflow=10,  # 最大溢出连接数
    pool_recycle=3600,  # 1小时回收连接，避免失效（适配mysql/postgresql）
    pool_pre_ping=true,  # 获取连接前检测有效性，防止死连接
    # sqlite 专属异步优化参数
    connect_args={
        "check_same_thread": false,  # 解决sqlite线程安全问题
        "timeout": 30,  # 数据库锁定超时时间（默认5秒，提升稳定性）
    },
)
# ====================== 3. 异步会话工厂优化（2.0+ 规范） ======================
asyncsessionlocal = async_sessionmaker(
    bind=async_engine,
    class_=asyncsession,  # 显式指定异步会话类（2.0+ 规范）
    autocommit=false,
    autoflush=false,
    expire_on_commit=false,  # 提交后不失效对象（异步场景减少重复查询）
)
# ====================== 4. 模型基类优化（2.0+ 官方推荐） ======================
# 替代旧的 declarative_base()，2.0+ 推荐直接继承 declarativebase
class base(declarativebase):
    """所有异步模型的基类（2.0+ 规范写法）"""
    __abstract__ = true  # 标记为抽象类，不生成数据库表
# ====================== 5. 异步会话获取函数（核心优化，解决原逻辑问题） ======================
async def get_async_db() -> asyncgenerator[asyncsession, none]:
    """
    获取异步数据库会话生成器（适配 fastapi 依赖注入）
    优化点：
    1. 移除 async with 与手动 close() 的冲突
    2. 取消自动提交（避免业务逻辑未完成时误提交）
    3. 完善异常处理和类型注解
    4. 确保会话最终关闭，防止连接泄漏
    """
    session: optional[asyncsession] = none
    try:
        session = asyncsessionlocal()  # 手动创建会话，替代 async with
        yield session  # 传递会话给业务逻辑
    except exception as e:
        # 异常时回滚所有未提交的操作
        if session:
            await session.rollback()
        raise e  # 重新抛出异常，让上层框架（如fastapi）处理
    finally:
        # 确保会话最终关闭，释放连接
        if session:
            await session.close()
# ====================== 6. 可选：创建数据库表的异步函数（通用工具） ======================
async def create_all_tables() -> none:
    """异步创建所有模型对应的数据库表（初始化用）"""
    async with async_engine.begin() as conn:
        # 异步执行表创建（2.0+ 推荐使用 engine.begin()）
        await conn.run_sync(base.metadata.create_all)
# 测试示例（可选）
async def main():
    # 初始化数据库表
    # await create_all_tables()
    # 使用会话
    async for db in get_async_db():
        # 业务逻辑示例
        print(f"会话状态：{db.is_active}")  # 输出: 会话状态：true
if __name__ == "__main__":
    asyncio.run(main())

from sqlalchemy import column, integer, string, datetime, boolean, index
from sqlalchemy.sql import func  # 2.0+ 推荐的函数工具
# 注意：base 需从之前优化后的 declarativebase 子类导入
# 导入 base 根据同步/异步场景选择
# from async_session import base
from sync_session import base
# ====================== 用户表模型（2.0+ 规范写法） ======================
class user(base):
    """
    用户表模型
    表名：users
    核心字段：id、用户名、邮箱、年龄、激活状态、创建/更新时间
    """
    __tablename__ = "users"
    # 1. 核心字段优化（2.0+ 规范 + 约束增强）
    id = column(
        integer,
        primary_key=true,
        index=true,
        comment="用户id（自增主键）",
        autoincrement=true  # 显式声明自增（兼容多数据库）
    )
    username = column(
        string(50),
        unique=true,
        nullable=false,
        comment="用户名（唯一，非空）",
        index=true  # 高频查询字段加索引，提升查询性能
    )
    email = column(
        string(100),
        unique=true,
        nullable=true,
        comment="邮箱（唯一，可选）",
        index=true  # 邮箱查询场景多，添加索引
    )
    age = column(
        integer,
        default=0,
        nullable=false,  # 显式声明非空（默认值已保证，增强约束）
        comment="年龄（默认0）"
    )
    is_active = column(
        boolean,
        default=true,
        nullable=false,
        comment="是否激活（默认true）",
        index=true  # 状态字段高频过滤，添加索引
    )
    # 时间字段优化：使用 func.now() 替代 datetime.now（数据库层面生成时间，兼容异步）
    create_time = column(
        datetime,
        default=func.now(),  # 2.0+ 推荐，数据库服务器时间（更精准）
        nullable=false,
        comment="创建时间（自动生成）"
    )
    # 新增更新时间字段（业务常用，2.0+ 支持 onupdate）
    update_time = column(
        datetime,
        default=func.now(),
        onupdate=func.now(),  # 数据更新时自动刷新
        nullable=false,
        comment="更新时间（自动更新）"
    )
    # 2. 复合索引（针对多字段查询场景，提升性能）
    __table_args__ = (
        # 示例：按「激活状态+创建时间」查询的复合索引
        index("idx_user_active_create", "is_active", "create_time"),
    )
    # 3. 自定义方法优化（类型注解 + 实用方法）
    def __repr__(self) -> str:
        """自定义打印格式（类型注解增强）"""
        return f"<user(id={self.id}, username='{self.username}', email='{self.email}', is_active={self.is_active})>"
    def to_dict(self) -> dict:
        """新增：模型转字典（业务常用，避免手动序列化）"""
        return {
            "id": self.id,
            "username": self.username,
            "email": self.email,
            "age": self.age,
            "is_active": self.is_active,
            "create_time": self.create_time.strftime("%y-%m-%d %h:%m:%s") if self.create_time else none,
            "update_time": self.update_time.strftime("%y-%m-%d %h:%m:%s") if self.update_time else none,
        }

# 注意：base 需从之前优化后的 declarativebase 子类导入
# 导入 base 根据同步/异步场景选择
# from session_async import base, async_engine
from session_sync import base, sync_engine
# 导入数据模型（必须在创建表之前导入，这样 sqlalchemy 才能知道要创建哪些表）
from models import user
# ====================== 表创建逻辑优化（兼容同步/异步） ======================
# 1. 同步场景创建表（2.0+ 规范写法）
def create_tables_sync(engine) -> none:
    """同步创建所有表（仅初始化时执行）"""
    # checkfirst=true：默认值，先检查表是否存在，避免重复创建
    base.metadata.create_all(bind=engine, checkfirst=true)
# 2. 异步场景创建表（2.0+ 推荐写法）
async def create_tables_async(engine) -> none:
    """异步创建所有表（适配异步引擎）"""
    async with engine.begin() as conn:
        # run_sync 适配异步引擎执行同步的 metadata.create_all
        await conn.run_sync(base.metadata.create_all, checkfirst=true)
# ====================== 调用示例 ======================
# 同步场景
create_tables_sync(sync_engine)
# # 异步场景（需在异步上下文执行）
# import asyncio
# asyncio.run(create_tables_async(async_engine))

from typing import list, optional, sequence
from sqlalchemy import select, update, delete, desc
from sqlalchemy.orm import session
# 假设 base、user 模型、get_db 已从优化后的配置文件导入
from models import user
from sync_session import get_db
# ====================== 通用工具函数（提升复用性） ======================
def get_db_session() -> session:
    """获取数据库会话（封装 get_db，简化调用）"""
    return next(get_db())
# ====================== 3.1.1 新增数据（create）- 2.0+ 优化 ======================
def create_users():
    """新增数据（单个/批量）- 2.0+ 规范写法"""
    db: session = get_db_session()
    try:
        # 方式1：创建单个对象（添加类型注解）
        user1: user = user(username="zhangsan", email="zhangsan@example.com", age=20)
        db.add(user1)
        # 方式2：批量创建对象（类型注解 + 列表初始化）
        batch_users: list[user] = [
            user(username="lisi", email="lisi@example.com", age=22),
            user(username="wangwu", email="wangwu@example.com", age=25)
        ]
        db.add_all(batch_users)
        # 提交事务（2.0+ 推荐先 flush 再 commit，确保获取自增id）
        db.flush()  # 预提交，生成id但不持久化
        db.commit()  # 最终提交
        # 刷新对象（获取数据库自动生成的字段）
        db.refresh(user1)
        print(f"新增用户id：{user1.id}")  # 输出：新增用户id：1
    except exception as e:
        db.rollback()  # 异常回滚
        raise runtimeerror(f"新增用户失败：{str(e)}") from e
    finally:
        db.close()  # 确保会话关闭

# ====================== 3.1.2 查询数据（read）- 2.0+ 核心优化 ======================
def query_users():
    """查询数据 - 2.0+ 推荐使用 select() 构造器（替代旧 query api）"""
    db: session = get_db_session()
    try:
        # 1. 查询所有用户
        stmt = select(user)
        users: sequence[user] = db.scalars(stmt).all()
        print("所有用户：", users)   # 输出: [<user(id=1, username='zhangsan', email='zhangsan@example.com', is_active=true)>, <user(id=2, username='lisi', email='lisi@example.com', is_active=true)>, <user(id=3, username='wangwu', email='wangwu@example.com', is_active=true)>]
        # 2. 查询单个用户（按主键）
        user: optional[user] = db.get(user, 1)  # 2.0+ 推荐直接用 session.get()
        print("主键为1的用户：", user)   # 输出: <user(id=1, username='zhangsan', email='zhangsan@example.com', is_active=true)>
        # 3. 条件查询
        # 方式1：filter（2.0+ 推荐，支持复杂条件）
        stmt = select(user).where(user.username == "zhangsan")
        user: optional[user] = db.scalars(stmt).first()
        print("方式1：用户名是zhangsan的用户：", user)   # 输出: <user(id=1, username='zhangsan', email='zhangsan@example.com', is_active=true)>
        # 方式2：简易条件（2.0+ 无 filter_by，用 where 简化）
        stmt = select(user).where(user.username == "zhangsan")
        user: optional[user] = db.scalars(stmt).first()
        print("方式2：用户名是zhangsan的用户：", user)   # 输出: <user(id=1, username='zhangsan', email='zhangsan@example.com', is_active=true)>
        # 4. 多条件查询
        stmt = select(user).where(user.age > 20, user.is_active == true)
        users: sequence[user] = db.scalars(stmt).all()
        print("年龄>20且激活的用户：", users)   # 输出: [<user(id=2, username='lisi', email='lisi@example.com', is_active=true)>, <user(id=3, username='wangwu', email='wangwu@example.com', is_active=true)>]
        # 5. 排序查询（2.0+ 用 desc()/asc() 函数）
        stmt = select(user).order_by(desc(user.age))
        users: sequence[user] = db.scalars(stmt).all()
        print("按年龄降序的用户：", users)   # 输出: [<user(id=3, username='wangwu', email='wangwu@example.com', is_active=true)>, <user(id=2, username='lisi', email='lisi@example.com', is_active=true)>, <user(id=1, username='zhangsan', email='zhangsan@example.com', is_active=true)>]
        # 6. 限制查询结果数量（2.0+ 用 limit() 方法）
        stmt = select(user).limit(2)
        users: sequence[user] = db.scalars(stmt).all()
        print("前2个用户：", users)   # 输出: [<user(id=1, username='zhangsan', email='zhangsan@example.com', is_active=true)>, <user(id=2, username='lisi', email='lisi@example.com', is_active=true)>]
    finally:
        db.close()

# ====================== 3.1.3 修改数据（update）- 2.0+ 优化 ======================
def update_users():
    """修改数据（单个/批量）- 2.0+ 规范写法"""
    db: session = get_db_session()
    try:
        # 1. 修改单个对象（查询-修改-提交）
        stmt = select(user).where(user.username == "zhangsan")
        user: optional[user] = db.scalars(stmt).first()
        if user:
            # 修改属性
            user.age = 21
            user.email = "zhangsan_new@example.com"
            db.commit()  # 提交修改
            db.refresh(user)  # 刷新对象获取最新数据
            print("修改后的用户：", user)  # 输出: <user(id=1, username='zhangsan', email='zhangsan_new@example.com', is_active=true)>

        # 2. 批量修改（2.0+ 新 api：update() 构造器）
        stmt = (
            update(user)
            .where(user.age < 25)
            .values(is_active=false)  # 2.0+ 用 values() 指定修改字段
            .execution_options(synchronize_session="fetch")  # 同步会话数据
        )
        result = db.execute(stmt)
        db.commit()
        # 输出: 批量修改完成，影响行数：2
        print(f"批量修改完成，影响行数：{result.rowcount}")  # type: ignore[attr-defined]
    except exception as e:
        db.rollback()
        raise runtimeerror(f"修改用户失败：{str(e)}") from e
    finally:
        db.close()

# ====================== 3.1.4 删除数据（delete）- 2.0+ 优化 ======================
def delete_users():
    """删除数据（单个/批量）- 2.0+ 规范写法"""
    db: session = get_db_session()
    try:
        # 1. 删除单个对象
        stmt = select(user).where(user.username == "wangwu")
        user: optional[user] = db.scalars(stmt).first()
        if user:
            db.delete(user)
            db.commit()
            print("删除用户完成")

        # 2. 批量删除（2.0+ 新 api：delete() 构造器）
        stmt = (
            delete(user)
            .where(user.is_active == false)
            .execution_options(synchronize_session="fetch")
        )
        result = db.execute(stmt)
        db.commit()
        # 输出: 批量删除完成，影响行数：2
        print(f"批量删除完成，影响行数：{result.rowcount}")  # type: ignore[attr-defined]
    except exception as e:
        db.rollback()
        raise runtimeerror(f"删除用户失败：{str(e)}") from e
    finally:
        db.close()


# ====================== 调用示例 ======================
if __name__ == "__main__":
    # 执行新增
    # create_users()
    # 执行查询
    # query_users()
    # 执行修改
    # update_users()
    # 执行删除
    delete_users()

import asyncio
from typing import list, optional
from sqlalchemy import select, update, delete, desc, sequence
from sqlalchemy.ext.asyncio import asyncsession
# 假设从优化后的异步配置文件导入以下依赖
from models import user
# ====================== 通用工具函数（异步） ======================
async def get_async_db_session() -> asyncsession:
    """获取异步数据库会话（封装 get_async_db，简化调用）"""
    # 由于 get_async_db 现在是上下文管理器，我们需要一个不同的实现
    # 直接创建会话而不是使用上下文管理器
    from async_session import asyncsessionlocal
    session = asyncsessionlocal()
    return session
# ====================== 3.1.1 新增数据（create）- 异步版 ======================
async def create_users_async():
    """异步新增数据（单个/批量）"""
    db: asyncsession = await get_async_db_session()
    try:
        # 方式1：创建单个对象
        user1: user = user(username="zhangsan", email="zhangsan@example.com", age=20)
        db.add(user1)
        # 方式2：批量创建对象
        batch_users: list[user] = [
            user(username="lisi", email="lisi@example.com", age=22),
            user(username="wangwu", email="wangwu@example.com", age=25)
        ]
        db.add_all(batch_users)
        # 异步提交（2.0+ 异步必须用 await）
        await db.flush()  # 预提交，生成自增id
        await db.commit()  # 最终提交
        # 异步刷新对象
        await db.refresh(user1)
        print(f"新增用户id：{user1.id}")  # 输出：新增用户id：1
    except exception as e:
        await db.rollback()  # 异步回滚
        raise runtimeerror(f"异步新增用户失败：{str(e)}") from e
    finally:
        await db.close()  # 异步关闭会话

# ====================== 3.1.2 查询数据（read）- 异步版 ======================
async def query_users_async():
    """异步查询数据（2.0+ 异步 select 构造器）"""
    db: asyncsession = await get_async_db_session()
    try:
        # 1. 查询所有用户（异步核心：await + scalars + all）
        stmt = select(user)
        result = await db.scalars(stmt)  # 异步执行查询
        users: sequence[user] = result.all()
        print("所有用户：",
              users)  # 输出: [<user(id=1, username='zhangsan', email='zhangsan@example.com', is_active=true)>, <user(id=2, username='lisi', email='lisi@example.com', is_active=true)>, <user(id=3, username='wangwu', email='wangwu@example.com', is_active=true)>]
        # 2. 查询单个用户（按主键，异步 get）
        user: optional[user] = await db.get(user, 1)  # 异步 get 方法
        print("主键为1的用户：",
              user)  # 输出: <user(id=1, username='zhangsan', email='zhangsan@example.com', is_active=true)>
        # 3. 条件查询
        # 方式1：where 条件 + 异步执行
        stmt = select(user).where(user.username == "zhangsan")
        user: optional[user] = (await db.scalars(stmt)).first()
        print("方式1：用户名是zhangsan的用户：",
              user)  # 输出: <user(id=1, username='zhangsan', email='zhangsan@example.com', is_active=true)>
        # 4. 多条件查询
        stmt = select(user).where(user.age > 20, user.is_active == true)
        users: sequence[user] = (await db.scalars(stmt)).all()
        print("年龄>20且激活的用户：",
              users)  # 输出: [<user(id=2, username='lisi', email='lisi@example.com', is_active=true)>, <user(id=3, username='wangwu', email='wangwu@example.com', is_active=true)>]
        # 5. 排序查询（2.0+ desc 函数）
        stmt = select(user).order_by(desc(user.age))
        users: sequence[user] = (await db.scalars(stmt)).all()
        print("按年龄降序的用户：",
              users)  # 输出: [<user(id=3, username='wangwu', email='wangwu@example.com', is_active=true)>, <user(id=2, username='lisi', email='lisi@example.com', is_active=true)>, <user(id=1, username='zhangsan', email='zhangsan@example.com', is_active=true)>]
        # 6. 限制查询结果数量
        stmt = select(user).limit(2)
        users: sequence[user] = (await db.scalars(stmt)).all()
        print("前2个用户：",
              users)  # 输出: [<user(id=1, username='zhangsan', email='zhangsan@example.com', is_active=true)>, <user(id=2, username='lisi', email='lisi@example.com', is_active=true)>]
    finally:
        await db.close()

# ====================== 3.1.3 修改数据（update）- 异步版 ======================
async def update_users_async():
    """异步修改数据（单个/批量）"""
    db: asyncsession = await get_async_db_session()
    try:
        # 1. 修改单个对象（查询-修改-提交）
        stmt = select(user).where(user.username == "zhangsan")
        user: optional[user] = (await db.scalars(stmt)).first()
        if user:
            user.age = 21
            user.email = "zhangsan_new@example.com"
            await db.commit()  # 异步提交
            await db.refresh(user)  # 异步刷新
            print("修改后的用户：",
                  user)  # 输出: <user(id=1, username='zhangsan', email='zhangsan_new@example.com', is_active=true)>
        # 2. 批量修改（2.0+ 异步 update 构造器）
        stmt = (
            update(user)
            .where(user.age < 25)
            .values(is_active=false)
            .execution_options(synchronize_session="fetch")
        )
        result = await db.execute(stmt)  # 异步执行批量更新
        await db.commit()
        # 输出: 批量修改完成，影响行数：2
        print(f"批量修改完成，影响行数：{result.rowcount}")  # type: ignore[attr-defined]
    except exception as e:
        await db.rollback()
        raise runtimeerror(f"异步修改用户失败：{str(e)}") from e
    finally:
        await db.close()

# ====================== 3.1.4 删除数据（delete）- 异步版 ======================
async def delete_users_async():
    """异步删除数据（单个/批量）"""
    db: asyncsession = await get_async_db_session()
    try:
        # 1. 删除单个对象
        stmt = select(user).where(user.username == "wangwu")
        user: optional[user] = (await db.scalars(stmt)).first()
        if user:
            await db.delete(user)  # 异步删除
            await db.commit()
            print("删除用户完成")
        # 2. 批量删除（2.0+ 异步 delete 构造器）
        stmt = delete(user).where(user.is_active == false)
        result = await db.execute(stmt)  # 异步执行批量删除
        await db.commit()
        # 输出: 批量删除完成，影响行数：2
        print(f"批量删除完成，影响行数：{result.rowcount}")  # type: ignore[attr-defined]
    except exception as e:
        await db.rollback()
        raise runtimeerror(f"异步删除用户失败：{str(e)}") from e
    finally:
        await db.close()
# ====================== 异步入口函数（执行所有操作） ======================
async def main():
    """异步主函数：按顺序执行所有 crud 操作"""
    # 1. 新增数据
    # await create_users_async()
    # 2. 查询数据
    # await query_users_async()
    # 3. 修改数据
    # await update_users_async()
    # 4. 删除数据
    await delete_users_async()
# ====================== 执行异步代码 ======================
if __name__ == "__main__":
    # 异步代码必须在 asyncio 事件循环中执行
    asyncio.run(main())

from typing import tuple
from sqlalchemy import (
    column, integer, string, datetime, boolean,
    or_, and_, not_, func, select, distinct, sequence
)
from sqlalchemy.orm import session, declarativebase
from sync_crud import get_db_session
from sync_session import sync_engine
# ====================== 基础模型定义（2.0+ 规范） ======================
class base(declarativebase):
    __abstract__ = true
class user(base):
    __tablename__ = "users"
    id = column(integer, primary_key=true, index=true, autoincrement=true, comment="用户id")
    username = column(string(50), unique=true, nullable=false, index=true, comment="用户名")
    email = column(string(100), unique=true, nullable=true, index=true, comment="邮箱")
    age = column(integer, default=0, nullable=false, comment="年龄")
    is_active = column(boolean, default=true, nullable=false, index=true, comment="是否激活")
    create_time = column(datetime, default=func.now(), nullable=false, comment="创建时间")
    def __repr__(self) -> str:
        return f"<user(id={self.id}, username='{self.username}', email='{self.email}')>"
class article(base):
    __tablename__ = "articles"
    id = column(integer, primary_key=true, index=true, autoincrement=true, comment="文章id")
    title = column(string(100), nullable=false, comment="文章标题")
    content = column(string(500), comment="文章内容")
    user_id = column(integer, nullable=false, index=true, comment="关联用户id")
    create_time = column(datetime, default=func.now(), nullable=false, comment="创建时间")
    def __repr__(self) -> str:
        return f"<article(id={self.id}, title='{self.title}', user_id={self.user_id})>"
# ====================== 1.1 初始化表和测试数据 ======================
def init_data():
    """初始化测试数据（仅首次执行）"""
    db: session = get_db_session()
    try:
        # 创建表
        print("正在创建数据库表...")
        base.metadata.create_all(bind=sync_engine)
        print("表创建完成")
        # 清空旧数据（避免重复）
        print("正在清空旧数据...")
        db.query(user).delete()
        db.query(article).delete()
        db.commit()
        print("旧数据已清空")
        # 新增用户
        print("正在新增用户...")
        users = [
            user(username="zhangsan", email="zhangsan@example.com", age=20),
            user(username="lisi", email="lisi@example.com", age=22),
            user(username="wangwu", email="wangwu@example.com", age=25),
            user(username="zhaoliu", email=none, age=30)  # type: ignore[arg-type]
        ]
        db.add_all(users)
        db.commit()
        print(f"用户新增完成，共 {len(users)} 个用户")
        # 新增文章
        print("正在新增文章...")
        articles = [
            article(title="python 入门", content="sqlalchemy 学习", user_id=1),
            article(title="sqlalchemy 进阶", content="关联查询", user_id=1),
            article(title="mysql 优化", content="索引使用", user_id=2)
        ]
        db.add_all(articles)
        db.commit()
        print(f"文章新增完成，共 {len(articles)} 篇文章")
        print("\n=== 测试数据初始化完成 ===\n")
    except exception as e:
        db.rollback()
        error_msg = f"初始化数据失败：{str(e)}"
        print(error_msg)
        import traceback
        traceback.print_exc()
        raise runtimeerror(error_msg) from e
    finally:
        db.close()

# ====================== 1.2 筛选查询（2.0+ 优化版） ======================
def filter_query():
    """筛选查询（2.0+ select 构造器）"""
    db: session = get_db_session()
    try:
        # 1. 或条件（or_）
        stmt = select(user).where(or_(user.age == 20, user.username == "lisi"))
        users: sequence[user] = db.scalars(stmt).all()
        print("或条件（or_）: ", users)  # 输出: [<user(id=1, username='zhangsan', email='zhangsan@example.com')>, <user(id=2, username='lisi', email='lisi@example.com')>]
        # 2. 与条件（and_）
        stmt = select(user).where(and_(user.age > 20, user.is_active == true))
        users = db.scalars(stmt).all()
        print("与条件（and_）: ", users)  # 输出: [<user(id=2, username='lisi', email='lisi@example.com')>, <user(id=3, username='wangwu', email='wangwu@example.com')>, <user(id=4, username='zhaoliu', email='none')>]
        # 3. 非条件（not_）
        stmt = select(user).where(not_(user.age == 20))
        users = db.scalars(stmt).all()
        print("非条件（not_）: ", users)  # 输出: [<user(id=2, username='lisi', email='lisi@example.com')>, <user(id=3, username='wangwu', email='wangwu@example.com')>, <user(id=4, username='zhaoliu', email='none')>]
        # 4. 模糊查询（like）
        # 以zhang开头
        stmt = select(user).where(user.username.like("zhang%"))
        users = db.scalars(stmt).all()
        print("模糊查询（以zhang开头）: ", users)  # 输出: [<user(id=1, username='zhangsan', email='zhangsan@example.com')>]
        # 包含si
        stmt = select(user).where(user.username.like("%si%"))
        users = db.scalars(stmt).all()
        print("模糊查询（包含si）: ", users)  # 输出: [<user(id=2, username='lisi', email='lisi@example.com')>]
        # 5. 范围查询（in_）
        stmt = select(user).where(user.age.in_([20, 21, 22]))
        users = db.scalars(stmt).all()
        print("范围查询（in_）: ", users)  # 输出: [<user(id=1, username='zhangsan', email='zhangsan@example.com')>, <user(id=2, username='lisi', email='lisi@example.com')>]
        # 6. 空值查询
        # 邮箱为空
        stmt = select(user).where(user.email.is_(none))
        users = db.scalars(stmt).all()
        print("空值查询（is_）: ", users)  # 输出: [<user(id=4, username='zhaoliu', email='none')>]
        # 邮箱不为空
        stmt = select(user).where(user.email.isnot(none))
        users = db.scalars(stmt).all()
        print("空值查询（isnot）: ", users)  # 输出: [<user(id=1, username='zhangsan', email='zhangsan@example.com')>, <user(id=2, username='lisi', email='lisi@example.com')>, <user(id=3, username='wangwu', email='wangwu@example.com')>]
    finally:
        db.close()

# ====================== 1.3 聚合查询（2.0+ 优化版） ======================
def aggregate_query():
    """聚合查询（2.0+ func 优化）"""
    db: session = get_db_session()
    try:
        # 1. 计数（推荐用 scalar() 获取单个值）
        stmt = select(func.count(user.id))
        total_users: int = db.scalar(stmt)
        print("总用户数：", total_users)  # 输出: 4
        # 2. 求和
        stmt = select(func.sum(user.age))
        total_age: int = db.scalar(stmt)
        print("年龄总和：", total_age)  # 输出: 97
        # 3. 平均值（转换为浮点数，避免精度丢失）
        stmt = select(func.avg(user.age))
        avg_age_value = db.scalar(stmt)
        avg_age: float = float(avg_age_value) if avg_age_value is not none else 0.0
        # 保留 2 位小数
        print("平均年龄：", round(avg_age, 2))  # 输出: 24.25
        # 4. 最大值/最小值
        stmt_max = select(func.max(user.age))
        max_age: int = db.scalar(stmt_max)
        stmt_min = select(func.min(user.age))
        min_age: int = db.scalar(stmt_min)
        print("最大年龄：", max_age, "最小年龄：", min_age)  # 输出: 最大年龄： 30 最小年龄： 20
        # 5. 分组聚合（按激活状态分组）
        stmt = select(user.is_active, func.count(user.id)).group_by(user.is_active)
        result: sequence[tuple[bool, int]] = db.execute(stmt).all()
        print("按激活状态分组：", result)  # 输出: [(true, 4)]
    finally:
        db.close()

# ====================== 1.4 关联查询（2.0+ 优化版） ======================
def join_query():
    """关联查询（2.0+ join/outerjoin 规范写法）"""
    db: session = get_db_session()
    try:
        # 1. 内连接（inner join）：查询用户及其文章
        # 生成的sql：select users.id, users.username, users.email, users.age, users.is_active, users.create_time, articles.id as id_1, articles.title, articles.content, articles.user_id, articles.create_time as create_time_1
        # from users join articles on users.id = articles.user_id
        stmt = select(user, article).join(article, user.id == article.user_id)
        result: sequence[tuple[user, article]] = db.execute(stmt).all()
        print("\n内连接结果：")
        for user, article in result:  # type: ignore[misc]
            print(f"用户：{user.username}，文章：{article.title}")
            # 输出:
            # 用户：zhangsan，文章：python 入门
            # 用户：zhangsan，文章：sqlalchemy 进阶
            # 用户：lisi，文章：mysql 优化
        # 2. 左连接（left join）：查询所有用户（包括无文章的）
        # 生成的sql：select users.id, users.username, users.email, users.age, users.is_active, users.create_time, articles.id as id_1, articles.title, articles.content, articles.user_id, articles.create_time as create_time_1
        # from users left outer join articles on users.id = articles.user_id
        stmt = select(user, article).outerjoin(article, user.id == article.user_id)
        result = db.execute(stmt).all()
        print("\n左连接结果：")
        for user, article in result:  # type: ignore[misc]
            print(f"用户：{user.username}，文章：{article.title if article else '无'}")
            # 输出:
            # 用户：zhangsan，文章：python 入门
            # 用户：zhangsan，文章：sqlalchemy 进阶
            # 用户：lisi，文章：mysql 优化
            # 用户：wangwu，文章：无
            # 用户：zhaoliu，文章：无
    finally:
        db.close()

# ====================== 1.5 子查询（2.0+ 优化版） ======================
def sub_query():
    """子查询（2.0+ scalar_subquery/subquery 规范）"""
    db: session = get_db_session()
    try:
        # 示例1：查询有文章的用户
        # 子查询：所有发布过文章的用户id（去重）
        subquery = select(distinct(article.user_id)).scalar_subquery()
        # 主查询
        stmt = select(user).where(user.id.in_(subquery))
        users = db.scalars(stmt).all()
        print("\n发布过文章的用户：", users)  # 输出: [<user(id=1, username='zhangsan', email='zhangsan@example.com')>, <user(id=2, username='lisi', email='lisi@example.com')>]
        # 示例2：查询年龄大于平均年龄的用户
        # 子查询：平均年龄
        subquery = select(func.avg(user.age)).scalar_subquery()
        # 主查询
        stmt = select(user).where(user.age > subquery)
        users = db.scalars(stmt).all()
        print("年龄大于平均年龄的用户：", users)  # 输出: [<user(id=3, username='wangwu', email='wangwu@example.com')>, <user(id=4, username='zhaoliu', email='none')>]
        # 示例3：查询发布过2篇以上文章的用户
        # 子查询：统计用户文章数（>2）
        subquery = (
            select(article.user_id, func.count(article.id).label("article_count"))
            .group_by(article.user_id)
            .having(func.count(article.id) > 2)
            .subquery()
        )
        # 主查询：关联子查询
        stmt = select(user).join(subquery, user.id == subquery.c.user_id)
        users = db.scalars(stmt).all()
        print("发布过2篇以上文章的用户：", users)  # 输出: []
    finally:
        db.close()

# ====================== 1.6 分页查询（2.0+ 优化版） ======================
def pagination_query():
    """分页查询（2.0+ 规范写法，带总页数计算）"""
    db: session = get_db_session()
    try:
        # 分页参数
        page = 1  # 当前页
        page_size = 2  # 每页条数
        offset = (page - 1) * page_size
        # 1. 分页查询数据
        stmt = select(user).offset(offset).limit(page_size).order_by(user.id)
        users: sequence[user] = db.scalars(stmt).all()
        print(f"\n第{page}页用户：", users)  # 输出: 第1页用户： [<user(id=1, username='zhangsan', email='zhangsan@example.com')>, <user(id=2, username='lisi', email='lisi@example.com')>]
        # 2. 计算总条数和总页数（优化：一次查询获取总条数）
        total_count_stmt = select(func.count(user.id))
        total_users: int = db.scalar(total_count_stmt)
        total_pages = (total_users + page_size - 1) // page_size  # 向上取整
        print(f"总用户数：{total_users}，总页数：{total_pages}")  # 输出: 总用户数：4，总页数：2
        # 扩展：分页通用函数
        def paginate_query(query_stmt, page_num: int, page_size_num: int):  # type: ignore[no-untyped-def]
            """通用分页函数"""
            offset_num = (page_num - 1) * page_size_num
            # 查询数据
            query_data = db.scalars(query_stmt.offset(offset_num).limit(page_size_num)).all()
            # 查询总条数
            query_total = db.scalar(select(func.count()).select_from(query_stmt.subquery()))
            query_total_pages = (query_total + page_size_num - 1) // page_size_num if query_total else 0
            return query_data, query_total or 0, query_total_pages
        # 测试通用分页函数
        data, total, pages = paginate_query(select(user), page_num=2, page_size_num=2)
        print(f"第 2 页用户：{data}，总条数：{total}，总页数：{pages}")  # 输出: 第 2 页用户：[<user(id=3, username='wangwu', email='wangwu@example.com')>, <user(id=4, username='zhaoliu', email='none')>]，总条数：4，总页数：2
    finally:
        db.close()
# ====================== 执行所有测试 ======================
if __name__ == "__main__":
    # 初始化测试数据（仅首次执行）
    init_data()
    # 执行筛选查询
    filter_query()
    # 执行聚合查询
    aggregate_query()
    # 执行关联查询
    join_query()
    # 执行子查询
    sub_query()
    # 执行分页查询
    pagination_query()

from typing import list, optional
from sqlalchemy import (
    column, integer, string, foreignkey, boolean, table, func, datetime
)
from sqlalchemy.orm import (
    session, declarativebase, relationship, mapped, mapped_column
)
# 假设从优化后的配置导入会话和引擎
from sync_crud import get_db_session
from sync_session import sync_engine
# ====================== 基础基类（2.0+ 规范） ======================
class base(declarativebase):
    """所有模型的基类（2.0+ 推荐 declarativebase）"""
    __abstract__ = true
    # 通用字段：创建/更新时间（所有表复用）
    create_time = mapped_column(datetime, default=func.now(), nullable=false, comment="创建时间")
    update_time = mapped_column(datetime, default=func.now(), onupdate=func.now(), nullable=false, comment="更新时间")
# ====================== 2.2 一对多（用户 - 文章） ======================
# 推荐：先定义被关联模型（article），避免循环引用；或使用字符串引用
class article(base):
    __tablename__ = "articles"
    # 2.0+ 推荐使用 mapped_column 替代旧 column 写法（类型注解更友好）
    id: mapped[int] = mapped_column(integer, primary_key=true, autoincrement=true, index=true, comment="文章id")
    title: mapped[str] = mapped_column(string(100), nullable=false, index=true, comment="文章标题")
    content: mapped[optional[str]] = mapped_column(string(500), comment="文章内容")
    user_id: mapped[int] = mapped_column(integer, foreignkey("users.id", ondelete="cascade"), comment="关联用户id")
    # 多对一关联：多个文章属于一个用户
    author: mapped["user"] = relationship(
        "user",
        back_populates="articles",
        lazy="selectin"  # 2.0+ 推荐的加载策略，避免n+1查询
    )
    def __repr__(self) -> str:
        return f"<article(id={self.id}, title='{self.title}', user_id={self.user_id})>"
# ====================== 2.4 多对多（用户 - 角色） ======================
# 中间表（2.0+ 规范写法）
user_role = table(
    "user_role",
    base.metadata,
    column("user_id", integer, foreignkey("users.id", ondelete="cascade"), primary_key=true, comment="用户id"),
    column("role_id", integer, foreignkey("roles.id", ondelete="cascade"), primary_key=true, comment="角色id"),
    # 新增中间表通用字段（可选）
    column("create_time", datetime, default=func.now(), nullable=false, comment="关联创建时间")
)
class role(base):
    __tablename__ = "roles"
    id: mapped[int] = mapped_column(integer, primary_key=true, autoincrement=true, index=true, comment="角色id")
    name: mapped[str] = mapped_column(string(50), unique=true, nullable=false, comment="角色名称")
    # 多对多关联：一个角色对应多个用户
    users: mapped[list["user"]] = relationship(
        "user",
        secondary=user_role,  # 指定中间表
        back_populates="roles",
        cascade="all",
        lazy="selectin"
    )
class user(base):
    __tablename__ = "users"
    # 2.0+ 新写法：mapped + mapped_column（类型提示更精准）
    id: mapped[int] = mapped_column(integer, primary_key=true, autoincrement=true, index=true, comment="用户id")
    username: mapped[str] = mapped_column(string(50), unique=true, nullable=false, index=true, comment="用户名")
    email: mapped[optional[str]] = mapped_column(string(100), unique=true, index=true, comment="邮箱")
    is_active: mapped[bool] = mapped_column(boolean, default=true, comment="是否激活")
    # 一对多关联：一个用户有多个文章
    articles: mapped[list["article"]] = relationship(
        "article",
        back_populates="author",
        cascade="all, delete-orphan",  # 级联删除：删用户时删文章，删文章自动解除关联
        passive_deletes=true,  # 配合外键 ondelete="cascade"，提升删除性能
        lazy="selectin"  # 预加载关联数据，避免n+1查询
    )
    # 一对一关联：uselist=false 表示非列表（一对一）
    profile: mapped[optional["userprofile"]] = relationship(
        "userprofile",
        back_populates="user",
        uselist=false,
        cascade="all, delete-orphan",
        passive_deletes=true,
        lazy="selectin"
    )
    # 多对多关联：一个用户对应多个角色
    roles: mapped[list["role"]] = relationship(
        "role",
        secondary=user_role,
        back_populates="users",
        cascade="all",
        lazy="selectin"
    )
    def __repr__(self) -> str:
        return f"<user(id={self.id}, username='{self.username}', email='{self.email}')>"
# ====================== 2.3 一对一（用户 - 用户详情） ======================
class userprofile(base):
    __tablename__ = "user_profiles"
    id: mapped[int] = mapped_column(integer, primary_key=true, autoincrement=true, comment="详情id")
    address: mapped[optional[str]] = mapped_column(string(200), comment="地址")
    phone: mapped[optional[str]] = mapped_column(string(20), comment="手机号")
    # 唯一外键：保证一对一关系
    user_id: mapped[int] = mapped_column(
        integer,
        foreignkey("users.id", ondelete="cascade"),
        unique=true,
        comment="关联用户id"
    )
    # 一对一关联：一个详情属于一个用户
    user: mapped["user"] = relationship(
        "user",
        back_populates="profile",
        lazy="selectin"
    )
# ====================== 2.1 工具函数：初始化表和测试数据 ======================
def init_relationship_tables():
    """初始化关系表（删除旧表+创建新表）"""
    db: session = get_db_session()
    try:
        # 先删除旧表（按依赖顺序）
        base.metadata.drop_all(bind=sync_engine)
        # 创建新表
        base.metadata.create_all(bind=sync_engine)
        print("关系表初始化完成")
    finally:
        db.close()

# ====================== 2.2 一对多使用示例 ======================
def one_to_many():
    """测试一对多关系（修复pycharm报错+规范写法）"""
    db: session = get_db_session()
    try:
        # 1. 新增用户+文章（修复pycharm参数报错：使用类型注解+分步创建）
        # 方式1：推荐写法（分步创建，无ide报错）
        new_user = user(username="tianqi", email="tianqi@example.com")
        # 创建文章并关联作者（推荐）
        article1 = article(title="sqlalchemy关系映射", content="一对多示例", author=new_user)
        article2 = article(title="pythonorm", content="orm入门", author=new_user)
        db.add_all([new_user, article1, article2])
        db.flush()  # 预提交，生成id
        db.commit()
        print("新增用户+文章完成")
        # 2. 查询用户的所有文章（2.0+ select 构造器）
        from sqlalchemy import select
        stmt = select(user).where(user.username == "tianqi")
        user = db.scalars(stmt).first()
        if user:
            print(f"\n{user.username}的文章：", [art.title for art in user.articles])   # 输出: tianqi的文章： ['sqlalchemy关系映射', 'pythonorm']
        # 3. 查询文章的作者
        stmt = select(article).where(article.title == "sqlalchemy关系映射")
        article = db.scalars(stmt).first()
        if article:
            print(f"{article.title}的作者：", article.author.username)  # 输出: sqlalchemy关系映射的作者： tianqi
    except exception as e:
        db.rollback()
        raise runtimeerror(f"一对多测试失败：{str(e)}") from e
    finally:
        db.close()

# ====================== 2.3 一对一使用示例） ======================
def one_to_one():
    """测试一对一关系"""
    db: session = get_db_session()
    try:
        # 1. 查询用户
        from sqlalchemy import select
        stmt = select(user).where(user.username == "tianqi")
        user = db.scalars(stmt).first()
        if user:
            # 2. 新增/更新用户详情
            if user.profile:
                db.delete(user.profile)
                db.flush()
            user.profile = userprofile(address="北京市海淀区", phone="13800138000")
            db.commit()
            # 输出: tianqi的地址： 北京市海淀区
            print(f"\n{user.username}的地址：", user.profile.address)  # type: ignore
            # 输出: tianqi的手机号： 13800138000
            print(f"{user.username}的手机号：", user.profile.phone)  # type: ignore
    except exception as e:
        db.rollback()
        raise runtimeerror(f"一对一测试失败：{str(e)}") from e
    finally:
        db.close()

# ====================== 2.4 多对多使用示例（优化版） ======================
def many_to_many():
    """测试多对多关系"""
    db: session = get_db_session()
    try:
        # 1. 新增角色
        admin_role = role(name="admin")
        guest_role = role(name="guest")
        db.add_all([admin_role, guest_role])
        db.commit()
        # 2. 给用户分配角色
        from sqlalchemy import select
        stmt = select(user).where(user.username == "tianqi")
        user = db.scalars(stmt).first()
        if user:
            user.roles.append(admin_role)
            user.roles.append(guest_role)
            db.commit()
            # 3. 查询用户的角色
            print(f"\n{user.username}的角色：", [role.name for role in user.roles]) # 输出: tianqi的角色： ['admin', 'guest']
            # 4. 查询角色下的用户
            stmt = select(role).where(role.name == "admin")
            role = db.scalars(stmt).first()
            if role:
                print(f"{role.name}角色下的用户：", [u.username for u in role.users])  # 输出: admin角色下的用户： ['tianqi']
    except exception as e:
        db.rollback()
        raise runtimeerror(f"多对多测试失败：{str(e)}") from e
    finally:
        db.close()
# ====================== 执行所有测试 ======================
if __name__ == "__main__":
    # 1. 初始化表
    init_relationship_tables()
    # 2. 测试一对多
    one_to_many()
    # 3. 测试一对一
    one_to_one()
    # 4. 测试多对多
    many_to_many()

from sqlite3 import row
from typing import list, tuple, optional, sequence
from sqlalchemy import integer, string, text
from sqlalchemy.engine import result  # 2.0+ 结果类型注解
from sqlalchemy.orm import session, declarativebase, mapped, mapped_column
# 定义数据库引擎/会话
from sync_crud import get_db_session
from sync_session import sync_engine
# ====================== 基础配置 ======================
class base(declarativebase):
    __abstract__ = true
# ====================== user 模型优化 ======================
class user(base):
    __tablename__ = "users"
    # 2.0+ 新写法：mapped + mapped_column（类型提示更精准）
    id: mapped[int] = mapped_column(integer, primary_key=true, autoincrement=true, index=true, comment="用户id")
    username: mapped[str] = mapped_column(string(50), unique=true, nullable=false, index=true, comment="用户名")
    email: mapped[optional[str]] = mapped_column(string(100), unique=true, index=true, comment="邮箱")
    age: mapped[int] = mapped_column(integer, default=0, nullable=false, comment="年龄")
    balance: mapped[int] = mapped_column(integer, default=0, nullable=false, comment="余额")
    def __repr__(self) -> str:
        return f"<user(id={self.id}, username='{self.username}', email='{self.email}', age={self.age}, balance={self.balance})>"
# ====================== 工具函数：初始化表和测试数据 ======================
def init_tables():
    """初始化关系表（删除旧表+创建新表）"""
    db: session = get_db_session()
    try:
        # 先删除旧表（按依赖顺序）
        base.metadata.drop_all(bind=sync_engine)
        # 创建新表
        base.metadata.create_all(bind=sync_engine)
        print("关系表初始化完成")
    finally:
        db.close()
# ====================== 核心：原生sql操作（2.0+ 优化版） ======================
def execute_sql() -> none:
    """执行原生sql操作"""
    db: session = get_db_session()  # 获取会话（封装为可复用函数）
    try:
        # 1. 单条插入数据（获取插入后的自增id）
        single_insert_sql = text("""
                    insert into users (username, email, age, balance) 
                    values (:username, :email, :age, :balance)
                """)
        single_insert_data = {"username": "wushi", "email": "wushi@example.com", "age": 30, "balance": 100}
        # mysql 下获取自增id（不同数据库语法不同）
        if sync_engine.dialect.name == "mysql":
            db.execute(single_insert_sql, single_insert_data)
            last_id = db.execute(text("select last_insert_id()")).scalar()
            db.commit()
            print(f"单条插入完成，自增id：{last_id}")
        # 2. 批量插入数据（2.0+ 优化：executemany 模式，性能更高）
        batch_insert_sql = text("""
            insert into users (username, email, age, balance) 
            values (:username, :email, :age, :balance)
        """)
        batch_data: list[dict] = [
            {"username": "zhangsan", "email": "zhangsan@example.com", "age": 20, "balance": 100},
            {"username": "lisi", "email": "lisi@example.com", "age": 22, "balance": 100},
            {"username": "wangwu", "email": "wangwu@example.com", "age": 25, "balance": 100}
        ]
        # 2.0+ 推荐使用 executemany_mode="values" 优化批量插入
        result: result = db.execute(
            batch_insert_sql,
            batch_data,
            execution_options={"executemany_mode": "values"}  # 批量插入优化
        )
        db.commit()
        # 批量创建数据完成，影响行数：3
        print(f"批量创建数据完成，影响行数：{result.rowcount}")  # type: ignore[attr-defined]
        # 3. 执行查询
        query_sql = text("select username, age from users where age > :age")
        query_result: result = db.execute(query_sql, {"age": 20})
        all_results: sequence[row] = query_result.fetchall()
        print("原生 sql 查询结果：", all_results)  # 输出: [('lisi', 22), ('wangwu', 25)]
        # 4. 执行单条查询（返回单个元组，2.0+ 空值处理优化）
        single_sql = text("select username from users where id = :id")
        single_result: optional[tuple[str]] = db.execute(single_sql, {"id": 1}).fetchone()
        if single_result:
            print("单个结果：", single_result[0])  # 输出: zhangsan
        else:
            print("未找到 id 为 1 的用户")
        # 5. 执行更新操作
        update_sql = text("update users set age = :age where username = :username")
        update_result: result = db.execute(update_sql, {"age": 23, "username": "zhangsan"})
        db.commit()
        # 使用rowcount属性
        update_row_count = update_result.rowcount  # type: ignore[attr-defined]
        print(f"原生 sql 修改完成，影响行数：{update_row_count}")  # 输出: 1
    except exception as e:
        db.rollback()  # 异常回滚，保证数据一致性
        raise runtimeerror(f"原生sql操作失败：{str(e)}") from e  # 保留异常栈
    finally:
        db.close()  # 确保会话关闭（2.0+ 同步会话必须显式关闭）
# ====================== 执行入口 ======================
if __name__ == "__main__":
    # 1. 初始化表
    init_tables()
    # 2. 执行原生sql操作
    execute_sql()

from sqlalchemy import select
from sqlalchemy.exc import sqlalchemyerror
# ====================== 核心：转账事务（2.0+ 优化版） ======================
def transfer_money(
        db: session,
        from_username: str,
        to_username: str,
        amount: int
) -> optional[tuple[bool, str]]:
    """
    安全的转账事务操作（2.0+ 规范）
    :param db: 数据库会话
    :param from_username: 转出用户名
    :param to_username: 转入用户名
    :param amount: 转账金额（正数）
    :return: (是否成功, 提示信息)
    """
    # 前置校验：金额合法性
    if amount <= 0:
        return false, "转账金额必须大于0"
    # 模拟异常
    if amount == 100:
        raise exception("模拟异常")
    try:
        # 1. 2.0+ 推荐使用 select 构造器（替代旧 query()）
        stmt_a = select(user).where(user.username == from_username)
        stmt_b = select(user).where(user.username == to_username)
        # 2. 悲观锁：锁定行，避免并发修改（关键！防止超卖/余额不一致）
        # for update：2.0+ 兼容，锁定查询到的行直到事务结束
        user_a: optional[user] = db.scalars(stmt_a.with_for_update()).first()
        user_b: optional[user] = db.scalars(stmt_b.with_for_update()).first()
        # 3. 业务校验：用户存在性 + 余额充足性
        if not user_a:
            return false, f"转出用户 {from_username} 不存在"
        if not user_b:
            return false, f"转入用户 {to_username} 不存在"
        if user_a.balance < amount:
            return false, f"用户 {from_username} 余额不足（当前：{user_a.balance}，需转出：{amount}）"
        # 4. 原子性更新操作
        user_a.balance -= amount
        user_b.balance += amount
        # 5. 可选：批量更新（替代对象修改，性能更高）
        # 适用于高并发场景，直接执行update语句，减少orm对象操作
        # stmt_update_a = update(user).where(user.username == from_username).values(balance=user.balance - amount)
        # stmt_update_b = update(user).where(user.username == to_username).values(balance=user.balance + amount)
        # db.execute(stmt_update_a)
        # db.execute(stmt_update_b)
        # 6. 提交事务（2.0+ 同步提交，无需await）
        db.commit()
        # 7. 刷新对象，获取最新数据（可选）
        db.refresh(user_a)
        db.refresh(user_b)
        return true, (
            f"转账成功！\n"
            f"{from_username} 余额：{user_a.balance}（原：{user_a.balance + amount}）\n"
            f"{to_username} 余额：{user_b.balance}（原：{user_b.balance - amount}）"
        )
    except sqlalchemyerror as e:
        # 2.0+ 专用异常捕获：仅捕获数据库相关异常，避免捕获所有异常
        db.rollback()
        return false, f"转账失败（数据库异常）：{str(e)}"
    except exception as e:
        # 其他业务异常
        db.rollback()
        return false, f"转账失败（业务异常）：{str(e)}"
    finally:
        # 可选：关闭会话（若会话是函数内创建，否则由调用方管理）
        # db.close()
        pass
# ====================== 执行入口 ======================
if __name__ == "__main__":
    """执行原生sql操作"""
    db: session = get_db_session()  # 获取会话（封装为可复用函数）
    # 测试1：正常转账
    success, msg = transfer_money(db, "zhangsan", "lisi", 10)
    print("\n测试1 - 正常转账：", success, msg)
    # 输出:
    # 测试1 - 正常转账： true 转账成功！
    # zhangsan 余额：90（原：100）
    # lisi ：110（原：100）
    # 测试2：余额不足（触发回滚）
    success, msg = transfer_money(db, "zhangsan", "lisi", 1000)
    print("\n测试2 - 余额不足：", success, msg)
    # 输出: 测试2 - 余额不足： false 用户 zhangsan 余额不足（当前：90，需转出：1000）
    # 测试3：用户不存在（触发回滚）
    success, msg = transfer_money(db, "alice", "lisi", 10)
    print("\n测试3 - 用户不存在：", success, msg)
    # 输出: 测试3 - 用户不存在： false 转出用户 alice 不存在
    # 测试4：手动触发异常（模拟业务报错）
    # 可取消注释测试：在transfer_money中添加 raise exception("模拟异常")
    # success, msg = transfer_money(db, "zhangsan", "lisi", 100)
    # print("\n测试4 - 手动异常：", success, msg)

# ====================== 5.1 加载策略优化（解决n+1查询） ======================
def optimize_loading_strategy(db: session):
    """2.0+ 加载策略优化（joinedload/selectinload）"""
    print("=== 5.1 加载策略优化 ===")
    # 问题：n+1查询（2.0+ 仍存在，需显式优化）
    print("\n【n+1查询示例】")
    users: list[user] = db.scalars(select(user)).all()
    for user in users[:2]:  # 仅演示前2个用户
        # 访问articles时触发新查询（n+1问题）
        print(f"用户 {user.username} 的文章数：{len(user.articles)}")
    # 优化1：joinedload（左连接，一次性加载，适合一对一/一对多）
    print("\n【joinedload优化（左连接）】")
    stmt = select(user).options(joinedload(user.articles))
    users = db.scalars(stmt).all()
    for user in users[:2]:
        # 无额外查询，直接访问关联数据
        print(f"用户 {user.username} 的文章：{[art.title for art in user.articles[:1]]}")
    # 优化2：selectinload（in查询，适合多对多/大数据量一对多）
    print("\n【selectinload优化（in查询）】")
    stmt = select(user).options(selectinload(user.roles))
    users = db.scalars(stmt).all()
    for user in users[:2]:
        print(f"用户 {user.username} 的角色：{[role.name for role in user.roles]}")
    # 扩展：2.0+ 高级加载策略
    # 1. lazyload：强制延迟加载（覆盖模型默认）
    stmt = select(user).options(lazyload(user.articles))
    # 2. contains_eager：配合手动join，复用已有连接
    stmt = select(user).join(user.articles).options(contains_eager(user.articles))

# ====================== 5.2 只查询指定字段 ======================
def optimize_selected_fields(db: session):
    """字段筛选优化（减少数据传输）"""
    # 基础写法：只查询指定字段（返回元组）
    print("\n【基础字段筛选】")
    stmt = select(user.username, user.email)
    result: result = db.execute(stmt)
    user_fields: list[tuple[str, str]] = result.all()
    for username, email in user_fields[:3]:
        print(f"用户名：{username}，邮箱：{email}")
    # 进阶：映射为字典（更易用）
    print("\n【字段筛选+映射字典】")
    stmt = select(
        user.username.label("name"),
        user.age.label("user_age")
    ).where(user.age > 18)
    result = db.execute(stmt)
    # 转为字典列表（2.0+ result对象支持mappings()）
    user_dicts = [dict(row) for row in result.mappings()]
    for user in user_dicts[:3]:
        print(f"姓名：{user['name']}，年龄：{user['user_age']}")
    # 性能优化：count查询仅查主键（避免count(*)）
    print("\n【count优化】")
    total = db.scalar(select(func.count(user.id)))  # 比count(*)更快
    print(f"总用户数：{total}")

# user模型
class user(base):
    __tablename__ = "users"
    # 基础字段 + 单字段索引
    id: mapped[int] = mapped_column(integer, primary_key=true, autoincrement=true, index=true)
    username: mapped[str] = mapped_column(
        string(50), unique=true, nullable=false, index=true, comment="用户名（索引）"
    )
    email: mapped[str] = mapped_column(
        string(100), unique=true, nullable=false, index=true, comment="邮箱（唯一索引）"
    )
    age: mapped[int] = mapped_column(integer, default=0, comment="年龄")
    is_active: mapped[bool] = mapped_column(boolean, default=true, comment="是否激活")
    balance: mapped[int] = mapped_column(integer, default=0, comment="余额")
    # 关联关系（默认lazy="select"，即延迟加载）
    articles: mapped[list[article]] = relationship("article", back_populates="author")
    roles: mapped[list[role]] = relationship("role", secondary=user_role, back_populates="users")
    # 2.0+ 复合索引（生产级优化：覆盖高频查询条件）
    __table_args__ = (
        # 复合索引：age + is_active（匹配 where age > ? and is_active = ?）
        index("idx_age_active", "age", "is_active"),
        # 唯一索引（避免重复）
        index("idx_username_unique", "username", unique=true),
        # 部分索引（仅对激活用户生效，sqlite不支持，mysql/pg支持）
        # index("idx_active_email", "email", postgresql_where=(is_active == true)),
    )
# ====================== 5.3 索引优化（2.0+ 规范） ======================
def index_optimization_demo():
    """索引优化说明（模型定义规范）"""
    print("\n=== 5.3 索引优化 ===")
    print("✅ 单字段索引：username/email/index=true（高频查询字段）")
    print("✅ 复合索引：idx_age_active（匹配多字段查询条件）")
    print("✅ 唯一索引：username/email（避免重复数据）")
    print("✅ 部分索引：仅对激活用户生效（mysql/pg支持，sqlite不支持）")
    print("❌ 避免过度索引：更新频繁的字段（如balance）不建索引")

# ====================== 5.4 批量操作（2.0+ 性能优化） ======================
def batch_operations_optimization(db: session):
    """批量操作优化（减少数据库交互）"""
    # 1. 批量插入
    print("\n【批量插入】")
    batch_users = [
        user(username=f"user{i}", email=f"user{i}@example.com", age=18+i)
        for i in range(1, 4)
    ]
    db.add_all(batch_users)
    db.commit()
    print(f"批量插入 {len(batch_users)} 个用户完成")
    # 2. 批量更新
    print("\n【批量更新】")
    stmt = update(user).where(user.age < 20).values(is_active=false)
    result = db.execute(stmt)
    db.commit()
    print(f"批量更新 {result.rowcount} 条记录（年龄<20的用户设为非激活）")
    # 3. 批量删除
    print("\n【批量删除】")
    stmt = delete(user).where(user.is_active == false)
    result = db.execute(stmt)
    db.commit()
    print(f"批量删除 {result.rowcount} 条记录（非激活用户）")
    # 批量插入优化（executemany_mode）
    # 适用于mysql，转为 values (...), (...) 格式
    # db.execute(
    #     insert(user),
    #     [{"username": "u1", "email": "u1@com"}, ...],
    #     execution_options={"executemany_mode": "values"}
    # )

d:\workspaces\python\test\test01\blog
│   blog.db
│   core.py
│   crud.py
│   __init__.py

import datetime
from typing import list, optional, generator
# sqlalchemy 2.0 核心导入
from sqlalchemy import (
    boolean,
    datetime,
    foreignkey,
    integer,
    string,
    create_engine,
)
from sqlalchemy.exc import sqlalchemyerror
from sqlalchemy.orm import (
    declarativebase,
    mapped,
    mapped_column,
    relationship,
    sessionmaker,
    session,
)
# --------------------------- 基础配置 ---------------------------
class base(declarativebase):
    """所有模型的基类"""
    # 为所有模型添加通用的创建时间字段
    __abstract__ = true  # 抽象基类，不会创建表
    id: mapped[int] = mapped_column(integer, primary_key=true, index=true)
    create_time: mapped[datetime.datetime] = mapped_column(
        datetime, default=lambda: datetime.datetime.now(datetime.utc), comment="创建时间"
    )
# 数据库连接配置
database_url = "sqlite:///./blog.db"
engine = create_engine(
    database_url,
    # check_same_thread=false 是 sqlite 专属配置:
    # 允许同一个数据库连接被多个线程共享使用，适配 sqlalchemy 连接池的多线程场景；
    # 解除「连接必须和创建它的线程绑定」的限制，避免 programmingerror 报错；
    connect_args={
        "check_same_thread": false, # 关闭 sqlite 的线程检查机制
        "uri": true  # 启用uri模式
    },
    # 或使用文件锁保证写安全
    execution_options={"isolation_level": "serializable"},
    echo=true,  # 设为 true 可打印 sql 语句，便于调试
)
# 创建会话工厂
sessionlocal = sessionmaker(
    autocommit=false,
    autoflush=false,
    bind=engine,
    expire_on_commit=false,  # 提交后不立即过期对象
)
# --------------------------- 数据模型 ---------------------------
class role(base):
    """角色模型"""
    __tablename__ = "roles"
    name: mapped[str] = mapped_column(string(50), unique=true, index=true, comment="角色名称")
    description: mapped[optional[str]] = mapped_column(string(200), nullable=true, comment="角色描述")
    # 多对多关联用户
    users: mapped[list["user"]] = relationship(
        "user", secondary="user_roles", back_populates="roles"
    )
    def __repr__(self) -> str:
        return f"<role(id={self.id}, name={self.name})>"
class userrole(base):
    """用户-角色 多对多中间表"""
    __tablename__ = "user_roles"
    __table_args__ = {"comment": "用户角色关联表"}  # 表注释
    # 复合主键
    user_id: mapped[int] = mapped_column(
        foreignkey("users.id"), primary_key=true, comment="用户id"
    )
    role_id: mapped[int] = mapped_column(
        foreignkey("roles.id"), primary_key=true, comment="角色id"
    )
class user(base):
    """用户模型"""
    __tablename__ = "users"
    __table_args__ = {"comment": "用户表"}
    username: mapped[str] = mapped_column(string(50), unique=true, index=true, comment="用户名")
    email: mapped[str] = mapped_column(string(100), unique=true, index=true, comment="邮箱")
    password: mapped[str] = mapped_column(string(100), comment="密码（建议加密存储）")
    is_active: mapped[bool] = mapped_column(
        boolean, default=true, comment="是否激活"
    )
    age: mapped[optional[int]] = mapped_column(integer, nullable=true, comment="年龄")
    # 关联角色（多对多）
    roles: mapped[list[role]] = relationship(
        "role", secondary="user_roles", back_populates="users"
    )
    # 关联文章（一对多）
    articles: mapped[list["article"]] = relationship(
        "article", back_populates="author", cascade="all, delete-orphan"
    )
    # 关联评论（一对多）
    comments: mapped[list["comment"]] = relationship(
        "comment", back_populates="author", cascade="all, delete-orphan"
    )
    def __repr__(self) -> str:
        return f"<user(id={self.id}, username={self.username}, email={self.email})>"
class article(base):
    """文章模型"""
    __tablename__ = "articles"
    __table_args__ = {"comment": "文章表"}
    title: mapped[str] = mapped_column(string(100), index=true, comment="文章标题")
    content: mapped[str] = mapped_column(string, comment="文章内容")
    read_count: mapped[int] = mapped_column(
        integer, default=0, comment="阅读量"
    )
    user_id: mapped[int] = mapped_column(
        foreignkey("users.id"), comment="作者id"
    )
    # 关联作者
    author: mapped["user"] = relationship(
        "user", back_populates="articles"
    )
    # 关联评论（一对多，删除文章时级联删除评论）
    comments: mapped[list["comment"]] = relationship(
        "comment", back_populates="article", cascade="all, delete-orphan"
    )
    def __repr__(self) -> str:
        return f"<article(id={self.id}, title={self.title[:20]}...)>"
class comment(base):
    """评论模型"""
    __tablename__ = "comments"
    __table_args__ = {"comment": "评论表"}
    content: mapped[str] = mapped_column(string(500), comment="评论内容")
    user_id: mapped[int] = mapped_column(
        foreignkey("users.id"), comment="评论用户id"
    )
    article_id: mapped[int] = mapped_column(
        foreignkey("articles.id"), comment="关联文章id"
    )
    # 关联用户
    author: mapped["user"] = relationship(
        "user", back_populates="comments"
    )
    # 关联文章
    article: mapped["article"] = relationship(
        "article", back_populates="comments"
    )
    def __repr__(self) -> str:
        return f"<comment(id={self.id}, content={self.content[:20]}...)>"
# --------------------------- 数据库操作工具函数 ---------------------------
def create_tables() -> none:
    """创建所有表"""
    try:
        base.metadata.create_all(bind=engine)
        print("数据库表创建成功！")
    except sqlalchemyerror as ex:
        print(f"创建表失败：{ex}")
        raise
def get_db() -> generator[session, none, none]:
    """
    获取数据库会话（依赖注入风格）
    2.0 版本推荐使用 session 类型提示
    """
    db_session: session = sessionlocal()
    try:
        yield db_session
    except sqlalchemyerror as ex:
        db_session.rollback()  # 异常时回滚
        print(f"数据库操作异常：{ex}")
        raise
    finally:
        db_session.close()
# --------------------------- 测试代码 ---------------------------
if __name__ == "__main__":
    # 创建所有表
    create_tables()
    # 测试代码
    # try:
    #     # with语句自动管理会话，结束后自动关闭
    #     with sessionlocal() as db:
    #         print("数据库会话创建成功！")
    #
    #         admin_role = db.query(role).filter(role.name == "admin").first()
    #         if not admin_role:
    #             admin_role = role(name="admin", description="管理员")
    #             db.add(admin_role)
    #             db.commit()
    #             db.refresh(admin_role)
    #             print(f"添加测试角色成功：{admin_role}")
    #         else:
    #             print(f"admin角色已存在：{admin_role}")
    # except exception as e:
    #     print(f"测试过程出错：{e}")

-- 插入角色
insert into roles (name, description) values 
('admin', '系统管理员，拥有所有权限'),
('user', '普通用户，只能操作自己的内容');
-- 插入管理员用户（密码：123456，实际项目需加密）
insert into users (username, email, password, is_active) values 
('admin', 'admin@example.com', '123456', 1);
-- 给管理员分配角色
insert into user_role (user_id, role_id) values (1, 1);

import os
import sys
from dataclasses import dataclass
from typing import dict, optional, any, sequence
# 添加项目根目录到 python 路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# sqlalchemy 2.0 核心导入
from sqlalchemy import (
    select,
    func,
    delete,
    update,
    or_
)
from sqlalchemy.engine import result
from sqlalchemy.exc import (
    sqlalchemyerror,
    integrityerror
)
from sqlalchemy.orm import (
    session,
    joinedload,
    selectinload
)
# 导入模型和数据库会话
from blog.core import user, role, article, comment, get_db
# --------------------------- 数据结构定义 ---------------------------
@dataclass
class paginationresult:
    """分页结果数据结构"""
    items: sequence[any]
    total: int
    page: int
    page_size: int
    total_pages: int
    def to_dict(self) -> dict[str, any]:
        """转换为字典格式"""
        return {
            "items": [item.__dict__ for item in self.items],  # 可替换为自定义序列化逻辑
            "total": self.total,
            "page": self.page,
            "page_size": self.page_size,
            "total_pages": self.total_pages
        }
# --------------------------- 异常定义 ---------------------------
class businessexception(exception):
    """业务异常基类"""
    def __init__(self, message: str, code: int = 400):
        self.message = message
        self.code = code
        super().__init__(self.message)
class resourceexistsexception(businessexception):
    """资源已存在异常"""
    def __init__(self, message: str):
        super().__init__(message, 409)
class resourcenotfoundexception(businessexception):
    """资源不存在异常"""
    def __init__(self, message: str):
        super().__init__(message, 404)
class permissiondeniedexception(businessexception):
    """权限不足异常"""
    def __init__(self, message: str):
        super().__init__(message, 403)
# --------------------------- 通用工具函数 ---------------------------
def commit_with_retry(db: session, max_retries: int = 3) -> none:
    """
    提交事务并支持重试（处理并发冲突）
    """
    retries = 0
    while retries < max_retries:
        try:
            db.commit()
            return
        except sqlalchemyerror as ex:
            retries += 1
            db.rollback()
            if retries >= max_retries:
                raise businessexception(f"数据库操作失败：{str(ex)}")
def get_password_hash(password: str) -> str:
    """
    密码加密（生产环境必需）
    """
    return password
# --------------------------- 用户管理 ---------------------------
def create_user(
        db: session,
        username: str,
        email: str,
        password: str,
        age: optional[int] = none
) -> user:
    """
    创建用户（sqlalchemy 2.0 优化版）
    args:
        db: 数据库会话
        username: 用户名
        email: 邮箱
        password: 原始密码（自动加密）
        age: 年龄
    returns:
        创建的用户对象
    raises:
        resourceexistsexception: 用户名/邮箱已存在
        businessexception: 数据库操作失败
    """
    # 2.0 优化：一次查询检查用户名和邮箱，减少数据库交互
    stmt = select(user).where(
        or_(user.username == username, user.email == email)
    )
    existing_user: optional[user] = db.execute(stmt).scalar_one_or_none()
    if existing_user:
        if existing_user.username == username:
            raise resourceexistsexception("用户名已存在")
        else:
            raise resourceexistsexception("邮箱已存在")
    # 密码加密（生产环境必需）
    hashed_password = get_password_hash(password)
    # 创建用户
    new_user = user(
        username=username,
        email=email,
        password=hashed_password,
        is_active=true,
        age=age
    )
    # 分配普通用户角色（优化：使用 selectinload 减少n+1查询）
    user_role: optional[role] = db.execute(
        select(role).where(role.name == "user")
    ).scalar_one_or_none()
    if user_role:
        new_user.roles.append(user_role)
    try:
        db.add(new_user)
        commit_with_retry(db)
        db.refresh(new_user)
        # 刷新后加载角色信息
        db.execute(select(user).options(joinedload(user.roles)).where(user.id == new_user.id))
        return new_user
    except sqlalchemyerror as ex:
        db.rollback()
        raise businessexception(f"创建用户失败：{str(ex)}")
def get_user_by_username(db: session, username: str) -> optional[user]:
    """
    根据用户名查询用户（包含角色）
    args:
        db: 数据库会话
        username: 用户名
    returns:
        用户对象或none
    """
    stmt = select(user).options(
        selectinload(user.roles)  # 2.0 推荐：selectinload 性能优于 joinedload 多对多
    ).where(user.username == username)
    return db.execute(stmt).scalar_one_or_none()
def get_user_by_id(db: session, user_id: int) -> optional[user]:
    """
    根据id查询用户（包含角色）
    args:
        db: 数据库会话
        user_id: 用户id
    returns:
        用户对象或none
    """
    stmt = select(user).options(
        selectinload(user.roles)
    ).where(user.id == user_id)
    return db.execute(stmt).scalar_one_or_none()
def update_user(
        db: session,
        user_id: int,
        **kwargs
) -> user:
    """
    修改用户信息
    args:
        db: 数据库会话
        user_id: 用户id
        **kwargs: 要更新的字段（如 email, age, is_active 等）
    returns:
        更新后的用户对象
    raises:
        resourcenotfoundexception: 用户不存在
        businessexception: 更新失败
    """
    # 检查用户是否存在
    db_user = get_user_by_id(db, user_id)
    if not db_user:
        raise resourcenotfoundexception("用户不存在")
    # 过滤合法字段，防止更新敏感字段
    allowed_fields = {"username", "email", "age", "is_active"}
    update_fields = {k: v for k, v in kwargs.items() if k in allowed_fields}
    if not update_fields:
        return db_user
    # 2.0 优化：批量更新字段
    try:
        for key, value in update_fields.items():
            setattr(db_user, key, value)
        commit_with_retry(db)
        db.refresh(db_user)
        return db_user
    except integrityerror as ex:
        db.rollback()
        raise resourceexistsexception(f"更新失败：{str(ex)}")
    except sqlalchemyerror as ex:
        db.rollback()
        raise businessexception(f"更新用户失败：{str(ex)}")
def delete_user(db: session, user_id: int) -> bool:
    """
    删除用户（2.0 批量删除写法，无需先查询）
    args:
        db: 数据库会话
        user_id: 用户id
    returns:
        删除成功返回true
    raises:
        resourcenotfoundexception: 用户不存在
    """
    stmt = delete(user).where(user.id == user_id)
    result: result = db.execute(stmt)
    if result.rowcount == 0:  # type: ignore[attr-defined]
        raise resourcenotfoundexception("用户不存在")
    commit_with_retry(db)
    return true
# --------------------------- 文章管理 ---------------------------
def create_article(
        db: session,
        title: str,
        content: str,
        user_id: int
) -> article:
    """
    发布文章
    args:
        db: 数据库会话
        title: 文章标题
        content: 文章内容
        user_id: 作者id
    returns:
        文章对象
    raises:
        resourcenotfoundexception: 用户不存在
        businessexception: 创建失败
    """
    # 检查用户是否存在
    if not db.execute(select(user.id).where(user.id == user_id)).scalar():
        raise resourcenotfoundexception("用户不存在")
    new_article = article(
        title=title,
        content=content,
        user_id=user_id
    )
    try:
        db.add(new_article)
        commit_with_retry(db)
        db.refresh(new_article)
        return new_article
    except sqlalchemyerror as ex:
        db.rollback()
        raise businessexception(f"发布文章失败：{str(ex)}")
def get_article_list(
        db: session,
        page: int = 1,
        page_size: int = 10
) -> paginationresult:
    """
    分页查询文章列表（包含作者）
    args:
        db: 数据库会话
        page: 页码（默认1）
        page_size: 每页条数（默认10）
    returns:
        分页结果对象
    """
    # 参数校验
    page = max(1, page)
    page_size = max(1, min(100, page_size))  # 限制最大页大小
    offset = (page - 1) * page_size
    # 2.0 优化：使用 with_for_update(read=true) 避免脏读（可选）
    # 查询文章列表（优化：只加载需要的字段）
    stmt = select(article).options(
        joinedload(article.author).load_only(user.id, user.username, user.email)
    ).order_by(article.create_time.desc()).offset(offset).limit(page_size)
    articles: sequence[article] = db.execute(stmt).scalars().all()
    # 总数量（优化：避免子查询）
    total: int = db.execute(select(func.count(article.id))).scalar()
    total_pages = (total + page_size - 1) // page_size
    return paginationresult(
        items=articles,
        total=total,
        page=page,
        page_size=page_size,
        total_pages=total_pages
    )
def get_article_detail(db: session, article_id: int) -> article:
    """
    查询文章详情（包含作者、评论及评论作者）
    args:
        db: 数据库会话
        article_id: 文章id
    returns:
        文章对象
    raises:
        resourcenotfoundexception: 文章不存在
    """
    stmt = select(article).options(
        joinedload(article.author).load_only(user.id, user.username),
        joinedload(article.comments).joinedload(
            comment.author
        ).load_only(user.id, user.username)
    ).where(article.id == article_id)
    # 使用 unique() 方法处理连接集合导致的重复数据
    db_article: optional[article] = db.execute(stmt).unique().scalar_one_or_none()
    if not db_article:
        raise resourcenotfoundexception("文章不存在")
    # 阅读量+1（2.0 优化：使用 update 语句，无需查询后修改）
    db.execute(
        update(article)
        .where(article.id == article_id)
        .values(read_count=article.read_count + 1)
    )
    commit_with_retry(db)
    # 刷新阅读量
    db.refresh(db_article, attribute_names=["read_count"])
    return db_article
def update_article(
        db: session,
        article_id: int,
        user_id: int,
        **kwargs
) -> article:
    """
    修改文章（仅作者可修改）
    args:
        db: 数据库会话
        article_id: 文章id
        user_id: 操作人id
        **kwargs: 要更新的字段（title, content）
    returns:
        更新后的文章对象
    raises:
        resourcenotfoundexception: 文章/用户不存在
        permissiondeniedexception: 无权限
    """
    # 检查文章是否存在并验证权限
    stmt = select(article).where(article.id == article_id)
    db_article: optional[article] = db.execute(stmt).scalar_one_or_none()
    if not db_article:
        raise resourcenotfoundexception("文章不存在")
    if db_article.user_id != user_id:
        raise permissiondeniedexception("无权限修改该文章")
    # 过滤合法字段
    allowed_fields = {"title", "content"}
    update_fields = {k: v for k, v in kwargs.items() if k in allowed_fields}
    if not update_fields:
        return db_article
    try:
        for key, value in update_fields.items():
            setattr(db_article, key, value)
        commit_with_retry(db)
        db.refresh(db_article)
        return db_article
    except sqlalchemyerror as ex:
        db.rollback()
        raise businessexception(f"修改文章失败：{str(ex)}")
def delete_article(
        db: session,
        article_id: int,
        user_id: int
) -> bool:
    """
    删除文章（仅作者/管理员可删除）
    args:
        db: 数据库会话
        article_id: 文章id
        user_id: 操作人id
    returns:
        删除成功返回true
    raises:
        resourcenotfoundexception: 文章/用户不存在
        permissiondeniedexception: 无权限
    """
    # 检查文章是否存在
    db_article: optional[article] = db.execute(
        select(article).where(article.id == article_id)
    ).scalar_one_or_none()
    if not db_article:
        raise resourcenotfoundexception("文章不存在")
    # 检查用户和权限
    db_user = get_user_by_id(db, user_id)
    if not db_user:
        raise resourcenotfoundexception("操作用户不存在")
    is_admin = any(role.name == "admin" for role in db_user.roles)
    if db_article.user_id != user_id and not is_admin:
        raise permissiondeniedexception("无权限删除该文章")
    try:
        db.delete(db_article)
        commit_with_retry(db)
        return true
    except sqlalchemyerror as ex:
        db.rollback()
        raise businessexception(f"删除文章失败：{str(ex)}")
# --------------------------- 评论管理 ---------------------------
def create_comment(
        db: session,
        content: str,
        user_id: int,
        article_id: int
) -> comment:
    """
    新增评论
    args:
        db: 数据库会话
        content: 评论内容
        user_id: 评论者id
        article_id: 文章id
    returns:
        评论对象
    raises:
        resourcenotfoundexception: 用户/文章不存在
    """
    # 2.0 优化：一次查询检查用户和文章
    user_exists = db.execute(select(user.id).where(user.id == user_id)).scalar()
    article_exists = db.execute(select(article.id).where(article.id == article_id)).scalar()
    if not user_exists:
        raise resourcenotfoundexception("用户不存在")
    if not article_exists:
        raise resourcenotfoundexception("文章不存在")
    db_comment = comment(
        content=content,
        user_id=user_id,
        article_id=article_id
    )
    try:
        db.add(db_comment)
        commit_with_retry(db)
        db.refresh(db_comment)
        return db_comment
    except sqlalchemyerror as ex:
        db.rollback()
        raise businessexception(f"新增评论失败：{str(ex)}")
def delete_comment(
        db: session,
        comment_id: int,
        user_id: int
) -> bool:
    """
    删除评论（仅评论者/管理员可删除）
    args:
        db: 数据库会话
        comment_id: 评论id
        user_id: 操作人id
    returns:
        删除成功返回true
    raises:
        resourcenotfoundexception: 评论/用户不存在
        permissiondeniedexception: 无权限
    """
    # 检查评论是否存在
    db_comment: optional[comment] = db.execute(
        select(comment).where(comment.id == comment_id)
    ).scalar_one_or_none()
    if not db_comment:
        raise resourcenotfoundexception("评论不存在")
    # 检查用户和权限
    db_user = get_user_by_id(db, user_id)
    if not db_user:
        raise resourcenotfoundexception("操作用户不存在")
    is_admin = any(role.name == "admin" for role in db_user.roles)
    if db_comment.user_id != user_id and not is_admin:
        raise permissiondeniedexception("无权限删除该评论")
    try:
        db.delete(db_comment)
        commit_with_retry(db)
        return true
    except sqlalchemyerror as ex:
        db.rollback()
        raise businessexception(f"删除评论失败：{str(ex)}")
# --------------------------- 测试功能 ---------------------------
if __name__ == "__main__":
    # 获取数据库会话
    db_session: session = next(get_db())
    try:
        # 1. 初始化角色
        roles = db_session.execute(select(role).where(role.name.in_(["admin", "guest"]))).scalars().all()
        role_names = {role.name for role in roles}
        missing_roles = []
        if "admin" not in role_names:
            missing_roles.append(role(name="admin", description="系统管理员"))
        if "guest" not in role_names:
            missing_roles.append(role(name="guest", description="来宾用户"))
        if missing_roles:
            db_session.add_all(missing_roles)
            commit_with_retry(db_session)
            print("初始化角色成功")
        # 2. 创建测试用户
        try:
            user = create_user(
                db_session,
                username="test_user",
                email="test@example.com",
                password="123456",
                age=23
            )
            print(f"创建用户成功：{user.username} (id: {user.id})")  # 输出: 创建用户成功：test_user (id: 1)
        except resourceexistsexception as e:
            print(f"创建用户失败：{e.message}")
            # 获取已存在的用户
            user = get_user_by_username(db_session, "test_user")
            if not user:
                raise
        # 3. 发布测试文章
        article = create_article(
            db_session,
            title="我的第一篇博客",
            content="sqlalchemy 2.0 实战案例",
            user_id=user.id
        )
        print(f"发布文章成功：{article.title} (id: {article.id})")  # 输出: 发布文章成功：我的第一篇博客 (id: 1)
        # 4. 新增评论
        comment = create_comment(
            db_session,
            content="这篇文章写得很好！",
            user_id=user.id,
            article_id=article.id
        )
        print(f"新增评论成功：{comment.content} (id: {comment.id})")  # 输出: 新增评论成功：这篇文章写得很好！ (id: 1)
        # 5. 查询文章详情
        detail = get_article_detail(db_session, article.id)
        print("\n=== 文章详情 ===")
        print(f"标题：{detail.title}")  # 输出: 标题：我的第一篇博客
        print(f"作者：{detail.author.username}")  # 输出: 作者：test_user
        print(f"阅读量：{detail.read_count}")  # 输出: 阅读量：1
        print(f"评论数：{len(detail.comments)}")  # 输出: 评论数：1
        print(f"第一条评论：{detail.comments[0].content}")  # 输出: 第一条评论：这篇文章写得很好！
        print(f"评论作者：{detail.comments[0].author.username}")  # 输出: 评论作者：test_user
        # 6. 测试分页查询
        pagination = get_article_list(db_session, page=1, page_size=10)
        print(f"\n=== 分页查询 ===")
        print(f"总文章数：{pagination.total}")  # 输出: 总文章数：1
        print(f"当前页：{pagination.page}/{pagination.total_pages}")  # 输出: 当前页：1/1
    except businessexception as e:
        print(f"业务异常：{e.message} (代码：{e.code})")
    except exception as e:
        print(f"测试失败：{str(e)}")
        db_session.rollback()
    finally:
        # 关闭会话
        db_session.close()
        print("\n数据库会话已关闭")

d:\workspaces\python\test\test01\mall
│   core.py
│   crud.py
│   mall.db
│   test.py
│   __init__.py

# -*- coding: utf-8 -*-
"""
电商订单管理系统 - 核心配置与数据模型
========================================
功能：
    1. 定义sqlalchemy 2.0的基础模型（declarativebase）
    2. 配置数据库连接与会话工厂
    3. 定义业务数据模型（用户、商品、分类、订单、订单项）
    4. 实现通用分页数据结构
依赖：
    - sqlalchemy 2.0.48
    - python 3.13.11
创建时间：2026-03-22
作者：inuex
版本：v1.0
"""
import datetime
from dataclasses import dataclass
from typing import list, optional, dict, any, generator, sequence
from sqlalchemy import (
    create_engine,
    foreignkey,
    integer,
    string,
    float,
    boolean,
    datetime,
    text,
)
from sqlalchemy.exc import sqlalchemyerror
from sqlalchemy.orm import (
    declarativebase,
    session,
    sessionmaker,
    mapped,
    mapped_column,
    relationship,
)
# --------------------------- 基础配置 ---------------------------
class base(declarativebase):
    """所有模型的抽象基类"""
    __abstract__ = true
    id: mapped[int] = mapped_column(integer, primary_key=true, index=true)
    create_time: mapped[datetime.datetime] = mapped_column(
        datetime, default=datetime.datetime.now(datetime.utc), comment="创建时间"
    )
    update_time: mapped[datetime.datetime] = mapped_column(
        datetime, default=datetime.datetime.now(datetime.utc), onupdate=datetime.datetime.now(datetime.utc),
        comment="更新时间"
    )
    is_deleted: mapped[bool] = mapped_column(
        boolean, default=false, comment="是否删除（软删除）"
    )
# 数据库连接（sqlite，可替换为mysql/postgresql）
database_url = "sqlite:///./mall.db"
engine = create_engine(
    database_url,
    connect_args={"check_same_thread": false},
    echo=false,  # 设为true可打印sql语句，便于学习
)
sessionlocal = sessionmaker(autocommit=false, autoflush=false, bind=engine)
# --------------------------- 数据模型 ---------------------------
class user(base):
    """用户模型"""
    __tablename__ = "users"
    username: mapped[str] = mapped_column(string(50), unique=true, index=true, comment="用户名")
    phone: mapped[str] = mapped_column(string(20), unique=true, index=true, comment="手机号")
    email: mapped[str] = mapped_column(string(100), nullable=true, comment="邮箱")
    address: mapped[str] = mapped_column(string(200), nullable=true, comment="收货地址")
    balance: mapped[float] = mapped_column(float, default=0.0, comment="账户余额")
    # 关联订单（一对多）
    orders: mapped[list["order"]] = relationship(
        "order", back_populates="user", cascade="all, delete-orphan"
    )
    def __repr__(self) -> str:
        return f"<user(id={self.id}, username={self.username}, balance={self.balance})>"
class category(base):
    """商品分类模型"""
    __tablename__ = "categories"
    name: mapped[str] = mapped_column(string(50), unique=true, comment="分类名称")
    description: mapped[optional[str]] = mapped_column(text, nullable=true, comment="分类描述")
    # 关联商品（一对多）
    products: mapped[list["product"]] = relationship(
        "product", back_populates="category", cascade="all, delete-orphan"
    )
    def __repr__(self) -> str:
        return f"<category(id={self.id}, name={self.name})>"
class product(base):
    """商品模型"""
    __tablename__ = "products"
    name: mapped[str] = mapped_column(string(100), index=true, comment="商品名称")
    price: mapped[float] = mapped_column(float, comment="商品价格")
    stock: mapped[int] = mapped_column(integer, default=0, comment="库存数量")
    sales: mapped[int] = mapped_column(integer, default=0, comment="销量")
    description: mapped[optional[str]] = mapped_column(text, nullable=true, comment="商品描述")
    category_id: mapped[int] = mapped_column(foreignkey("categories.id"), comment="分类id")
    # 关联分类
    category: mapped["category"] = relationship("category", back_populates="products")
    # 关联订单项
    order_items: mapped[list["orderitem"]] = relationship(
        "orderitem", back_populates="product", cascade="all, delete-orphan"
    )
    def __repr__(self) -> str:
        return f"<product(id={self.id}, name={self.name}, price={self.price}, stock={self.stock})>"
class order(base):
    """订单模型"""
    __tablename__ = "orders"
    order_no: mapped[str] = mapped_column(string(32), unique=true, index=true, comment="订单编号")
    user_id: mapped[int] = mapped_column(foreignkey("users.id"), comment="用户id")
    total_amount: mapped[float] = mapped_column(float, default=0.0, comment="订单总金额")
    status: mapped[str] = mapped_column(string(20), default="pending",
                                        comment="订单状态：pending/paid/shipped/completed/cancelled")
    payment_time: mapped[optional[datetime.datetime]] = mapped_column(datetime, nullable=true, comment="支付时间")
    # 关联用户
    user: mapped["user"] = relationship("user", back_populates="orders")
    # 关联订单项
    items: mapped[list["orderitem"]] = relationship(
        "orderitem", back_populates="order", cascade="all, delete-orphan"
    )
    def __repr__(self) -> str:
        return f"<order(id={self.id}, order_no={self.order_no}, status={self.status}, total_amount={self.total_amount})>"
class orderitem(base):
    """订单项模型（订单-商品 多对多关联）"""
    __tablename__ = "order_items"
    order_id: mapped[int] = mapped_column(foreignkey("orders.id"), comment="订单id")
    product_id: mapped[int] = mapped_column(foreignkey("products.id"), comment="商品id")
    quantity: mapped[int] = mapped_column(integer, default=1, comment="购买数量")
    unit_price: mapped[float] = mapped_column(float, comment="购买时的单价")
    # 关联订单
    order: mapped["order"] = relationship("order", back_populates="items")
    # 关联商品
    product: mapped["product"] = relationship("product", back_populates="order_items")
    def __repr__(self) -> str:
        return f"<orderitem(id={self.id}, order_id={self.order_id}, product_id={self.product_id}, quantity={self.quantity})>"
# --------------------------- 分页数据结构 ---------------------------
@dataclass
class pagination:
    """通用分页返回结构"""
    items: sequence[any]
    total: int
    page: int
    page_size: int
    total_pages: int
    has_next: bool
    has_prev: bool
    def to_dict(self) -> dict[str, any]:
        """转换为字典格式"""
        return {
            "items": [i.__dict__ for i in self.items],
            "pagination": {
                "total": self.total,
                "page": self.page,
                "page_size": self.page_size,
                "total_pages": self.total_pages,
                "has_next": self.has_next,
                "has_prev": self.has_prev
            }
        }
# --------------------------- 核心业务逻辑 ---------------------------
def create_tables() -> none:
    """创建所有表"""
    base.metadata.create_all(bind=engine)
    print("数据库表创建成功！")
def get_db() -> generator[session, none, none]:
    """获取数据库会话"""
    db_session: session = sessionlocal()
    try:
        yield db_session
    except sqlalchemyerror as ex:
        db_session.rollback()
        raise ex
    finally:
        db_session.close()

# -*- coding: utf-8 -*-
"""
电商订单管理系统 - 数据操作层（crud）
=====================================
功能：
    1. 实现商品、订单、用户等核心业务的增删改查
    2. 封装分页查询、条件过滤、排序等通用查询逻辑
    3. 处理电商核心业务逻辑（订单创建、库存扣减、状态更新）
    4. 实现事务管理与异常处理
依赖：
    - core.py（数据模型与数据库配置）
    - sqlalchemy 2.0.48
创建时间：2026-03-22
作者：inuex
版本：v1.0
注意：
    1. 所有数据库操作需通过session会话执行，禁止直接操作数据库连接
    2. 订单创建包含事务逻辑，异常时自动回滚
    3. 分页查询默认最大页大小为50，防止大数据量查询
"""
import datetime
from typing import list, tuple, sequence
from sqlalchemy import (
    select,
    func,
    and_,
    desc,
    asc,
)
from sqlalchemy.exc import sqlalchemyerror
from sqlalchemy.orm import (
    session,
    joinedload,
    selectinload,
)
from mall.core import category, product, user, pagination, order, orderitem
# ====================== 商品管理 ======================
def create_product(
        db: session,
        name: str,
        price: float,
        category_id: int,
        stock: int = 0,
        description: str = none
) -> product:
    """创建商品"""
    # 检查分类是否存在
    category = db.execute(select(category).where(category.id == category_id)).scalar_one_or_none()
    if not category:
        raise valueerror(f"分类id {category_id} 不存在")
    db_product = product(
        name=name,
        price=price,
        stock=stock,
        description=description,
        category_id=category_id
    )
    db.add(db_product)
    db.commit()
    db.refresh(db_product)
    return db_product
def get_product_list(
        db: session,
        page: int = 1,
        page_size: int = 10,
        category_id: int = none,
        min_price: float = none,
        max_price: float = none,
        keyword: str = none,
        sort_by: str = "create_time",
        sort_order: str = "desc"
) -> pagination:
    """
    分页查询商品列表（支持多条件过滤、排序）
    :param db: 数据库会话
    :param page: 页码
    :param page_size: 每页条数
    :param category_id: 分类id（可选）
    :param min_price: 最低价格（可选）
    :param max_price: 最高价格（可选）
    :param keyword: 商品名称关键词（可选）
    :param sort_by: 排序字段（create_time/price/stock/sales）
    :param sort_order: 排序方向（asc/desc）
    :return: 分页结果
    """
    # 参数校验
    page = max(1, page)
    page_size = max(1, min(50, page_size))  # 限制最大页大小
    offset = (page - 1) * page_size
    # 构建查询条件
    conditions = [product.is_deleted == false]
    if category_id:
        conditions.append(product.category_id == category_id)
    if min_price is not none:
        conditions.append(product.price >= min_price)
    if max_price is not none:
        conditions.append(product.price <= max_price)
    if keyword:
        conditions.append(product.name.like(f"%{keyword}%"))
    # 构建排序规则
    sort_column = getattr(product, sort_by, product.create_time)
    sort_func = desc if sort_order.lower() == "desc" else asc
    # 构建查询
    stmt = select(product).options(
        joinedload(product.category).load_only(category.id, category.name)  # 关联加载分类（只加载必要字段）
    ).where(and_(*conditions)).order_by(sort_func(sort_column)).offset(offset).limit(page_size)
    # 执行查询
    products: sequence[product] = db.execute(stmt).scalars().all()
    # 查询总数（复用条件，不包含分页和排序）
    count_stmt = select(func.count(product.id)).where(and_(*conditions))
    total = db.execute(count_stmt).scalar()
    # 计算分页信息
    total_pages = (total + page_size - 1) // page_size
    has_next = page < total_pages
    has_prev = page > 1
    return pagination(
        items=products,
        total=total,
        page=page,
        page_size=page_size,
        total_pages=total_pages,
        has_next=has_next,
        has_prev=has_prev
    )
# ====================== 订单管理 ======================
def generate_order_no() -> str:
    """生成唯一订单编号"""
    return datetime.datetime.now().strftime("%y%m%d%h%m%s") + str(datetime.datetime.now().microsecond)[:4]
def create_order(
        db: session,
        user_id: int,
        product_items: list[tuple[int, int]]  # [(product_id, quantity), ...]
) -> order:
    """
    创建订单（包含事务管理、库存检查、金额计算）
    :param db: 数据库会话
    :param user_id: 用户id
    :param product_items: 商品列表 [(商品id, 数量), ...]
    :return: 创建的订单
    """
    if not product_items:
        raise valueerror("订单中至少包含一个商品")
    # 1. 检查用户是否存在
    user = db.execute(select(user).where(user.id == user_id)).scalar_one_or_none()
    if not user:
        raise valueerror(f"用户id {user_id} 不存在")
    # 2. 检查商品库存并计算总金额
    total_amount = 0.0
    order_items = []
    for product_id, quantity in product_items:
        if quantity <= 0:
            raise valueerror(f"商品id {product_id} 购买数量必须大于0")
        # 加锁查询商品（防止超卖）
        product = db.execute(
            select(product).where(and_(product.id == product_id, product.is_deleted == false)).with_for_update()
        ).scalar_one_or_none()
        if not product:
            raise valueerror(f"商品id {product_id} 不存在或已删除")
        if product.stock < quantity:
            raise valueerror(f"商品 {product.name} 库存不足（当前库存：{product.stock}，请求数量：{quantity}）")
        # 计算金额
        item_amount = product.price * quantity
        total_amount += item_amount
        # 创建订单项
        order_items.append(orderitem(
            product_id=product_id,
            quantity=quantity,
            unit_price=product.price
        ))
        # 扣减库存，增加销量
        product.stock -= quantity
        product.sales += quantity
    # 3. 创建订单
    order = order(
        order_no=generate_order_no(),
        user_id=user_id,
        total_amount=total_amount,
        status="pending",
        items=order_items
    )
    try:
        db.add(order)
        db.commit()
        db.refresh(order)
        # 关联加载订单项和商品信息
        db.execute(
            select(order).options(
                joinedload(order.items).joinedload(orderitem.product),
                joinedload(order.user)
            ).where(order.id == order.id)
        )
        return order
    except sqlalchemyerror as ex:
        db.rollback()
        raise ex
def get_order_list(
        db: session,
        page: int = 1,
        page_size: int = 10,
        user_id: int = none,
        status: str = none,
        start_time: datetime.datetime = none,
        end_time: datetime.datetime = none
) -> pagination:
    """
    分页查询订单列表（支持多条件过滤）
    """
    page = max(1, page)
    page_size = max(1, min(50, page_size))
    offset = (page - 1) * page_size
    # 构建条件
    conditions = [order.is_deleted == false]
    if user_id:
        conditions.append(order.user_id == user_id)
    if status:
        conditions.append(order.status == status)
    if start_time:
        conditions.append(order.create_time >= start_time)
    if end_time:
        conditions.append(order.create_time <= end_time)
    # 构建查询（关联加载用户和订单项）
    stmt = select(order).options(
        joinedload(order.user).load_only(user.id, user.username, user.phone),
        selectinload(order.items).joinedload(orderitem.product).load_only(product.id, product.name, product.price)
    ).where(and_(*conditions)).order_by(desc(order.create_time)).offset(offset).limit(page_size)
    orders = db.execute(stmt).scalars().all()
    # 查询总数
    count_stmt = select(func.count(order.id)).where(and_(*conditions))
    total = db.execute(count_stmt).scalar()
    total_pages = (total + page_size - 1) // page_size
    has_next = page < total_pages
    has_prev = page > 1
    return pagination(
        items=orders,
        total=total,
        page=page,
        page_size=page_size,
        total_pages=total_pages,
        has_next=has_next,
        has_prev=has_prev
    )
def update_order_status(
        db: session,
        order_id: int,
        status: str,
        payment_time: datetime.datetime = none
) -> order:
    """
    更新订单状态
    :param db: 数据库会话
    :param order_id: 订单id
    :param status: 新状态
    :param payment_time: 支付时间（仅status=paid时需要）
    :return: 更新后的订单
    """
    valid_status = ["pending", "paid", "shipped", "completed", "cancelled"]
    if status not in valid_status:
        raise valueerror(f"无效的订单状态，可选值：{valid_status}")
    db_order = db.execute(select(order).where(order.id == order_id)).scalar_one_or_none()
    if not db_order:
        raise valueerror(f"订单id {order_id} 不存在")
    # 如果是取消订单，恢复库存
    if status == "cancelled" and db_order.status != "cancelled":
        for i in db_order.items:
            db_product = db.execute(select(product).where(product.id == i.product_id)).scalar()
            if db_product:
                db_product.stock += i.quantity
                db_product.sales -= i.quantity
    # 更新订单状态
    db_order.status = status
    if status == "paid" and payment_time:
        db_order.payment_time = payment_time
    db.commit()
    db.refresh(db_order)
    return db_order

# -*- coding: utf-8 -*-
"""
电商订单管理系统 - 功能测试脚本
=================================
功能：
    1. 验证数据库表创建
    2. 测试核心crud功能（商品创建、订单生成、状态更新）
    3. 验证分页查询、条件过滤、事务回滚等高级特性
    4. 模拟电商业务流程（用户下单→库存扣减→订单状态更新）
测试场景：
    - 商品分类初始化
    - 商品创建与分页查询
    - 订单创建（含库存检查）
    - 订单状态更新与库存恢复（取消订单）
    - 多条件过滤的商品查询
依赖：
    - core.py + crud.py（核心业务层）
    - sqlalchemy 2.0.48
创建时间：2026-03-22
作者：inuex
版本：v1.0
运行方式：
    python -m mall.test
"""
import datetime
from sqlalchemy import (
    select,
)
from sqlalchemy.exc import sqlalchemyerror
from mall.core import create_tables, category, product, user, get_db
from mall.crud import create_product, create_order, get_product_list, update_order_status, get_order_list
# --------------------------- 测试代码 ---------------------------
def main():
    # 获取数据库会话
    db = next(get_db())
    try:
        # 1. 初始化分类
        electronics = db.execute(select(category).where(category.name == "电子产品")).scalar_one_or_none()
        if not electronics:
            electronics = category(name="电子产品", description="手机、电脑、平板等")
            db.add(electronics)
            db.commit()
        # 2. 创建测试商品
        iphone = db.execute(select(product).where(product.name == "iphone 15")).scalar_one_or_none()
        if not iphone:
            iphone = create_product(
                db,
                name="iphone 15",
                price=5999.0,
                category_id=electronics.id,
                stock=100,
                description="苹果15手机，128g版本"
            )
        macbook = db.execute(select(product).where(product.name == "macbook pro")).scalar_one_or_none()
        if not macbook:
            macbook = create_product(
                db,
                name="macbook pro",
                price=12999.0,
                category_id=electronics.id,
                stock=50,
                description="苹果笔记本电脑，m3芯片"
            )
        # 3. 创建测试用户
        test_user = db.execute(select(user).where(user.username == "test_buyer")).scalar_one_or_none()
        if not test_user:
            test_user = user(
                username="test_buyer",
                phone="13800138000",
                email="test@example.com",
                address="北京市海淀区",
                balance=20000.0
            )
            db.add(test_user)
            db.commit()
        # 4. 创建订单（购买2个iphone 15 + 1个macbook pro）
        try:
            order = create_order(
                db,
                user_id=test_user.id,
                product_items=[(iphone.id, 2), (macbook.id, 1)]
            )
            print(f"创建订单成功：{order}") # 输出: <order(id=1, order_no=202603221724427528, status=pending, total_amount=24997.0)>
            print(f"订单总金额：{order.total_amount}") # 输出: 24997.0
            print(f"订单项数量：{len(order.items)}") # 输出: 2
            # 打印订单项详情
            for item in order.items:
                print(f"  - 商品：{item.product.name}，数量：{item.quantity}，单价：{item.unit_price}")
                # 输出:
                # - 商品：iphone 15，数量：2，单价：5999.0
                # - 商品：macbook pro，数量：1，单价：12999.0
        except valueerror as e:
            print(f"创建订单失败：{e}") # 输出:
        # 5. 分页查询商品（测试多条件过滤）
        print("\n=== 分页查询商品（价格区间 5000-15000，按价格升序）===")
        product_pagination = get_product_list(
            db,
            page=1,
            page_size=5,
            min_price=5000.0,
            max_price=15000.0,
            sort_by="price",
            sort_order="asc"
        )
        print(f"总商品数：{product_pagination.total}") # 输出: 2
        print(f"当前页：{product_pagination.page}/{product_pagination.total_pages}") # 输出: 1/1
        print(f"是否有下一页：{product_pagination.has_next}") # 输出: false
        for product in product_pagination.items:
            print(f"  - {product.name}，价格：{product.price}，分类：{product.category.name}")
            # 输出:
            # - iphone 15，价格：5999.0，分类：电子产品
            # - macbook pro，价格：12999.0，分类：电子产品
        # 6. 分页查询订单
        print("\n=== 分页查询订单 ===")
        order_pagination = get_order_list(
            db,
            page=1,
            page_size=10,
            user_id=test_user.id
        )
        print(f"总订单数：{order_pagination.total}") # 输出: 1
        for order in order_pagination.items:
            print(f"  - 订单编号：{order.order_no}，状态：{order.status}，金额：{order.total_amount}")
            # 输出:
            # - 订单编号：202603221724427528，状态：pending，金额：24997.0
        # 7. 更新订单状态为已支付
        if order_pagination.items:
            updated_order = update_order_status(
                db,
                order_id=order_pagination.items[0].id,
                status="paid",
                payment_time=datetime.datetime.now(datetime.utc)
            )
            print(f"\n更新订单状态成功：{updated_order.order_no} -> {updated_order.status}") # 输出: 202603221724427528 -> paid
    except sqlalchemyerror as e:
        db.rollback()
        print(f"数据库错误：{e}")
    except valueerror as e:
        print(f"业务错误：{e}")
    finally:
        db.close()
        print("\n数据库会话已关闭")
if __name__ == "__main__":
    # 创建表
    create_tables()
    # 运行测试
    main()