Flask-sqlalchemy 查询日期时间间隔

Question

我用flask-sqlalchemy 定义了一个表。显示如下。

class Notes(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    notes = db.Column(db.Text, nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
    added_at = db.Column(db.DateTime, default=db.func.now())

@staticmethod
def newest(num):
    return Notes.query.order_by(desc(Notes.added_at)).limit(num)

我正在尝试编写一个查询来替换和已经存在的直接查询，它看起来像这样。

select notes,user,added_at from notes where added_at >= now() - INTERVAL 8 HOUR;

但是，根据我可以找到的文档，我无法找到执行相同操作的方法。我能够进行更简单的查询，但我正在努力重新创建 sql 本身中非常简单的内容。

我非常愿意阅读一些围绕它的文档，但也无法准确地确定下来。你能提供的任何方向都会很棒。

Answer 1

我总是让 Python 的datetime库为我提供“现在”和“8 小时前”，然后使用日期时间进行过滤：

from datetime import datetime, timedelta

now = datetime.now()
eight_hours_ago = now - timedelta(hours=8)

Notes.query.filter(Notes.added_at > eight_hours_ago).filter(Notes.added_at < now).all()

Answer 2

以下内容也应该有效：

from sqlalchemy import func
from sqlalchemy.dialects.postgresql import INTERVAL
from sqlalchemy.sql.functions import concat

Notes.query\
    .filter(
        Notes.added_at >= (func.now() - func.cast(concat(8, ' HOURS'), INTERVAL))
    )\
    .limit(num)

它有一个很好的属性，8可以用数据库内部的值替换，例如，如果您加入另一个具有动态间隔的表。我也在这里给出了这个答案。

Answer 3

你可以尝试类似的东西

Notes.query.order_by(desc(Notes.added_at)).filter(
    Notes.added_at >= text('NOW() - INTERVAL 8 HOURS').limit(num)

由于我只使用纯 sqlalchemy，因此我使用以下语法进行了测试：

>>> from sqlalchemy import text
>>> # s is a standard sqlalchemy session created from elsewhere.
>>> print s.query(Notes).order_by(desc(Notes.added_at)).filter(
...     Notes.added_at >= text('NOW() - INTERVAL 8 HOURS'))
SELECT notes.id AS notes_id, notes.notes AS notes_notes, notes.added_at AS notes_added_at 
FROM notes 
WHERE notes.added_at >= NOW() - INTERVAL 8 HOURS ORDER BY notes.added_at DESC

使用text该部分的原因很简单，因为所有 sql 实现的NOW()用法INTERVAL并不一致（某些实现需要使用 来DATEADD进行日期时间算术，虽然 sqlalchemy 确实支持该Interval类型，但它并没有很好的记录，而且在我的简短测试中它实际上并没有做你需要的事情（使用这个答案中的示例，对于sqlite和MySQL）。如果你打算使用SQL后端作为有序（但愚蠢）的数据存储，你可以从内部构建实际的查询Python，也许像这样：

q = s.query(Notes).order_by(desc(Notes.added_at)).filter(
    Notes.added_at >= (datetime.utcnow() - timedelta(3600 * 8))
)

有些人不喜欢这一点，因为某些数据库（如 postgresql）可以比 Python 更好地处理日期时间（例如 timedelta 不知道闰年）。