forked from volpino/euscan
-
Notifications
You must be signed in to change notification settings - Fork 0
/
managers.py
139 lines (104 loc) · 3.74 KB
/
managers.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
"""
djeuscan.managers
"""
from django.db import models
from djeuscan.helpers import xint, rename_fields, select_related_last_versions
def gen_n_function(field_name):
def n_method(self):
res = self.aggregate(models.Sum(field_name))[field_name + '__sum']
return xint(res)
n_method.func_name = field_name
return n_method
def gen_for_function(field):
def for_method(self, val, last_versions=False):
"""
Returns packages that belong to the given parametrs
"""
res = self.filter(**{field : val})
if last_versions:
select_related_last_versions(res)
return res
for_method.func_name = 'for_' + field
return for_method
N_LIST = ['n_packaged','n_overlay','n_versions']
ANNOTATE_DICT = { name: models.Sum(name) for name in N_LIST }
class PackageMixin(object):
for name in N_LIST:
locals()[name] = gen_n_function(name)
def n_upstream(self):
return self.n_versions() - self.n_packaged() - self.n_overlay()
def categories(self):
"""
Returns all the available categories
"""
return self.values('category').annotate(**ANNOTATE_DICT)
def herds(self, rename=False):
"""
Returns all the available herds
"""
# FIXME: optimize the query, it uses 'LEFT OUTER JOIN' instead of
# 'INNER JOIN'
res = self.filter(herds__isnull=False)
res = res.values('herds__herd').annotate(**ANNOTATE_DICT)
if rename:
res = rename_fields(res, [('herds__herd', 'herd')])
return res
def maintainers(self, rename=False):
"""
Returns all the available maintainers
"""
res = self.filter(maintainers__isnull=False).values(
'maintainers__id', 'maintainers__name', 'maintainers__email'
)
res = res.annotate(**ANNOTATE_DICT)
if rename:
res = rename_fields(
res,
[('maintainers__id', 'id'),
('maintainers__name', 'name'),
('maintainers__email', 'email')]
)
return res
def overlays(self):
"""
Returns the all available overlays
"""
res = self.values('version__overlay').exclude(version__overlay='')
return res.distinct()
def for_overlay(self, overlay):
"""
Returns packages that belong to the given overlay
"""
packages = self.values(
'id', 'name', 'category', 'n_versions', 'n_packaged', 'n_overlay'
)
return packages.filter(version__overlay=overlay).distinct()
for_maintainer = gen_for_function('maintainers')
for_herd = gen_for_function('herds')
for_category = gen_for_function('category')
class PackageQuerySet(models.query.QuerySet, PackageMixin):
pass
class PackageManager(models.Manager, PackageMixin):
def get_query_set(self):
return PackageQuerySet(self.model, using=self._db)
class VersionLogMixin(object):
def for_package(self, package, order=False):
res = filter(package=package)
if order:
res = res.order_by('-id')
return res
def for_maintainer(self, maintainer, order=False):
res = self.filter(package__maintainers__id=maintainer.id)
if order:
res = res.order_by('-id')
return res
def for_category(self, category, order=False):
res = self.filter(package__category=category)
if order:
res = res.order_by('-id')
return res
class VersionLogQuerySet(models.query.QuerySet, VersionLogMixin):
pass
class VersionLogManager(models.Manager, VersionLogMixin):
def get_query_set(self):
return VersionLogQuerySet(self.model, using=self._db)