我有以下代码片段,用于从数据库服务器获取当前时间戳:
class DbUtils:
@staticmethod
def get_current_timestamp():
from django.db import connection
cursor = connection.cursor()
cursor.execute("SELECT CURRENT_TIMESTAMP")
rows = cursor.fetchone()
return rows[0]为了测试这段代码,我编写了以下测试:
def test_get_current_timestamp(self):
ts_0 = DbUtils.get_current_timestamp()
time.sleep(1)
ts_1 = DbUtils.get_current_timestamp()
delta_seconds = (ts_1 - ts_0).seconds
assert 1 <= delta_seconds <= 3这个测试没有通过,所以我深入研究了一下。原因是每次对DbUtils.get_current_timestamp()的调用都返回相同的值。
In [31]: DbUtils.get_current_timestamp()
Out[31]: datetime.datetime(2012, 11, 1, 17, 17, 48, 950799, tzinfo=<UTC>)
# Wait a couple seconds
In [32]: DbUtils.get_current_timestamp()
Out[32]: datetime.datetime(2012, 11, 1, 17, 17, 48, 950799, tzinfo=<UTC>)我监视postgresql日志--每个查询都会命中数据库。
我能够做到这一点的唯一方法是在执行SELECT之前导入django.db.transaction并执行transaction.commit_unless_managed()。在SELECT之前提交事务似乎是不正确的,因为我不知道将在什么上下文中调用get_current_timestamp。有没有什么办法来解决这个问题呢?时间戳需要来自数据库服务器,以确保正确地完成所有相对时间计算。
我正在运行:
Django==1.4.2
psycopg2==2.4.5 (with hstore extension via django-hstore==1.1.1)
PostgreSQL 9.1发布于 2012-11-02 02:44:39
在与同事讨论后,我们意识到这实际上可能是理想的行为,因为我们只能在每个web请求中获得一个时间戳。在任何情况下,我通常存储get_current_timestamp的返回值,并将其用于该请求期间的任何操作。
为了让我的测试通过,我需要做以下事情:
def test_get_current_timestamp(self):
from django.db import transaction
ts_0 = DbUtils.get_current_timestamp()
transaction.commit_unless_managed()
time.sleep(1)
ts_1 = DbUtils.get_current_timestamp()
transaction.commit_unless_managed()
delta_seconds = (ts_1 - ts_0).seconds
assert 1 <= delta_seconds <= 3正如@a_horse_with_no_name所指出的: CURRENT_TIMESTAMP是事务开始时的时间。
https://stackoverflow.com/questions/13182624
复制相似问题