3

我有以下代码片段用于从数据库服务器获取当前时间戳：

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 日志——每个查询都在访问数据库。

我能够完成这项工作的唯一方法是导入 django.db.transaction 并transaction.commit_unless_managed()在执行 SELECT 之前执行。在 SELECT 之前提交事务似乎不正确，因为我不知道将在什么上下文中调用 get_current_timestamp。有关如何解决此问题或解决此问题的任何想法？时间戳需要来自数据库服务器，以确保正确完成所有相对时间计算。

我在跑步：

Django==1.4.2
psycopg2==2.4.5 (with hstore extension via django-hstore==1.1.1)
PostgreSQL 9.1
4

1 回答 1

3

在与同事讨论后，我们意识到这实际上可能是理想的行为，因为每个 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 是事务开始的时间。

于 2012-11-01T18:44:39.270 回答