<p><strong>免责声明1:</strong><em>首先,考虑一下实际使用<code>postgresql</code>代替sqlite是否“更便宜”。我假设你在开发/生产上有差异,你应该避免。在任何现代操作系统上安装<code>postgresql</code>都非常简单。</em><br/>
假设以上不是一个选项/不需要,让我们继续。在</p>
<p><strong>免责声明2:</strong>使用自定义SQL构造的解决方案(根据@Josh的回答)是实现这一目标的唯一合理方法。
不幸的是,建议的解决方案实际上不适用于<code>sqlite</code>,并且不能仅用几行代码来修复,因此需要单独的答案。在</p>
<p><strong>解决方案:</strong>
假设您有以下型号:</p>
<pre><code>class Appointment(Base):
__tablename__ = 'appointment'
id = Column(Integer, primary_key=True)
name = Column(String(255))
datetime = Column(DateTime) # @note: should be better named `start_date`?
duration = Column(Integer)
</code></pre>
<p><code>sqlite</code>处理日期操作非常棘手,尤其是从日期中加/减间隔。因此,让我们以稍微不同的方式处理它,并创建自定义函数,以分钟为单位获取两个日期之间的间隔:</p>
^{pr2}$
<p>您已经可以使用以下查询实现检查(<em>我没有在我的示例中添加<code>limit(1).one_or_none</code>,当您需要它时,显然可以这样做:</em>):</p>
<pre><code>q = (
session
.query(Appointment)
.filter(Appointment.datetime <= func.now())
.filter(diff_minutes(func.now(), Appointment.datetime) <= Appointment.duration)
)
</code></pre>
<p>但现在您不受当前时间(<code>func.now()</code>)的限制,您可以随时检查(和单元测试)您的数据:</p>
<pre><code># at_time = func.now()
at_time = datetime.datetime(2017, 11, 11, 17, 50, 0)
q = (
session
.query(Appointment)
.filter(Appointment.datetime <= at_time)
.filter(diff_minutes(at_time, Appointment.datetime) <= Appointment.duration)
)
</code></pre>
<p>基本上,<strong>问题在这里得到了解决</strong>,解决方案应该适用于您使用的两个数据库引擎。在</p>
<p><strong>奖金:</strong></p>
<p>您可以使用<a href="http://docs.sqlalchemy.org/en/rel_1_1/orm/extensions/hybrid.html" rel="nofollow noreferrer">Hybrid Methods</a>隐藏检查事件是否为当前事件的实现。在</p>
<p>让我们将以下内容添加到<code>Appointment</code>类中:</p>
<pre><code>@hybrid_method
def is_current(self, at_time=None):
if at_time is None:
at_time = datetime.datetime.now()
return self.datetime <= at_time <= self.datetime + datetime.timedelta(minutes=self.duration)
@is_current.expression
def is_current(cls, at_time=None):
if at_time is None:
at_time = datetime.datetime.now()
stime = cls.datetime
diffm = diff_minutes(at_time, cls.datetime)
return and_(diffm >= 0, cls.duration >= diffm).label('is_current')
</code></pre>
<p>第一种方法允许您运行签入内存(在python上,而不是在SQL端):</p>
<pre><code>print(my_appointment.is_current())
</code></pre>
<p>第二种方法允许您构造如下查询:</p>
<pre><code>q = session.query(Appointment).filter(Appointment.is_current(at_time))
</code></pre>
<p>如果<code>at_time</code>未指定,则使用当前时间。当然,您可以根据需要修改查询:</p>
<pre><code>current_appointment = session.query(Appointment).filter(Appointment.is_current()).limit(1).one_or_none()
</code></pre>