如何用SQLAlchemy通过User表关联筛选FCM推送令牌？

我正在开发FCM通知功能，需要从user_notification_tokens表中筛选所有推送令牌，筛选条件依赖UserPrescription表的datetime字段，但这两张表无直接关联，均与User表存在关联。我尝试了两种SQLAlchemy查询写法，均触发关联报错，请问该如何正确实现此查询？

尝试的代码 #

def get_tokens():
    db: Session = next(db_service.get_session())

    x_minutes_to_event = datetime.now(pytz.utc) + timedelta(minutes=config.MINUTES_TO_PRESCRIPTION)
    tokens = [
        item[0]
        for item in db.query(models.UserNotificationsToken)\
            .join(
                models.User,
                models.User.id == models.UserNotificationsToken.user_id,
            )\
                .join(
                    models.UserPrescription,
                    models.UserPrescription.user_id == models.User.id
                )\
                    .filter(
                        models.UserPrescription.visiting_at <= x_minutes_to_event,
                        models.UserPrescription.visiting_at > datetime.now(pytz.utc),
                    )\
                        .values(column('token'))
    ]
    
    # 另一种写法
    tokens = [
        item[0]
        for item in db.query(models.UserNotificationsToken)\
            .join(
                models.UserPrescription,
                models.UserNotificationsToken.user_id == models.UserPrescription.user_id,
            )\
                .filter(
                    models.UserPrescription.visiting_at <= x_minutes_to_event,
                    models.UserPrescription.visiting_at > datetime.now(pytz.utc),
                )\
                    .values(column('token'))
    ]

报错信息 #

sqlalchemy.exc.InvalidRequestError: Don't know how to join to <Mapper at 0x7f0921681fd0; User>. Please use the .select_from() method to establish an explicit left side, as well as providing an explicit ON clause if not present already to help resolve the ambiguity.

或者

sqlalchemy.exc.InvalidRequestError: Don't know how to join to <Mapper at 0x7f54970e4970; UserPrescription>. Please use the .select_from() method to establish an explicit left side, as well as providing an explicit ON clause if not present already to help resolve the ambiguity.

报错原因是SQLAlchemy无法自动推断关联的左表对象，需要明确指定关联逻辑或用select_from显式声明查询起始表。以下是两种可行写法：

写法一：用select_from明确关联链 #

通过select_from指定从UserNotificationsToken开始，依次关联User和UserPrescription，让SQLAlchemy清晰理解关联关系：

def get_tokens():
    db: Session = next(db_service.get_session())

    now = datetime.now(pytz.utc)
    x_minutes_to_event = now + timedelta(minutes=config.MINUTES_TO_PRESCRIPTION)
    
    tokens = [
        item[0]
        for item in db.query(models.UserNotificationsToken.token)\
            .select_from(models.UserNotificationsToken)\
            .join(models.User, models.User.id == models.UserNotificationsToken.user_id)\
            .join(models.UserPrescription, models.UserPrescription.user_id == models.User.id)\
            .filter(
                models.UserPrescription.visiting_at <= x_minutes_to_event,
                models.UserPrescription.visiting_at > now
            )\
            .distinct()  # 避免同一用户有多条令牌时重复返回
    ]
    return tokens

写法二：子查询先筛选符合条件的用户ID #

先从UserPrescription中筛选出符合时间条件的用户ID集合，再关联UserNotificationsToken获取令牌，逻辑更直观：

def get_tokens():
    db: Session = next(db_service.get_session())

    now = datetime.now(pytz.utc)
    x_minutes_to_event = now + timedelta(minutes=config.MINUTES_TO_PRESCRIPTION)
    
    # 子查询：获取符合条件的用户ID
    eligible_users_subquery = db.query(models.UserPrescription.user_id)\
        .filter(
            models.UserPrescription.visiting_at <= x_minutes_to_event,
            models.UserPrescription.visiting_at > now
        )\
        .distinct()\
        .subquery()
    
    # 关联查询令牌
    tokens = [
        item[0]
        for item in db.query(models.UserNotificationsToken.token)\
            .filter(models.UserNotificationsToken.user_id.in_(eligible_users_subquery))\
            .distinct()
    ]
    return tokens

关键说明 #

• 两种写法都加入distinct()，避免同一用户拥有多个令牌时重复返回，可根据业务需求调整。
• 提取now变量复用，避免多次调用datetime.now(pytz.utc)导致时间不一致。
• 写法二的子查询方式在数据量较大时性能更优，先筛选小范围用户ID再关联令牌表。

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