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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
|
# Copyright 2009 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Implementation for the IRevisionCache and IRevisionCollection."""
__metaclass__ = type
__all__ = [
'GenericRevisionCollection',
]
from datetime import (
datetime,
timedelta,
)
import pytz
from storm.expr import (
Desc,
Func,
SQL,
)
from zope.component import getUtility
from zope.interface import implements
from lp.code.interfaces.revisioncache import IRevisionCollection
from lp.code.model.revision import (
Revision,
RevisionAuthor,
RevisionCache,
)
from lp.registry.model.distroseries import DistroSeries
from lp.registry.model.product import Product
from lp.registry.model.teammembership import TeamParticipation
from lp.services.webapp.interfaces import (
DEFAULT_FLAVOR,
IStoreSelector,
MAIN_STORE,
)
class GenericRevisionCollection:
"""See `IRevisionCollection`."""
implements(IRevisionCollection)
def __init__(self, store=None, filter_expressions=None):
self._store = store
if filter_expressions is None:
epoch = datetime.now(pytz.UTC) - timedelta(days=30)
filter_expressions = [
RevisionCache.revision_date >= epoch]
self._filter_expressions = filter_expressions
@property
def store(self):
# Although you might think we could set the default value for store in
# the constructor, we can't. The IStoreSelector utility is not
# available at the time that the branchcollection.zcml is parsed,
# which means we get an error if this code is in the constructor.
if self._store is None:
return getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)
else:
return self._store
def _filterBy(self, expressions):
return self.__class__(
self.store,
self._filter_expressions + expressions)
def count(self):
"""See `IRevisionCollection`."""
result_set = self.store.find(
RevisionCache.revision_id, self._filter_expressions)
result_set.config(distinct=True)
return result_set.count()
def authorCount(self):
"""See `IRevisionCollection`."""
# Revision authors that are linked to Launchpad people are only
# counted once even if the revision text that they use in the commit
# is different.
author = Func(
'coalesce',
RevisionAuthor.personID,
SQL(0) - RevisionAuthor.id)
expressions = [
RevisionCache.revision_author == RevisionAuthor.id]
expressions.extend(self._filter_expressions)
result_set = self.store.find(author, expressions)
result_set.config(distinct=True)
return result_set.count()
def getRevisions(self):
"""See `IRevisionCollection`."""
expressions = [
RevisionCache.revision == Revision.id]
expressions.extend(self._filter_expressions)
result_set = self.store.find(Revision, expressions)
result_set.config(distinct=True)
result_set.order_by(Desc(Revision.revision_date))
return result_set
def public(self):
"""See `IRevisionCollection`."""
return self._filterBy(
[RevisionCache.private == False])
def inProduct(self, product):
"""See `IRevisionCollection`."""
return self._filterBy(
[RevisionCache.product == product])
def inProject(self, project):
"""See `IRevisionCollection`."""
return self._filterBy(
[RevisionCache.product == Product.id,
Product.project == project])
def inSourcePackage(self, package):
"""See `IRevisionCollection`."""
return self._filterBy(
[RevisionCache.distroseries == package.distroseries,
RevisionCache.sourcepackagename == package.sourcepackagename])
def inDistribution(self, distribution):
"""See `IRevisionCollection`."""
return self._filterBy(
[DistroSeries.distribution == distribution,
RevisionCache.distroseries == DistroSeries.id])
def inDistroSeries(self, distro_series):
"""See `IRevisionCollection`."""
return self._filterBy(
[RevisionCache.distroseries == distro_series])
def inDistributionSourcePackage(self, distro_source_package):
"""See `IRevisionCollection`."""
distribution = distro_source_package.distribution
sourcepackagename = distro_source_package.sourcepackagename
return self._filterBy(
[DistroSeries.distribution == distribution,
RevisionCache.distroseries == DistroSeries.id,
RevisionCache.sourcepackagename == sourcepackagename])
def authoredBy(self, person):
"""See `IRevisionCollection`."""
if person.is_team:
query = [
TeamParticipation.team == person,
RevisionAuthor.personID == TeamParticipation.personID]
else:
query = [RevisionAuthor.person == person]
query.append(RevisionCache.revision_author == RevisionAuthor.id)
return self._filterBy(query)
|