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中相当简单的操作方面却遇到了困难。

我非常愿意阅读一些相关的文档，但是也无法准确地确定需要哪些文档。如果您能提供任何指导，那将非常棒。

- Harry

---

可能是SQLAlchemy：如何过滤日期字段？的重复问题。 - nathancahill

我看到了这个，但是它没有足够的信息来重新创建我试图完成的间隔函数。不过还是谢谢。 - Harry

@nathancahill 不是重复的。链接的问题没有使用数据库内部的“间隔”。 - exhuma

Answer 1

以下内容也应该可以正常工作：

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，例如，如果您将另一个具有动态间隔的表加入其中。我也在这里给出了答案。

- André C. Andersen

Answer 2

我通常使用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()

- TheGrimmScientist

---

这可能会导致与时区相关的微妙问题。此外，在Python进程中的now()与Postgres服务器上的now()不会相同（即使不考虑时区）。如果您有时间关键的查询，应该使用数据库内部的时间戳（也就是在SQL查询中使用NOW()）。 - exhuma

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来进行该部分的原因是因为NOW()和INTERVAL在所有SQL实现中的使用并不一致（某些实现需要使用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对闰年无知）。

- metatoaster

---

text('NOW() - INTERVAL 8 HOURS') 不起作用，你需要在内部加上引号，像这样 text("NOW() - INTERVAL '8 HOURS'")。 - Konstantin Smolyanin

@KonstantinSmolyanin 这是直接复制了 OP 写的查询，我修改了它以生成他们想要的等效查询。此外，其他答案肯定比这个过时的初始答案更好。 - metatoaster

Answer 4

也可以尝试

from sqlalchemy import func, text

@staticmethod
def newest(num):
    return Notes.query.filter(Notes.added_at >= (func.date_sub(func.now(),  text('INTERVAL  8 HOUR')).order_by(desc(Notes.added_at)).limit(num)

或者

from datetime import datetime, timedelta
from dateutil import tz 

@staticmethod
def newest(num):
    recent = datetime.now(tz=tz.tzlocal()) - timedelta(hours=8)

    return Notes.query.filter(Notes.added_at >= recent).order_by(desc(Notes.added_at)).limit(num)

- jackotonye

---

date_sub 不是一个 postgres 函数 :( - Peter Lada

是的，这是一个可以使用NOW() - '8 HOUR'::INTERVAL的MySQL函数。 - jackotonye

Answer 5

SQLAlchemy会自动将Python的timedelta实例适应为正确的interval查询。要使用它们，您需要以这样的方式重写查询，使得比较运算符的左侧或右侧之一是一个间隔。这样它就可以与时间差进行比较。

您当前的查询如下所示：

WHERE added_at >= now() - INTERVAL 8 HOUR

在比较符 >= 两侧，您有绝对时间戳。这很难转换为SQLAlchemy。您可以像这样重写它：

WHERE now() - added_at >= INTERVAL 8 HOUR

这给你两边的间隔。现在将其翻译成SQLAlchemy非常容易：

from datetime import timedelta
from sqlalchemy import func

class Notes(db.Model):
    @staticmethod
    def newest(num):
        query = Notes.query.filter(
            func.now() - Notes.added_at >= timedelta(hours=8)
        )
        return query.order_by(desc(Notes.added_at)).limit(num)

与现有答案相比，这个方法确保您使用数据库内部的时间戳（而不是使用Python中的datetime.now()）。

它不需要任何类型转换，也不使用字符串拼接或原始的text()方法。

- exhuma