SQLAlchemy异步2.0：博客点赞功能的实例检查与存储问题

我正在开发一个练手项目——带点赞功能的博客，使用异步SQLAlchemy 2.0，现有三个SQLAlchemy模型：

from sqlalchemy import String, Text, TIMESTAMP, ForeignKey, func
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column, relationship, List, Set
from sqlalchemy.ext.declarative import Base
from fastapi_users.db import SQLAlchemyBaseUserTableUUID
import uuid
from enum import Enum

class ReactionType(Enum):
    LIKE = "like"
    DISLIKE = "dislike"

class Reaction(Base):
    __tablename__ = "reaction"

    user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("user.id"), primary_key=True)
    post_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("post.id"), primary_key=True)
    type: Mapped[Enum[ReactionType]] = mapped_column(Enum(ReactionType))


class User(SQLAlchemyBaseUserTableUUID, Base):
    """
    User model:
        username: Mapped[str]
        ---------------------------
        fastapi_users default columns:
        id: Mapped[UUID_ID]
        email: Mapped[str]
        hashed_password: Mapped[str]
        is_active: Mapped[bool]
        is_superuser: Mapped[bool]
        is_verified: Mapped[bool]
    ---------------------------
    __tablename__ = "user"
    """

    username: Mapped[str] = mapped_column(
        String(length=100), unique=True, index=True, nullable=False
    )
    posts: Mapped[List["Post"]] = relationship("Post", back_populates="owner")


class Post(Base):
    __tablename__ = "post"

    id: Mapped[uuid.UUID] = mapped_column(UUID, primary_key=True, default=uuid.uuid4)
    owner_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("user.id"))
    owner: Mapped["User"] = relationship("User", back_populates="posts", lazy="selectin")
    title: Mapped[str] = mapped_column(String(200))
    description: Mapped[str] = mapped_column(Text, nullable=True)
    creation_date: Mapped[datetime] = mapped_column(
        TIMESTAMP, server_default=func.now()
    )
    last_update_date: Mapped[datetime] = mapped_column(
        TIMESTAMP, server_default=func.now(), onupdate=func.now()
    )
    user_reactions: Mapped[Set["Reaction"]] = relationship(lazy="selectin")

问题 #

1. 如何正确检查某用户是否可以对某篇文章点赞/取消点赞，或该用户是否已点赞过该文章？
2. 在内存数据库中存储文章点赞与点踩的最佳方式是什么？

补充说明 #

查询Reaction表中指定user_id和post_id的行不符合需求，且不清楚如何正确使用user_reactions关系。曾尝试创建与表中主键值相同的Reaction实例并添加到会话，但它们仍是不同对象：

# Objects with same pk`s
<posts.models.Reaction object at 0x000001F78BE11730>
InstrumentedSet({<posts.models.Reaction object at 0x000001F78BEDC430>})

另外，是否可以将关联的对象集合替换为主键集合？例如：

# Since this is a many-to-many relationship
InstrumentedSet({'post_id__user_id'})

而非

InstrumentedSet({<posts.models.Reaction object at 0x000001F78BEDC430>})

解决方案 #

问题1：检查用户点赞状态与正确使用关系 #

优化模型双向关联

先完善Reaction与User、Post的双向关联，让关联数据的访问更灵活：

class Reaction(Base):
    __tablename__ = "reaction"

    user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("user.id"), primary_key=True)
    post_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("post.id"), primary_key=True)
    type: Mapped[Enum[ReactionType]] = mapped_column(Enum(ReactionType))
    
    # 添加双向关联关系
    user: Mapped["User"] = relationship("User", back_populates="reactions")
    post: Mapped["Post"] = relationship("Post", back_populates="user_reactions")

class User(SQLAlchemyBaseUserTableUUID, Base):
    # ... 原有字段 ...
    posts: Mapped[List["Post"]] = relationship("Post", back_populates="owner")
    reactions: Mapped[Set["Reaction"]] = relationship("Reaction", back_populates="user")

