SQLAlchemy 异步 ORM 实践

uv add sqlalchemy pgvector psycopg2-binary

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase
from .config import HOST, USER, PORT, PASSWORD
 
PG_DATABASE_URL = f'postgresql+asyncpg://{USER}:{PASSWORD}@{HOST}:{PORT}/{USER}'
 
"""
create_async_engine 创建实例
async_sessionmaker 异步会话工厂，用于创建数据库会话
AsyncSession 异步数据库会话类
DeclarativeBase ORM 声明式基类，用于定义数据模型
 
"""
 
SessionLocal = async_sessionmaker(
    bind=engine,               # 绑定到之前创建的异步引擎
    class_=AsyncSession,       # 指定使用异步会话类
    expire_on_commit=False     # 重要设置！
)
"""
expire_on_commit=False 的含义：
默认情况（True）：事务提交后，所有对象的属性都会过期，下次访问时需要重新从数据库加载
 
设为 False：提交后对象保持"新鲜"，可以直接访问其属性
 
"""
class Base(DeclarativeBase):
    pass
"""
 
#所有数据模型都继承自这个基类
 
"""
 
# 依赖注入函数
async def get_db():
    async with SessionLocal() as session:
        yield session

# ==================== 导入部分 ====================
from datetime import datetime
from typing import Optional
 
# SQLAlchemy 数据类型
from sqlalchemy import (
    TEXT,           # 长文本
    TIMESTAMP,      # 时间戳
    Boolean,        # 布尔值
    String,         # 字符串
    Integer,        # 整数
    BigInteger,     # 大整数
    ForeignKey      # 外键
)
 
# ORM 核心
from sqlalchemy.orm import (
    Mapped,         # 类型注解
    mapped_column,  # 列定义
    relationship    # 表关系
)
 
# SQL 函数
from sqlalchemy.sql import func  # 数据库函数，如 NOW()
 
# ==================== 实体类定义 ====================
class GirlfriendConfig(Base):
    """
    女友配置实体类
    对应数据库表：girlfriend_configs
    """
 
    # 1. 表名定义
    __tablename__ = "girlfriend_configs"
 
    # 2. 表级参数（可选）
    __table_args__ = {
        'comment': '用户女友配置表',      # 表注释
        'mysql_charset': 'utf8mb4',      # MySQL 字符集
        # 'extend_existing': True         # 如果表已存在，扩展它
    }
 
    # ========== 字段定义 ==========
 
    # 3. 主键字段
    id: Mapped[int] = mapped_column(
        Integer,                    # 数据类型：整数
        primary_key=True,           # ✅ 这是主键
        autoincrement=True,         # ✅ 自动递增（1,2,3...）
        comment="配置ID"             # ✅ 字段注释
    )
 
    # 4. 外键字段（关联用户表）
    user_id: Mapped[int] = mapped_column(
        BigInteger,                 # 数据类型：大整数（适合用户ID）
        ForeignKey("users.id", ondelete="CASCADE"),  # ✅ 外键，删除时级联
        nullable=False,             # ✅ 不允许为NULL
        index=True,                 # ✅ 创建索引（提升查询速度）
        comment="关联的用户ID"
    )
 
    # 5. 字符串字段
    nickname: Mapped[str] = mapped_column(
        String(50),                 # ✅ 最大50字符的字符串
        nullable=False,             # ✅ 必填字段
        default="默认昵称",          # ✅ Python默认值（应用层）
        comment="女友昵称"
    )
 
    # 6. 长文本字段
    personality: Mapped[Optional[str]] = mapped_column(
        # Optional[str] 表示这个字段可以是 str 或 None
        TEXT,                       # ✅ 长文本类型
        nullable=True,              # ✅ 允许为NULL
        comment="性格描述，支持长文本"
    )
 
    # 7. 时间字段 - 创建时间
    created_at: Mapped[datetime] = mapped_column(
        TIMESTAMP(timezone=True),   # ✅ 带时区的时间戳
        server_default=func.now(),  # ✅ 数据库默认值（插入时自动设置）
        nullable=False,             # ✅ 不允许为NULL
        comment="创建时间"
    )
 
    # 8. 时间字段 - 更新时间
    updated_at: Mapped[datetime] = mapped_column(
        TIMESTAMP(timezone=True),
        server_default=func.now(),
        onupdate=func.now(),        # ✅ 更新时自动设置为当前时间
        nullable=False,
        comment="最后更新时间"
    )
 
    # 9. 布尔字段
    is_active: Mapped[bool] = mapped_column(
        Boolean,                    # ✅ 布尔类型
        default=True,               # ✅ 默认值为True
        nullable=False,
        comment="是否激活（True=激活，False=禁用）"
    )
 
    # ========== 关系定义 ==========
 
    # 10. 定义关联关系（如果有关联表）
    # back_populates 表示双向关系
    user: Mapped["User"] = relationship(
        back_populates="girlfriend_configs",  # ✅ User类中对应的属性名
        lazy="select"                         # ✅ 加载策略
    )
 
    # ========== 特殊方法 ==========
 
    def __repr__(self):
        """对象字符串表示，用于调试"""
        return f"<GirlfriendConfig(id={self.id}, nickname='{self.nickname}')>"
 
    def to_dict(self):
        """转换为字典，用于API响应"""
        return {
            "id": self.id,
            "nickname": self.nickname,
            "personality": self.personality,
            "created_at": self.created_at.isoformat() if self.created_at else None,
            "is_active": self.is_active
        }

