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
|
# Copyright 2009 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
__metaclass__ = type
__all__ = ['ArchiveArch', 'ArchiveArchSet']
from storm.expr import (
And,
Join,
LeftJoin,
)
from storm.locals import (
Int,
Reference,
Storm,
)
from zope.component import getUtility
from zope.interface import implements
from canonical.launchpad.webapp.interfaces import (
DEFAULT_FLAVOR,
IStoreSelector,
MAIN_STORE,
)
from lp.soyuz.interfaces.archivearch import (
IArchiveArch,
IArchiveArchSet,
)
from lp.soyuz.model.processor import ProcessorFamily
class ArchiveArch(Storm):
"""See `IArchiveArch`."""
implements(IArchiveArch)
__storm_table__ = 'ArchiveArch'
id = Int(primary=True)
archive_id = Int(name='archive', allow_none=False)
archive = Reference(archive_id, 'Archive.id')
processorfamily_id = Int(name='processorfamily', allow_none=True)
processorfamily = Reference(processorfamily_id, 'ProcessorFamily.id')
class ArchiveArchSet:
"""See `IArchiveArchSet`."""
implements(IArchiveArchSet)
def new(self, archive, processorfamily):
"""See `IArchiveArchSet`."""
store = getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)
archivearch = ArchiveArch()
archivearch.archive = archive
archivearch.processorfamily = processorfamily
store.add(archivearch)
return archivearch
def getByArchive(self, archive, processorfamily=None):
"""See `IArchiveArchSet`."""
store = getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)
base_clauses = (ArchiveArch.archive == archive,)
if processorfamily is not None:
optional_clauses = (
ArchiveArch.processorfamily == processorfamily,)
else:
optional_clauses = ()
results = store.find(
ArchiveArch, *(base_clauses + optional_clauses))
results = results.order_by(ArchiveArch.id)
return results
def getRestrictedFamilies(self, archive):
"""See `IArchiveArchSet`."""
store = getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)
origin = (
ProcessorFamily,
LeftJoin(
ArchiveArch,
And(ArchiveArch.archive == archive.id,
ArchiveArch.processorfamily == ProcessorFamily.id)))
result_set = store.using(*origin).find(
(ProcessorFamily, ArchiveArch),
(ProcessorFamily.restricted == True))
return result_set.order_by(ProcessorFamily.name)
|