class Post(Base):
    # ... 原有字段 ...
    user_reactions: Mapped[Set["Reaction"]] = relationship("Reaction", back_populates="post", lazy="selectin")

检查用户是否已点赞

利用已加载的user_reactions集合，无需额外查询Reaction表：

async def has_user_reacted(post: Post, user_id: uuid.UUID) -> tuple[bool, ReactionType | None]:
    # 遍历已加载的Reaction对象匹配用户ID
    for reaction in post.user_reactions:
        if reaction.user_id == user_id:
            return (True, reaction.type)
    return (False, None)

如果担心遍历效率，也可以在查询Post时直接过滤加载该用户的Reaction：

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

async def get_user_reaction_for_post(session: AsyncSession, post_id: uuid.UUID, user_id: uuid.UUID) -> Reaction | None:
    stmt = select(Reaction).where(Reaction.post_id == post_id, Reaction.user_id == user_id)
    result = await session.execute(stmt)
    return result.scalar_one_or_none()

解决相同主键实例不同的问题

SQLAlchemy会话会维护对象唯一性，不要手动创建主键相同的实例，正确做法是先从会话获取，不存在再创建：

# 错误：手动创建同主键实例
new_reaction = Reaction(user_id=user_id, post_id=post_id, type=ReactionType.LIKE)

# 正确：先尝试从会话获取，无则创建
existing_reaction = await session.get(Reaction, (user_id, post_id))
if not existing_reaction:
    existing_reaction = Reaction(user_id=user_id, post_id=post_id, type=ReactionType.LIKE)
    session.add(existing_reaction)
await session.commit()

问题2：内存数据库存储点赞的最佳方式 #

场景1：测试用内存数据库（SQLite）

如果是本地测试，直接使用SQLite内存模式，完全复用现有模型：

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

场景2：生产缓存用内存存储（Redis）

如果是生产环境用来缓存点赞数据，推荐用Redis哈希结构：

• 每个帖子对应哈希键：post:{post_id}:reactions
• 哈希字段为用户ID，值为点赞类型（like/dislike）

操作示例（使用异步Redis客户端）：

import redis.asyncio as redis

redis_client = redis.Redis(host="localhost", port=6379, db=0)

# 记录点赞
async def add_reaction(post_id: uuid.UUID, user_id: uuid.UUID, reaction_type: ReactionType):
    await redis_client.hset(f"post:{post_id}:reactions", str(user_id), reaction_type.value)

# 取消点赞
async def remove_reaction(post_id: uuid.UUID, user_id: uuid.UUID):
    await redis_client.hdel(f"post:{post_id}:reactions", str(user_id))

# 检查用户点赞状态
async def get_user_reaction(post_id: uuid.UUID, user_id: uuid.UUID) -> ReactionType | None:
    value = await redis_client.hget(f"post:{post_id}:reactions", str(user_id))
    return ReactionType(value) if value else None

补充问题：用主键集合替代对象集合 #

可以使用SQLAlchemy的association_proxy扩展，将关联的Reaction对象集合映射为用户ID集合或键值对：

首先安装扩展：

pip install sqlalchemy[associationproxy]

修改模型：

from sqlalchemy.ext.associationproxy import association_proxy

class Post(Base):
    # ... 原有字段 ...
    user_reactions: Mapped[Set["Reaction"]] = relationship("Reaction", back_populates="post", lazy="selectin")
    # 映射为{user_id: reaction_type}的字典
    reaction_map = association_proxy(
        "user_reactions",
        "type",
        creator=lambda user_id, type: Reaction(user_id=user_id, type=type)
    )
    # 映射为已参与互动的用户ID集合
    reacted_user_ids = association_proxy("user_reactions", "user_id")

使用时直接操作代理属性：

# 检查用户是否已点赞
if user_id in post.reacted_user_ids:
    print("用户已参与互动")

# 获取用户的点赞类型
reaction_type = post.reaction_map.get(user_id)

内容的提问来源于stack exchange，提问作者Patrick7575