@router.get('/info',response_model=response_success,summary='获取用户信息',tags=['角色信息'])
# step1. 定义传参和数据库会话
async def get_user_info(db: AsyncSession = Depends(get_db),code:Optional[str] = Header(None)):
    # step 2. 定义边缘条件
    if code is None:
        raise UnicornException(name="用户不存在")
    user = select(User.create_at,User.id,User.user_name).where(User.code == code) # step 4.定义sql
    result = await db.execute(user)# step 5. 提交并执行sql
    """
	db.execute:提交sql语句
	db.commit:提交事务(只有增删改需要)
    db.add(Object): 只有新增的sql用
    db.add_all(List(object)) 批量新增
    db.delete()： 删除
    """
    row = result.one_or_none() #根据sql返回内容定义接口返回值
    """
    result 是数据库执行 SQL 后返回的结果容器。
 
    result.rowcount  返回 SQL 影响的行数（删除或修改）。
 
    ==========查询==========
	result.one_or_none()     # 取单行，不存在返回 None，多行则报错；适合查询可能不存在的用户
		result = await db.execute(select(User.psd, User.code).where(...))  # 查询多列
	result.scalar_one_or_none() # 取单行，不存在返回 None，多行则报错；自动解包为对象
		result = await db.execute(select(User).where(User.id == userid)) # 查询单列
	下面只可以单列
	result.scalar()
	result.scalars().all()
	下面单列多列都可以
	result.one()             # 取单行，不存在直接报错 场景：查当前登录用户信息，理论上一定存在
	result.first()           # 取第一行，不存在返回None
	result.all()             # 取所有行，返回列表
	result.scalar()          # 取第一行第一列的值
	result.scalars().all()   # 只查一列时，返回扁平列表 [1, 2, 3]
	result.mappings().all()  # 返回字典格式 [{'id':1, 'name':'张三'}]
    """
    if row is None:
        raise UnicornException(name="用户不存在")
    createTime,userId,username = row
    success = {
        "createTime":createTime,
        "userId":userId,
        "userName":username
    }
    response_success.data = success # 返回结果
    return response_success

from sqlalchemy import (
    select,      # 查询 ✅
    insert,      # 插入
    update,      # 更新
    delete,      # 删除
    text,        # 原生SQL
    func,        # SQL函数
    case,        # CASE语句
    cast,        # 类型转换
    and_, or_, not_,  # 逻辑操作
    union, union_all, intersect, except_  # 集合操作
)
 
stmt = (
    select(User)                    # 1. 选择表
    .join(Order, User.id == Order.user_id)  # 2. JOIN
    .where(User.is_active == True)           # 3. WHERE条件
    .where(User.age >= 18)                   # 4. 可以多个where
    .group_by(User.city)                     # 5. GROUP BY
    .having(func.count(Order.id) > 5)        # 6. HAVING条件
    .order_by(User.created_at.desc())        # 7. ORDER BY
    .limit(10)                               # 8. LIMIT
    .offset(20)                              # 9. OFFSET
    .distinct()                              # 10. DISTINCT
)