(Django 聚合)如何获得平均花费时间及其日期

(Django aggregate) how to get average spent time and the dates of them

我的模型


class UserRetention(models.Model):
    user = models.ForeignKey('users.User', on_delete=models.PROTECT,null=True, blank=True,related_name='user_retention')
    in_date = models.DateTimeField(auto_now_add=True)
    out_date = models.DateTimeField(blank=True, null=True)
    @property
    def duration(self):
        try:
            return self.out_date - self.in_date
        except:
            pass

    class Meta:
        get_latest_by = 'in_date'

我的目标

  1. 首先我需要按周对数据进行分组。
  2. 我需要做这样的事情 UserRetention.objects.all().aggregate(..... 来获得一个字典对象 avg_weekly_duration 按小时数计算?
[
{
"week": 1,
"avg_weekly_duration": "3",
},
{
"week": 2,
"avg_weekly_duration": "10",
},
{
"week": 3,
"avg_weekly_duration": "8",
},
{
"week": 4,
"avg_weekly_duration": "15",
}
]

解决方案

是一个自我解释的解决方案

from django.db.models import F, Sum, Count

#querying data:
q = (
    UserRetention
    .objects
    .values('in_date__week')
    .annotate(
        h=Sum(F('out_date')-F('in_date')),
        n=Count('user_id', distinct=True),                
    )
)
# formatting
l = [
        {
        'week': x['in_date__week'],
        'avg_weekly_duration': x['h'].total_seconds()/(x['n']*3600)
        }
    for x in q
    ]

# print result
print(l)

结果

[{'week': 1, 'avg_weekly_duration': 3.0}, {'week': 2, 'avg_weekly_duration': 4.5}]

测试

整个测试:

from aa.models import UserRetention
from django.contrib.auth.models import User
from django.test import TestCase
from django.test import override_settings
from django.db import connection, reset_queries
from datetime import datetime, timedelta
from django.db.models import F, Sum, Count

class SOTestCase(TestCase):
    def setUp(self):
        u1 = User.objects.create_user('john', 'lennon@thebeatles.com', 'johnpassword')
        u2 = User.objects.create_user('madonna', 'madonna@thebeatles.com', 'madonnapassword')
        #
        d2=datetime(2022,1,3)
        UserRetention.objects.create(user=u1, in_date=d2, out_date=d2+timedelta(hours=+2))
        d2+=+timedelta(hours=+4)
        UserRetention.objects.create(user=u1, in_date=d2, out_date=d2+timedelta(hours=+2))
        UserRetention.objects.create(user=u2, in_date=d2, out_date=d2+timedelta(hours=+2))
        #
        d2=datetime(2022,1,10)
        UserRetention.objects.create(user=u1, in_date=d2, out_date=d2+timedelta(hours=+3))
        d2+=+timedelta(hours=+4)
        UserRetention.objects.create(user=u1, in_date=d2, out_date=d2+timedelta(hours=+3))
        UserRetention.objects.create(user=u2, in_date=d2, out_date=d2+timedelta(hours=+3))

    @override_settings(DEBUG=True)
    def test_query(self):
        reset_queries()
        q = (
            UserRetention
            .objects
            .values('in_date__week')
            .annotate(
                h=Sum(F('out_date')-F('in_date')),
                n=Count('user_id', distinct=True),                
            )
        )

        l = [
               {
                'week': x['in_date__week'],
                'avg_weekly_duration': x['h'].total_seconds()/(x['n']*3600)
               }
            for x in q
         ]

        expected = [{'week': 1, 'avg_weekly_duration': 3.0}, {'week': 2, 'avg_weekly_duration': 4.5}]
        self.assertListEqual(l, expected)

        print(connection.queries[0]['sql'])

结果:

% python manage.py test
Found 1 test(s).
Creating test database for alias 'default'...
System check identified no issues (0 silenced).
SELECT Django_datetime_extract('week', "aa_userretention"."in_date", NULL, NULL)
       ,
       Sum(Django_timestamp_diff("aa_userretention"."out_date",
               "aa_userretention"."in_date"))       AS "h",
       Count(DISTINCT "aa_userretention"."user_id") AS "n"
FROM   "aa_userretention"
GROUP  BY Django_datetime_extract('week', "aa_userretention"."in_date", NULL,
          NULL) 
.
----------------------------------------------------------------------
Ran 1 test in 0.105s

OK
Destroying test database for alias 'default'...