Annotate an `exists` subquery in SQLAlchemy

Question

I have a simple news app where users can see latest posts and they can like some of these posts.

The models:

from sqlalchemy.orm import relationship, backref

from app import db


class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String, nullable=False)
    ...

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String)
    body = db.Column(db.String)
    ...

class Like(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.ForeignKey(User.id))
    post_id = db.Column(db.ForeignKey(Post.id))
    post = relationship(Post, backref=backref('likes'))

What I'm trying to achieve is to get each post with an additional field that indicates whether the post is liked by the logged in user or not but in an optimized way without making a query for each post.

I tried to do this is the schema level, using marshmallow-sqlalchemy:

from marshmallow import fields

from app import ma
from news.models import Post

class PostSchema(ma.ModelSchema):
    is_liked = fields.Method("get_is_liked")

    def get_is_liked(self, obj):
        user = self.context['user']
        return bool(Like.filter_by(user_id=user.id, post_id=obj.id).first())

    class Meta:
        model = Post

But this would result an N queries for each post.

I also tried to implement this like the annotations in Django ORM:

class PostListView(Resource):

    # I have the user object here using a decorator
    def get(self, user):
        # I know that this is a bad syntax
        query = Post.query(Post.likes.filter_by(user_id=user.id).exists().label('is_liked')))

        schema = PostSchema(strict=True, many=True, context={"user": user})
        result = schema.dump(query.all())

        return result.data

I have also seen this answer which do the same as I need in the database level but I want to do it with SQLAlchemy.

Thanks in advance.

Answer 1

Would this work? The idea is to load the posts so that Post.likes contains only the cases where the current user is the one who has liked the post. contains_eager('likes') is supposed to load the likes array in Post.

# user_id is the id of the current user
q = session.query(Post) \
           .outerjoin(Like, and_(Like.post_id == Post.id,
                                 Like.user_id == user_id)) \
           .options(contains_eager('likes'))

Then is_liked could be something like return len(self.likes) > 0