Coverage for watcher/api/controllers/v1/scoring_engine.py: 92%
108 statements
« prev ^ index » next coverage.py v7.8.2, created at 2025-06-17 12:22 +0000
1# -*- encoding: utf-8 -*-
2# Copyright 2016 Intel
3# All Rights Reserved.
4#
5# Licensed under the Apache License, Version 2.0 (the "License");
6# you may not use this file except in compliance with the License.
7# You may obtain a copy of the License at
8#
9# http://www.apache.org/licenses/LICENSE-2.0
10#
11# Unless required by applicable law or agreed to in writing, software
12# distributed under the License is distributed on an "AS IS" BASIS,
13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
14# implied.
15# See the License for the specific language governing permissions and
16# limitations under the License.
18"""
19A :ref:`Scoring Engine <scoring_engine_definition>` is an executable that has
20a well-defined input, a well-defined output, and performs a purely mathematical
21task. That is, the calculation does not depend on the environment in which it
22is running - it would produce the same result anywhere.
24Because there might be multiple algorithms used to build a particular data
25model (and therefore a scoring engine), the usage of scoring engine might
26vary. A metainfo field is supposed to contain any information which might
27be needed by the user of a given scoring engine.
28"""
30import pecan
31from pecan import rest
32from wsme import types as wtypes
33import wsmeext.pecan as wsme_pecan
35from watcher.api.controllers import base
36from watcher.api.controllers import link
37from watcher.api.controllers.v1 import collection
38from watcher.api.controllers.v1 import types
39from watcher.api.controllers.v1 import utils as api_utils
40from watcher.common import exception
41from watcher.common import policy
42from watcher import objects
45def hide_fields_in_newer_versions(obj):
46 """This method hides fields that were added in newer API versions.
48 Certain node fields were introduced at certain API versions.
49 These fields are only made available when the request's API version
50 matches or exceeds the versions when these fields were introduced.
51 """
52 pass
55class ScoringEngine(base.APIBase):
56 """API representation of a scoring engine.
58 This class enforces type checking and value constraints, and converts
59 between the internal object model and the API representation of a scoring
60 engine.
61 """
63 uuid = types.uuid
64 """Unique UUID of the scoring engine"""
66 name = wtypes.text
67 """The name of the scoring engine"""
69 description = wtypes.text
70 """A human readable description of the Scoring Engine"""
72 metainfo = wtypes.text
73 """A metadata associated with the scoring engine"""
75 links = wtypes.wsattr([link.Link], readonly=True)
76 """A list containing a self link and associated action links"""
78 def __init__(self, **kwargs):
79 super(ScoringEngine, self).__init__()
81 self.fields = []
82 self.fields.append('uuid')
83 self.fields.append('name')
84 self.fields.append('description')
85 self.fields.append('metainfo')
86 setattr(self, 'uuid', kwargs.get('uuid', wtypes.Unset))
87 setattr(self, 'name', kwargs.get('name', wtypes.Unset))
88 setattr(self, 'description', kwargs.get('description', wtypes.Unset))
89 setattr(self, 'metainfo', kwargs.get('metainfo', wtypes.Unset))
91 @staticmethod
92 def _convert_with_links(se, url, expand=True):
93 if not expand:
94 se.unset_fields_except(
95 ['uuid', 'name', 'description'])
97 se.links = [link.Link.make_link('self', url,
98 'scoring_engines', se.uuid),
99 link.Link.make_link('bookmark', url,
100 'scoring_engines', se.uuid,
101 bookmark=True)]
102 return se
104 @classmethod
105 def convert_with_links(cls, scoring_engine, expand=True):
106 scoring_engine = ScoringEngine(**scoring_engine.as_dict())
107 hide_fields_in_newer_versions(scoring_engine)
108 return cls._convert_with_links(
109 scoring_engine, pecan.request.host_url, expand)
111 @classmethod
112 def sample(cls, expand=True):
113 sample = cls(uuid='81bbd3c7-3b08-4d12-a268-99354dbf7b71',
114 name='sample-se-123',
115 description='Sample Scoring Engine 123 just for testing')
116 return cls._convert_with_links(sample, 'http://localhost:9322', expand)
119class ScoringEngineCollection(collection.Collection):
120 """API representation of a collection of scoring engines."""
122 scoring_engines = [ScoringEngine]
123 """A list containing scoring engine objects"""
125 def __init__(self, **kwargs):
126 super(ScoringEngineCollection, self).__init__()
127 self._type = 'scoring_engines'
129 @staticmethod
130 def convert_with_links(scoring_engines, limit, url=None, expand=False,
131 **kwargs):
133 collection = ScoringEngineCollection()
134 collection.scoring_engines = [ScoringEngine.convert_with_links(
135 se, expand) for se in scoring_engines]
136 collection.next = collection.get_next(limit, url=url, **kwargs)
137 return collection
139 @classmethod
140 def sample(cls):
141 sample = cls()
142 sample.scoring_engines = [ScoringEngine.sample(expand=False)]
143 return sample
146class ScoringEngineController(rest.RestController):
147 """REST controller for Scoring Engines."""
149 def __init__(self):
150 super(ScoringEngineController, self).__init__()
152 from_scoring_engines = False
153 """A flag to indicate if the requests to this controller are coming
154 from the top-level resource Scoring Engines."""
156 _custom_actions = {
157 'detail': ['GET'],
158 }
160 def _get_scoring_engines_collection(self, marker, limit,
161 sort_key, sort_dir, expand=False,
162 resource_url=None):
163 api_utils.validate_sort_key(
164 sort_key, list(objects.ScoringEngine.fields))
165 limit = api_utils.validate_limit(limit)
166 api_utils.validate_sort_dir(sort_dir)
168 marker_obj = None
169 if marker: 169 ↛ 170line 169 didn't jump to line 170 because the condition on line 169 was never true
170 marker_obj = objects.ScoringEngine.get_by_uuid(
171 pecan.request.context, marker)
173 filters = {}
175 sort_db_key = (sort_key if sort_key in objects.ScoringEngine.fields
176 else None)
178 scoring_engines = objects.ScoringEngine.list(
179 context=pecan.request.context,
180 limit=limit,
181 marker=marker_obj,
182 sort_key=sort_db_key,
183 sort_dir=sort_dir,
184 filters=filters)
186 return ScoringEngineCollection.convert_with_links(
187 scoring_engines,
188 limit,
189 url=resource_url,
190 expand=expand,
191 sort_key=sort_key,
192 sort_dir=sort_dir)
194 @wsme_pecan.wsexpose(ScoringEngineCollection, wtypes.text,
195 int, wtypes.text, wtypes.text)
196 def get_all(self, marker=None, limit=None, sort_key='id',
197 sort_dir='asc'):
198 """Retrieve a list of Scoring Engines.
200 :param marker: pagination marker for large data sets.
201 :param limit: maximum number of resources to return in a single result.
202 :param sort_key: column to sort results by. Default: name.
203 :param sort_dir: direction to sort. "asc" or "desc". Default: asc.
204 """
205 context = pecan.request.context
206 policy.enforce(context, 'scoring_engine:get_all',
207 action='scoring_engine:get_all')
209 return self._get_scoring_engines_collection(
210 marker, limit, sort_key, sort_dir)
212 @wsme_pecan.wsexpose(ScoringEngineCollection, wtypes.text,
213 int, wtypes.text, wtypes.text)
214 def detail(self, marker=None, limit=None, sort_key='id', sort_dir='asc'):
215 """Retrieve a list of Scoring Engines with detail.
217 :param marker: pagination marker for large data sets.
218 :param limit: maximum number of resources to return in a single result.
219 :param sort_key: column to sort results by. Default: name.
220 :param sort_dir: direction to sort. "asc" or "desc". Default: asc.
221 """
222 context = pecan.request.context
223 policy.enforce(context, 'scoring_engine:detail',
224 action='scoring_engine:detail')
226 parent = pecan.request.path.split('/')[:-1][-1]
227 if parent != "scoring_engines":
228 raise exception.HTTPNotFound
229 expand = True
230 resource_url = '/'.join(['scoring_engines', 'detail'])
231 return self._get_scoring_engines_collection(
232 marker, limit, sort_key, sort_dir, expand, resource_url)
234 @wsme_pecan.wsexpose(ScoringEngine, wtypes.text)
235 def get_one(self, scoring_engine):
236 """Retrieve information about the given Scoring Engine.
238 :param scoring_engine_name: The name of the Scoring Engine.
239 """
240 context = pecan.request.context
241 policy.enforce(context, 'scoring_engine:get',
242 action='scoring_engine:get')
244 if self.from_scoring_engines: 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true
245 raise exception.OperationNotPermitted
247 rpc_scoring_engine = api_utils.get_resource(
248 'ScoringEngine', scoring_engine)
250 return ScoringEngine.convert_with_links(rpc_scoring_engine)