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
|
# Copyright 2009 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
# pylint: disable-msg=E0611,W0212
__metaclass__ = type
__all__ = ['Processor', 'ProcessorFamily', 'ProcessorFamilySet']
from sqlobject import (
ForeignKey,
SQLMultipleJoin,
StringCol,
)
from storm.locals import Bool
from zope.component import getUtility
from zope.interface import implements
from canonical.database.sqlbase import SQLBase
from canonical.launchpad.webapp.interfaces import (
DEFAULT_FLAVOR,
IStoreSelector,
MAIN_STORE,
)
from lp.soyuz.interfaces.processor import (
IProcessor,
IProcessorFamily,
IProcessorFamilySet,
IProcessorSet,
ProcessorNotFound,
)
class Processor(SQLBase):
implements(IProcessor)
_table = 'Processor'
family = ForeignKey(dbName='family', foreignKey='ProcessorFamily',
notNull=True)
name = StringCol(dbName='name', notNull=True)
title = StringCol(dbName='title', notNull=True)
description = StringCol(dbName='description', notNull=True)
def __repr__(self):
return "<Processor %r>" % self.title
class ProcessorSet:
"""See `IProcessorSet`."""
implements(IProcessorSet)
def getByName(self, name):
"""See `IProcessorSet`."""
store = getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)
processor = store.find(Processor, Processor.name == name).one()
if processor is None:
raise ProcessorNotFound(name)
return processor
def getAll(self):
"""See `IProcessorSet`."""
store = getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)
return store.find(Processor)
class ProcessorFamily(SQLBase):
implements(IProcessorFamily)
_table = 'ProcessorFamily'
name = StringCol(dbName='name', notNull=True)
title = StringCol(dbName='title', notNull=True)
description = StringCol(dbName='description', notNull=True)
processors = SQLMultipleJoin('Processor', joinColumn='family')
restricted = Bool(allow_none=False, default=False)
def addProcessor(self, name, title, description):
"""See `IProcessorFamily`."""
return Processor(family=self, name=name, title=title,
description=description)
def __repr__(self):
return "<ProcessorFamily %r>" % self.title
class ProcessorFamilySet:
implements(IProcessorFamilySet)
def getByName(self, name):
"""Please see `IProcessorFamilySet`."""
# Please note that ProcessorFamily.name is unique i.e. the database
# will return a result set that's either empty or contains just one
# ProcessorFamily row.
store = getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)
rset = store.find(ProcessorFamily, ProcessorFamily.name == name)
return rset.one()
def getRestricted(self):
"""See `IProcessorFamilySet`."""
store = getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)
return store.find(ProcessorFamily, ProcessorFamily.restricted == True)
def getByProcessorName(self, name):
"""Please see `IProcessorFamilySet`."""
store = getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)
rset = store.find(
ProcessorFamily,
Processor.name == name, Processor.family == ProcessorFamily.id)
# Each `Processor` is associated with exactly one `ProcessorFamily`
# but there is also the possibility that the user specified a name for
# a non-existent processor.
return rset.one()
def new(self, name, title, description, restricted=False):
"""See `IProcessorFamily`."""
return ProcessorFamily(name=name, title=title,
description=description, restricted=restricted)
|