Блог

Я только недавно начал использовать SQLAlchemy, и у меня все еще возникают проблемы с осмыслением некоторых концепций.

Если свести к основным элементам, у меня есть две таблицы, подобные этой (это через Flask-SQLAlchemy):

 class User(db.Model):
    __tablename__ = 'users'
    user_id = db.Column(db.Integer, primary_key=True)

class Posts(db.Model):
    __tablename__ = 'posts'
    post_id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey('users.user_id'))
    post_time = db.Column(db.DateTime)

    user = db.relationship('User', backref='posts')
  

Как бы я мог запросить список пользователей и их самую новую запись (исключая пользователей, у которых нет записей). Если бы я использовал SQL, я бы сделал:

 SELECT [whatever]
FROM posts AS p
    LEFT JOIN users AS u ON u.user_id = p.user_id
WHERE p.post_time = (SELECT MAX(post_time) FROM posts WHERE user_id = u.user_id)
  

Итак, я точно знаю «желаемый» SQL для получения желаемого эффекта, но понятия не имею, как выразить это «правильно» в SQLAlchemy.

Редактировать: на случай, если это важно, я использую SQLAlchemy 0.6.6.

Это должно сработать (другой SQL, тот же результат):

 t = Session.query(
    Posts.user_id,
    func.max(Posts.post_time).label('max_post_time'),
).group_by(Posts.user_id).subquery('t')

query = Session.query(User, Posts).filter(and_(
    User.user_id == Posts.user_id,
    User.user_id == t.c.user_id,
    Posts.post_time == t.c.max_post_time,
))

for user, post in query:
    print user.user_id, post.post_id
  

Где c означает «столбцы»

предыдущий ответ работает, но также точный sql, который вы просили, написан так же, как и фактическое утверждение:

 print s.query(User, Posts).
    outerjoin(Posts.user).
    filter(Posts.post_time==
        s.query(
            func.max(Posts.post_time)
        ).
        filter(Posts.user_id==User.user_id).
        correlate(User).
        as_scalar()
    )
  

Я предполагаю, что «концепция», которая не обязательно очевидна, заключается в том, что as_scalar() в настоящее время необходим для установления подзапроса как «скалярного» (вероятно, следует предположить, что из контекста против == ).

Редактировать: Подтверждено, что это некорректное поведение, заполненный тикет # 2190. В текущем руководстве или выпуске 0.7.2 as_scalar() вызывается автоматически, и приведенный выше запрос может быть:

 print s.query(User, Posts).
    outerjoin(Posts.user).
    filter(Posts.post_time==
        s.query(
            func.max(Posts.post_time)
        ).
        filter(Posts.user_id==User.user_id).
        correlate(User)
    )
  

Обычно это выражается аналогично фактическому SQL — вы создаете подзапрос, который возвращает один результат, и сравниваете с ним — однако то, что иногда может быть настоящей болью, — это если вам приходится использовать таблицу в подзапросе, к которой вы уже запрашиваете или к которой присоединяетесь.

Решение заключается в создании версии модели с псевдонимом для ссылки в подзапросе.

Итак, предположим, что вы уже работаете в соединении, где у вас есть существующее Posts model и некоторые базовые query готовые — теперь вы хотели бы запросить список последних (одиночных) сообщений от каждого пользователя, вы бы отфильтровали запрос следующим образом:

 from sqlalchemy.orm import aliased
posts2 = aliased(Posts) # create aliased version

query = query.filter(
    model.post_id
    ==
    Posts.query # create query directly from model, NOT from the aliased version!
        .with_entities(posts2.post_id) # only select column "post_id"
        .filter(
            posts2.user_id == model.user_id
        )
        .order_by(posts2.post_id.desc()) # assume higher id == newer post
        .limit(1) # we must limit to a single row so we only get 1 value
)
  

Я намеренно не использовал func.max , потому что я считаю, что это более простая версия, и она уже есть в других ответах, этот пример, я думаю, будет полезен людям, которые обычно находят этот вопрос, потому что они ищут решение, как выполнить подзапрос к той же таблице.