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
|
# Copyright 2009 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
# pylint: disable-msg=W0231
"""The content classes for links from source packages to branches.."""
__metaclass__ = type
__all__ = [
'SeriesSourcePackageBranch',
'SeriesSourcePackageBranchSet',
]
from datetime import datetime
import pytz
from storm.locals import (
DateTime,
Int,
Reference,
Storm,
)
from zope.component import getUtility
from zope.interface import implements
from canonical.database.enumcol import DBEnum
from canonical.launchpad.webapp.interfaces import (
DEFAULT_FLAVOR,
IStoreSelector,
MAIN_STORE,
MASTER_FLAVOR,
)
from lp.code.interfaces.seriessourcepackagebranch import (
IFindOfficialBranchLinks,
IMakeOfficialBranchLinks,
ISeriesSourcePackageBranch,
)
from lp.registry.interfaces.pocket import PackagePublishingPocket
class SeriesSourcePackageBranch(Storm):
"""See `ISeriesSourcePackageBranch`."""
__storm_table__ = 'SeriesSourcePackageBranch'
implements(ISeriesSourcePackageBranch)
id = Int(primary=True)
distroseriesID = Int('distroseries')
distroseries = Reference(distroseriesID, 'DistroSeries.id')
pocket = DBEnum(enum=PackagePublishingPocket)
sourcepackagenameID = Int('sourcepackagename')
sourcepackagename = Reference(
sourcepackagenameID, 'SourcePackageName.id')
branchID = Int('branch')
branch = Reference(branchID, 'Branch.id')
registrantID = Int('registrant')
registrant = Reference(registrantID, 'Person.id')
date_created = DateTime(allow_none=False)
def __init__(self, distroseries, pocket, sourcepackagename, branch,
registrant, date_created):
"""Construct an `ISeriesSourcePackageBranch`."""
self.distroseries = distroseries
self.pocket = pocket
self.sourcepackagename = sourcepackagename
self.branch = branch
self.registrant = registrant
self.date_created = date_created
@property
def sourcepackage(self):
return self.distroseries.getSourcePackage(self.sourcepackagename)
@property
def suite_sourcepackage(self):
return self.sourcepackage.getSuiteSourcePackage(self.pocket)
class SeriesSourcePackageBranchSet:
"""See `ISeriesSourcePackageBranchSet`."""
implements(IFindOfficialBranchLinks, IMakeOfficialBranchLinks)
def new(self, distroseries, pocket, sourcepackagename, branch, registrant,
date_created=None):
"""See `IMakeOfficialBranchLinks`."""
if date_created is None:
date_created = datetime.now(pytz.UTC)
sspb = SeriesSourcePackageBranch(
distroseries, pocket, sourcepackagename, branch, registrant,
date_created)
store = getUtility(IStoreSelector).get(MAIN_STORE, MASTER_FLAVOR)
store.add(sspb)
return sspb
def findForBranch(self, branch):
"""See `IFindOfficialBranchLinks`."""
return self.findForBranches([branch])
def findForBranches(self, branches):
"""See `IFindOfficialBranchLinks`."""
branch_ids = set(branch.id for branch in branches)
store = getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)
return store.find(
SeriesSourcePackageBranch,
SeriesSourcePackageBranch.branchID.is_in(branch_ids))
def findForSourcePackage(self, sourcepackage):
"""See `IFindOfficialBranchLinks`."""
store = getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)
distroseries = sourcepackage.distroseries
sourcepackagename = sourcepackage.sourcepackagename
return store.find(
SeriesSourcePackageBranch,
SeriesSourcePackageBranch.distroseries == distroseries.id,
SeriesSourcePackageBranch.sourcepackagename ==
sourcepackagename.id)
def findForDistributionSourcePackage(self, distrosourcepackage):
"""See `IFindOfficialBranchLinks`."""
# To prevent circular imports.
from lp.registry.model.distroseries import DistroSeries
store = getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)
distro = distrosourcepackage.distribution
sourcepackagename = distrosourcepackage.sourcepackagename
return store.find(
SeriesSourcePackageBranch,
DistroSeries.distribution == distro.id,
SeriesSourcePackageBranch.distroseries == DistroSeries.id,
SeriesSourcePackageBranch.sourcepackagename ==
sourcepackagename.id)
def delete(self, sourcepackage, pocket):
"""See `IMakeOfficialBranchLinks`."""
store = getUtility(IStoreSelector).get(MAIN_STORE, MASTER_FLAVOR)
distroseries = sourcepackage.distroseries
sourcepackagename = sourcepackage.sourcepackagename
return store.find(
SeriesSourcePackageBranch,
SeriesSourcePackageBranch.distroseries == distroseries.id,
SeriesSourcePackageBranch.sourcepackagename ==
sourcepackagename.id,
SeriesSourcePackageBranch.pocket == pocket).remove()
|