--- /dev/null
+# Copyright (C) 2021-2022 Wind River Systems, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from sqlalchemy.sql.elements import ColumnElement
+
+from o2common.helper import o2logging
+logger = o2logging.get_logger(__name__)
+
+
+def gen_filter(obj: ColumnElement, filter_str: str):
+ if filter_str == '':
+ return []
+ filter_without_space = filter_str.replace(" ", "")
+ items = filter_without_space.split(';')
+
+ filter_list = list()
+ for i in items:
+ if '(' in i:
+ i = i.replace("(", "")
+ if ')' in i:
+ i = i.replace(")", "")
+ filter_expr = i.split(',')
+ if len(filter_expr) < 3:
+ continue
+ filter_op = filter_expr[0]
+ filter_key = filter_expr[1]
+ filter_vals = filter_expr[2:]
+ filter_list.extend(toFilterArgs(
+ filter_op, obj, filter_key, filter_vals))
+ logger.info('Filter list length: %d' % len(filter_list))
+ return filter_list
+
+
+def toFilterArgs(operation: str, obj: ColumnElement, key: str, values: list):
+ if not hasattr(obj, key):
+ logger.warning('Filter attrName %s not in Object %s.' %
+ (key, str(obj)))
+ return []
+
+ ll = list()
+ if operation == 'eq':
+ for val in values:
+ if val.lower() == 'null':
+ val = None
+ ll.append(getattr(obj, key) == val)
+ elif operation == 'neq':
+ for val in values:
+ if val.lower() == 'null':
+ val = None
+ ll.append(getattr(obj, key) != val)
+ elif operation == 'gt':
+ for val in values:
+ ll.append(getattr(obj, key) > val)
+ elif operation == 'lt':
+ for val in values:
+ ll.append(getattr(obj, key) < val)
+ elif operation == 'gte':
+ for val in values:
+ ll.append(getattr(obj, key) >= val)
+ elif operation == 'lte':
+ for val in values:
+ ll.append(getattr(obj, key) <= val)
+ elif operation == 'in':
+ pass
+ elif operation == 'nin':
+ pass
+ elif operation == 'count':
+ pass
+ elif operation == 'ncount':
+ pass
+ else:
+ raise KeyError('Filter operation value not support.')
+
+ return ll
return self.session.query(alarm_obj.AlarmEventRecord).filter_by(
alarmEventRecordId=alarm_event_record_id).first()
- def _list(self, **kwargs) -> Tuple[int, List[alarm_obj.AlarmEventRecord]]:
+ def _list(self, *args, **kwargs) -> Tuple[
+ int, List[alarm_obj.AlarmEventRecord]]:
size = kwargs.pop('limit') if 'limit' in kwargs else None
offset = kwargs.pop('start') if 'start' in kwargs else 0
- result = self.session.query(alarm_obj.AlarmEventRecord).filter_by(
- **kwargs).order_by('alarmEventRecordId')
+ result = self.session.query(alarm_obj.AlarmEventRecord).filter(
+ *args).order_by('alarmEventRecordId')
count = result.count()
if size is not None and size != -1:
return (count, result.limit(size).offset(offset))
return self.session.query(alarm_obj.AlarmSubscription).filter_by(
alarmSubscriptionId=subscription_id).first()
- def _list(self, **kwargs) -> Tuple[int, List[alarm_obj.AlarmSubscription]]:
+ def _list(self, *args, **kwargs) -> Tuple[
+ int, List[alarm_obj.AlarmSubscription]]:
size = kwargs.pop('limit') if 'limit' in kwargs else None
offset = kwargs.pop('start') if 'start' in kwargs else 0
- result = self.session.query(alarm_obj.AlarmSubscription).filter_by(
- **kwargs).order_by('alarmSubscriptionId')
+ result = self.session.query(alarm_obj.AlarmSubscription).filter(
+ *args).order_by('alarmEventRecordId')
count = result.count()
if size is not None and size != -1:
return (count, result.limit(size).offset(offset))
return self.session.query(ocloud.ResourceType).filter_by(
name=resource_type_name).first()
- def _list(self, **kwargs) -> Tuple[int, List[ocloud.ResourceType]]:
+ def _list(self, *args, **kwargs) -> Tuple[int, List[ocloud.ResourceType]]:
size = kwargs.pop('limit') if 'limit' in kwargs else None
offset = kwargs.pop('start') if 'start' in kwargs else 0
- result = self.session.query(ocloud.ResourceType).filter_by(
- **kwargs).order_by('resourceTypeId')
+ result = self.session.query(ocloud.ResourceType).filter(
+ *args).order_by('resourceTypeId')
count = result.count()
if size is not None and size != -1:
return (count, result.limit(size).offset(offset))
return self.session.query(ocloud.ResourcePool).filter_by(
resourcePoolId=resource_pool_id).first()
- def _list(self, **kwargs) -> Tuple[int, List[ocloud.ResourcePool]]:
+ def _list(self, *args, **kwargs) -> Tuple[int, List[ocloud.ResourcePool]]:
size = kwargs.pop('limit') if 'limit' in kwargs else None
offset = kwargs.pop('start') if 'start' in kwargs else 0
- result = self.session.query(ocloud.ResourcePool).filter_by(
- **kwargs).order_by('resourcePoolId')
+ result = self.session.query(ocloud.ResourcePool).filter(
+ *args).order_by('resourcePoolId')
count = result.count()
if size is not None and size != -1:
return (count, result.limit(size).offset(offset))
return res
return recursive(resource_id)
- def _list(self, resourcepool_id, **kwargs) -> \
+ def _list(self, resourcepool_id, *args, **kwargs) -> \
Tuple[int, List[ocloud.Resource]]:
if 'sort' in kwargs:
kwargs.pop('sort')
size = kwargs.pop('limit') if 'limit' in kwargs else None
offset = kwargs.pop('start') if 'start' in kwargs else 0
- result = self.session.query(ocloud.Resource).filter_by(
- resourcePoolId=resourcepool_id, **kwargs).order_by('resourceId')
+ result = self.session.query(ocloud.Resource).filter(
+ *args).order_by('resourceId')
count = result.count()
if size is not None and size != -1:
return (count, result.limit(size).offset(offset))
return self.session.query(ocloud.DeploymentManager).filter_by(
deploymentManagerId=deployment_manager_id).first()
- def _list(self, **kwargs) -> Tuple[int, List[ocloud.DeploymentManager]]:
+ def _list(self, *args, **kwargs) -> Tuple[int,
+ List[ocloud.DeploymentManager]]:
size = kwargs.pop('limit') if 'limit' in kwargs else None
offset = kwargs.pop('start') if 'start' in kwargs else 0
- result = self.session.query(ocloud.DeploymentManager).filter_by(
- **kwargs).order_by('deploymentManagerId')
+ result = self.session.query(ocloud.DeploymentManager).filter(
+ *args).order_by('deploymentManagerId')
count = result.count()
if size is not None and size != -1:
return (count, result.limit(size).offset(offset))
return self.session.query(subscription_obj.Subscription).filter_by(
subscriptionId=subscription_id).first()
- def _list(self, **kwargs) -> \
+ def _list(self, *args, **kwargs) -> \
Tuple[int, List[subscription_obj.Subscription]]:
size = kwargs.pop('limit') if 'limit' in kwargs else None
offset = kwargs.pop('start') if 'start' in kwargs else 0
- result = self.session.query(subscription_obj.Subscription).filter_by(
- **kwargs).order_by('subscriptionId')
+ result = self.session.query(subscription_obj.Subscription).filter(
+ *args).order_by('subscriptionId')
count = result.count()
if size is not None and size != -1:
return (count, result.limit(size).offset(offset))
return alarm_event_record
def list(self, **kwargs) -> List[obj.AlarmEventRecord]:
- return self._list(**kwargs)[1]
+ return self._list(*[], **kwargs)[1]
- def list_with_count(self, **kwargs) -> \
+ def list_with_count(self, *args, **kwargs) -> \
Tuple[int, List[obj.AlarmEventRecord]]:
- return self._list(**kwargs)
+ return self._list(*args, **kwargs)
def update(self, alarm_event_record: obj.AlarmEventRecord):
self._update(alarm_event_record)
return subscription
def list(self, **kwargs) -> List[obj.AlarmSubscription]:
- return self._list(**kwargs)[1]
+ return self._list(*[], **kwargs)[1]
- def list_with_count(self, **kwargs) -> \
+ def list_with_count(self, *args, **kwargs) -> \
Tuple[int, List[obj.AlarmSubscription]]:
- return self._list(**kwargs)
+ return self._list(*args, **kwargs)
def update(self, subscription: obj.AlarmSubscription):
self._update(subscription)
return resource_type
def list(self, **kwargs) -> List[ocloud.ResourceType]:
- return self._list(**kwargs)[1]
+ return self._list(*[], **kwargs)[1]
- def list_with_count(self, **kwargs) -> \
+ def list_with_count(self, *args, **kwargs) -> \
Tuple[int, List[ocloud.ResourceType]]:
- return self._list(**kwargs)
+ return self._list(*args, **kwargs)
def update(self, resource_type: ocloud.ResourceType):
self._update(resource_type)
return resource_pool
def list(self, **kwargs) -> List[ocloud.ResourcePool]:
- return self._list(**kwargs)[1]
+ return self._list(*[], **kwargs)[1]
- def list_with_count(self, **kwargs) -> \
+ def list_with_count(self, *args, **kwargs) -> \
Tuple[int, List[ocloud.ResourcePool]]:
- return self._list(**kwargs)
+ return self._list(*args, **kwargs)
def update(self, resource_pool: ocloud.ResourcePool):
self._update(resource_pool)
return resource
def list(self, resourcepool_id, **kwargs) -> List[ocloud.Resource]:
- return self._list(resourcepool_id, **kwargs)[1]
+ return self._list(resourcepool_id, *[], **kwargs)[1]
- def list_with_count(self, resourcepool_id, **kwargs) -> \
+ def list_with_count(self, resourcepool_id, *args, **kwargs) -> \
Tuple[int, List[ocloud.Resource]]:
- return self._list(resourcepool_id, **kwargs)
+ return self._list(resourcepool_id, *args, **kwargs)
def update(self, resource: ocloud.Resource):
self._update(resource)
return deployment_manager
def list(self, **kwargs) -> List[ocloud.DeploymentManager]:
- return self._list(**kwargs)[1]
+ return self._list(*[], **kwargs)[1]
- def list_with_count(self, **kwargs) -> \
+ def list_with_count(self, *args, **kwargs) -> \
Tuple[int, List[ocloud.DeploymentManager]]:
- return self._list(**kwargs)
+ return self._list(*args, **kwargs)
def update(self, deployment_manager: ocloud.DeploymentManager):
self._update(deployment_manager)
return subscription
def list(self, **kwargs) -> List[subobj.Subscription]:
- return self._list(**kwargs)[1]
+ return self._list(*[], **kwargs)[1]
- def list_with_count(self, **kwargs) -> \
+ def list_with_count(self, *args, **kwargs) -> \
Tuple[int, List[subobj.Subscription]]:
- return self._list(**kwargs)
+ return self._list(*args, **kwargs)
def update(self, subscription: subobj.Subscription):
self._update(subscription)
_in='query')
@api_monitoring_v1.param(
'fields',
- 'Set fields to show, split by comman, "/" for parent and children.' +
+ 'Set fields to show, split by comma, "/" for parent and children.' +
' Like "name,parent/children". This value will cover' +
' "exculde_fields".',
_in='query')
@api_monitoring_v1.param(
'exclude_fields',
- 'Set fields to exclude showing, split by comman, "/" for parent and ' +
+ 'Set fields to exclude showing, split by comma, "/" for parent and ' +
'children. Like "name,parent/children". This value will cover ' +
'"exclude_default".',
_in='query')
'exclude_default',
'Exclude showing all default fields, Set "true" to enable.',
_in='query')
+@api_monitoring_v1.param(
+ 'exclude_default',
+ 'Exclude showing all default fields, Set "true" to enable.',
+ _in='query')
class AlarmListRouter(Resource):
model = AlarmDTO.alarm_event_record_get
def get(self):
parser = reqparse.RequestParser()
parser.add_argument(PAGE_PARAM, location='args')
+ parser.add_argument('filter', location='args')
args = parser.parse_args()
kwargs = {}
if args.nextpage_opaque_marker is not None:
kwargs['page'] = args.nextpage_opaque_marker
+ kwargs['filter'] = args.filter if args.filter is not None else ''
ret = alarm_view.alarm_event_records(bus.uow, **kwargs)
return link_header(request.full_path, ret)
_in='query')
@api_monitoring_v1.param(
'fields',
- 'Set fields to show, split by comman, "/" for parent and children.' +
+ 'Set fields to show, split by comma, "/" for parent and children.' +
' Like "name,parent/children". This value will cover' +
' "exculde_fields".',
_in='query')
@api_monitoring_v1.param(
'exclude_fields',
- 'Set fields to exclude showing, split by comman, "/" for parent and ' +
+ 'Set fields to exclude showing, split by comma, "/" for parent and ' +
'children. Like "name,parent/children". This value will cover ' +
'"exclude_default".',
_in='query')
_in='query')
@api_monitoring_v1.param(
'fields',
- 'Set fields to show, split by comman, "/" for parent and children.' +
+ 'Set fields to show, split by comma, "/" for parent and children.' +
' Like "name,parent/children". This value will cover' +
' "exculde_fields".',
_in='query')
@api_monitoring_v1.param(
'exclude_fields',
- 'Set fields to exclude showing, split by comman, "/" for parent and ' +
+ 'Set fields to exclude showing, split by comma, "/" for parent and ' +
'children. Like "name,parent/children". This value will cover ' +
'"exclude_default".',
_in='query')
'exclude_default',
'Exclude showing all default fields, Set "true" to enable.',
_in='query')
+ @api_monitoring_v1.param(
+ 'exclude_default',
+ 'Exclude showing all default fields, Set "true" to enable.',
+ _in='query')
def get(self):
parser = reqparse.RequestParser()
parser.add_argument(PAGE_PARAM, location='args')
+ parser.add_argument('filter', location='args')
args = parser.parse_args()
kwargs = {}
if args.nextpage_opaque_marker is not None:
kwargs['page'] = args.nextpage_opaque_marker
+ kwargs['filter'] = args.filter if args.filter is not None else ''
ret = alarm_view.subscriptions(bus.uow, **kwargs)
return link_header(request.full_path, ret)
_in='query')
@api_monitoring_v1.param(
'fields',
- 'Set fields to show, split by comman, "/" for parent and children.' +
+ 'Set fields to show, split by comma, "/" for parent and children.' +
' Like "name,parent/children". This value will cover' +
' "exculde_fields".',
_in='query')
@api_monitoring_v1.param(
'exclude_fields',
- 'Set fields to exclude showing, split by comman, "/" for parent and ' +
+ 'Set fields to exclude showing, split by comma, "/" for parent and ' +
'children. Like "name,parent/children". This value will cover ' +
'"exclude_default".',
_in='query')
from o2common.service import unit_of_work
from o2common.views.pagination_view import Pagination
+from o2common.views.view import gen_filter
from o2ims.views.alarm_dto import SubscriptionDTO
-from o2ims.domain.alarm_obj import AlarmSubscription
+from o2ims.domain.alarm_obj import AlarmSubscription, AlarmEventRecord
from o2common.helper import o2logging
# from o2common.config import config
def alarm_event_records(uow: unit_of_work.AbstractUnitOfWork, **kwargs):
pagination = Pagination(**kwargs)
filter_kwargs = pagination.get_filter()
+ args = gen_filter(AlarmEventRecord,
+ kwargs['filter']) if 'filter' in kwargs else []
with uow:
- li = uow.alarm_event_records.list_with_count(**filter_kwargs)
+ li = uow.alarm_event_records.list_with_count(*args, **filter_kwargs)
return pagination.get_result(li)
def subscriptions(uow: unit_of_work.AbstractUnitOfWork, **kwargs):
pagination = Pagination(**kwargs)
filter_kwargs = pagination.get_filter()
+ args = gen_filter(AlarmSubscription,
+ kwargs['filter']) if 'filter' in kwargs else []
with uow:
- li = uow.alarm_subscriptions.list_with_count(**filter_kwargs)
+ li = uow.alarm_subscriptions.list_with_count(*args, **filter_kwargs)
return pagination.get_result(li)
_in='query')
@api_ims_inventory_v1.param(
'fields',
- 'Set fields to show, split by comman, "/" for parent and children.' +
+ 'Set fields to show, split by comma, "/" for parent and children.' +
' Like "name,parent/children". This value will cover "exculde_fields".',
_in='query')
@api_ims_inventory_v1.param(
'exclude_fields',
- 'Set fields to exclude showing, split by comman, "/" for parent and ' +
+ 'Set fields to exclude showing, split by comma, "/" for parent and ' +
'children. Like "name,parent/children". This value will cover ' +
'"exclude_default".',
_in='query')
_in='query')
@api_ims_inventory_v1.param(
'fields',
- 'Set fields to show, split by comman, "/" for parent and children.' +
+ 'Set fields to show, split by comma, "/" for parent and children.' +
' Like "name,parent/children". This value will cover "exculde_fields".',
_in='query')
@api_ims_inventory_v1.param(
'exclude_fields',
- 'Set fields to exclude showing, split by comman, "/" for parent and ' +
+ 'Set fields to exclude showing, split by comma, "/" for parent and ' +
'children. Like "name,parent/children". This value will cover ' +
'"exclude_default".',
_in='query')
'exclude_default',
'Exclude showing all default fields, Set "true" to enable.',
_in='query')
+@api_ims_inventory_v1.param(
+ 'filter',
+ 'Filter of the query.',
+ _in='query')
class ResourceTypesListRouter(Resource):
model = ResourceTypeDTO.resource_type_get
def get(self):
parser = reqparse.RequestParser()
parser.add_argument(PAGE_PARAM, location='args')
+ parser.add_argument('filter', location='args')
args = parser.parse_args()
kwargs = {}
if args.nextpage_opaque_marker is not None:
kwargs['page'] = args.nextpage_opaque_marker
+ kwargs['filter'] = args.filter if args.filter is not None else ''
ret = ocloud_view.resource_types(bus.uow, **kwargs)
return link_header(request.full_path, ret)
_in='query')
@api_ims_inventory_v1.param(
'fields',
- 'Set fields to show, split by comman, "/" for parent and children.' +
+ 'Set fields to show, split by comma, "/" for parent and children.' +
' Like "name,parent/children". This value will cover "exculde_fields".',
_in='query')
@api_ims_inventory_v1.param(
'exclude_fields',
- 'Set fields to exclude showing, split by comman, "/" for parent and ' +
+ 'Set fields to exclude showing, split by comma, "/" for parent and ' +
'children. Like "name,parent/children". This value will cover ' +
'"exclude_default".',
_in='query')
_in='query')
@api_ims_inventory_v1.param(
'fields',
- 'Set fields to show, split by comman, "/" for parent and children.' +
+ 'Set fields to show, split by comma, "/" for parent and children.' +
' Like "name,parent/children". This value will cover "exculde_fields".',
_in='query')
@api_ims_inventory_v1.param(
'exclude_fields',
- 'Set fields to exclude showing, split by comman, "/" for parent and ' +
+ 'Set fields to exclude showing, split by comma, "/" for parent and ' +
'children. Like "name,parent/children". This value will cover ' +
'"exclude_default".',
_in='query')
'exclude_default',
'Exclude showing all default fields, Set "true" to enable.',
_in='query')
+@api_ims_inventory_v1.param(
+ 'filter',
+ 'Filter of the query.',
+ _in='query')
class ResourcePoolsListRouter(Resource):
model = ResourcePoolDTO.resource_pool_get
def get(self):
parser = reqparse.RequestParser()
parser.add_argument(PAGE_PARAM, location='args')
+ parser.add_argument('filter', location='args')
args = parser.parse_args()
kwargs = {}
if args.nextpage_opaque_marker is not None:
kwargs['page'] = args.nextpage_opaque_marker
+ kwargs['filter'] = args.filter if args.filter is not None else ''
ret = ocloud_view.resource_pools(bus.uow, **kwargs)
return link_header(request.full_path, ret)
_in='query')
@api_ims_inventory_v1.param(
'fields',
- 'Set fields to show, split by comman, "/" for parent and children.' +
+ 'Set fields to show, split by comma, "/" for parent and children.' +
' Like "name,parent/children". This value will cover "exculde_fields".',
_in='query')
@api_ims_inventory_v1.param(
'exclude_fields',
- 'Set fields to exclude showing, split by comman, "/" for parent and ' +
+ 'Set fields to exclude showing, split by comma, "/" for parent and ' +
'children. Like "name,parent/children". This value will cover ' +
'"exclude_default".',
_in='query')
# ---------- Resources ---------- #
@api_ims_inventory_v1.route("/resourcePools/<resourcePoolID>/resources")
@api_ims_inventory_v1.param('resourcePoolID', 'ID of the resource pool')
-@api_ims_inventory_v1.param('resourceTypeName', 'filter resource type',
- _in='query')
-@api_ims_inventory_v1.param('parentId', 'filter parentId',
- _in='query')
# @api_ims_inventory_v1.param('sort', 'sort by column name',
# _in='query')
# @api_ims_inventory_v1.param('per_page', 'The number of results per page ' +
_in='query')
@api_ims_inventory_v1.param(
'fields',
- 'Set fields to show, split by comman, "/" for parent and children.' +
+ 'Set fields to show, split by comma, "/" for parent and children.' +
' Like "name,parent/children". This value will cover "exculde_fields".',
_in='query')
@api_ims_inventory_v1.param(
'exclude_fields',
- 'Set fields to exclude showing, split by comman, "/" for parent and ' +
+ 'Set fields to exclude showing, split by comma, "/" for parent and ' +
'children. Like "name,parent/children". This value will cover ' +
'"exclude_default".',
_in='query')
'exclude_default',
'Exclude showing all default fields, Set "true" to enable.',
_in='query')
+@api_ims_inventory_v1.param(
+ 'filter',
+ 'Filter of the query.',
+ _in='query')
class ResourcesListRouter(Resource):
model = ResourceDTO.resource_list
@api_ims_inventory_v1.marshal_list_with(model)
def get(self, resourcePoolID):
parser = reqparse.RequestParser()
- parser.add_argument('resourceTypeName', location='args')
- parser.add_argument('parentId', location='args')
parser.add_argument(PAGE_PARAM, location='args')
+ parser.add_argument('filter', location='args')
args = parser.parse_args()
kwargs = {}
- if args.resourceTypeName is not None:
- kwargs['resourceTypeName'] = args.resourceTypeName
- if args.parentId is not None:
- kwargs['parentId'] = args.parentId
- if args.parentId.lower() == 'null':
- kwargs['parentId'] = None
# if args.per_page is not None:
# kwargs['per_page'] = args.per_page
# base_url = base_url + 'per_page=' + args.per_page + '&'
if args.nextpage_opaque_marker is not None:
kwargs['page'] = args.nextpage_opaque_marker
+ kwargs['filter'] = args.filter if args.filter is not None else ''
ret = ocloud_view.resources(resourcePoolID, bus.uow, **kwargs)
return link_header(request.full_path, ret)
_in='query')
@api_ims_inventory_v1.param(
'fields',
- 'Set fields to show, split by comman, "/" for parent and children.' +
+ 'Set fields to show, split by comma, "/" for parent and children.' +
' Like "name,parent/children". This value will cover "exculde_fields".',
_in='query')
@api_ims_inventory_v1.param(
'exclude_fields',
- 'Set fields to exclude showing, split by comman, "/" for parent and ' +
+ 'Set fields to exclude showing, split by comma, "/" for parent and ' +
'children. Like "name,parent/children". This value will cover ' +
'"exclude_default".',
_in='query')
_in='query')
@api_ims_inventory_v1.param(
'fields',
- 'Set fields to show, split by comman, "/" for parent and children.' +
+ 'Set fields to show, split by comma, "/" for parent and children.' +
' Like "name,parent/children". This value will cover "exculde_fields".',
_in='query')
@api_ims_inventory_v1.param(
'exclude_fields',
- 'Set fields to exclude showing, split by comman, "/" for parent and ' +
+ 'Set fields to exclude showing, split by comma, "/" for parent and ' +
'children. Like "name,parent/children". This value will cover ' +
'"exclude_default".',
_in='query')
'exclude_default',
'Exclude showing all default fields, Set "true" to enable.',
_in='query')
+@api_ims_inventory_v1.param(
+ 'filter',
+ 'Filter of the query.',
+ _in='query')
class DeploymentManagersListRouter(Resource):
model = DeploymentManagerDTO.deployment_manager_list
def get(self):
parser = reqparse.RequestParser()
parser.add_argument(PAGE_PARAM, location='args')
+ parser.add_argument('filter', location='args')
args = parser.parse_args()
kwargs = {}
if args.nextpage_opaque_marker is not None:
kwargs['page'] = args.nextpage_opaque_marker
+ kwargs['filter'] = args.filter if args.filter is not None else ''
ret = ocloud_view.deployment_managers(bus.uow, **kwargs)
return link_header(request.full_path, ret)
_in='query')
@api_ims_inventory_v1.param(
'fields',
- 'Set fields to show, split by comman, "/" for parent and children.' +
+ 'Set fields to show, split by comma, "/" for parent and children.' +
' Like "name,parent/children". This value will cover "exculde_fields".',
_in='query')
@api_ims_inventory_v1.param(
'exclude_fields',
- 'Set fields to exclude showing, split by comman, "/" for parent and ' +
+ 'Set fields to exclude showing, split by comma, "/" for parent and ' +
'children. Like "name,parent/children". This value will cover ' +
'"exclude_default".',
_in='query')
_in='query')
@api_ims_inventory_v1.param(
'fields',
- 'Set fields to show, split by comman, "/" for parent and children.' +
+ 'Set fields to show, split by comma, "/" for parent and children.' +
' Like "name,parent/children". This value will cover' +
' "exculde_fields".',
_in='query')
@api_ims_inventory_v1.param(
'exclude_fields',
- 'Set fields to exclude showing, split by comman, "/" for parent and ' +
+ 'Set fields to exclude showing, split by comma, "/" for parent and ' +
'children. Like "name,parent/children". This value will cover ' +
'"exclude_default".',
_in='query')
'exclude_default',
'Exclude showing all default fields, Set "true" to enable.',
_in='query')
+ @api_ims_inventory_v1.param(
+ 'filter',
+ 'Filter of the query.',
+ _in='query')
def get(self):
parser = reqparse.RequestParser()
parser.add_argument(PAGE_PARAM, location='args')
+ parser.add_argument('filter', location='args')
args = parser.parse_args()
kwargs = {}
if args.nextpage_opaque_marker is not None:
kwargs['page'] = args.nextpage_opaque_marker
+ kwargs['filter'] = args.filter if args.filter is not None else ''
ret = ocloud_view.subscriptions(bus.uow, **kwargs)
return link_header(request.full_path, ret)
_in='query')
@api_ims_inventory_v1.param(
'fields',
- 'Set fields to show, split by comman, "/" for parent and children.' +
+ 'Set fields to show, split by comma, "/" for parent and children.' +
' Like "name,parent/children". This value will cover' +
' "exculde_fields".',
_in='query')
@api_ims_inventory_v1.param(
'exclude_fields',
- 'Set fields to exclude showing, split by comman, "/" for parent and ' +
+ 'Set fields to exclude showing, split by comma, "/" for parent and ' +
'children. Like "name,parent/children". This value will cover ' +
'"exclude_default".',
_in='query')
from o2common.service import unit_of_work
from o2common.config import config
from o2common.views.pagination_view import Pagination
+from o2common.views.view import gen_filter
from o2ims.domain import ocloud
from o2ims.views.ocloud_dto import SubscriptionDTO
from o2ims.domain.subscription_obj import Subscription
def resource_types(uow: unit_of_work.AbstractUnitOfWork, **kwargs):
pagination = Pagination(**kwargs)
filter_kwargs = pagination.get_filter()
+ args = gen_filter(ocloud.ResourceType,
+ kwargs['filter']) if 'filter' in kwargs else []
with uow:
- li = uow.resource_types.list_with_count(**filter_kwargs)
+ li = uow.resource_types.list_with_count(*args, **filter_kwargs)
return pagination.get_result(li)
def resource_pools(uow: unit_of_work.AbstractUnitOfWork, **kwargs):
pagination = Pagination(**kwargs)
filter_kwargs = pagination.get_filter()
+ args = gen_filter(ocloud.ResourcePool,
+ kwargs['filter']) if 'filter' in kwargs else []
with uow:
- li = uow.resource_pools.list_with_count(**filter_kwargs)
+ li = uow.resource_pools.list_with_count(*args, **filter_kwargs)
return pagination.get_result(li)
res_type = uow.resource_types.get_by_name(resource_type_name)
restype_id = '' if res_type is None else res_type.resourceTypeId
filter_kwargs['resourceTypeId'] = restype_id
+ args = gen_filter(
+ ocloud.Resource, kwargs['filter']) if 'filter' in kwargs else []
+ args.append(ocloud.Resource.resourcePoolId == resourcePoolId)
+ # args.append(ocloud.Resource.parentId == None)
if 'parentId' in kwargs:
filter_kwargs['parentId'] = kwargs['parentId']
filter_kwargs['sort'] = kwargs['sort']
with uow:
- ret = uow.resources.list_with_count(resourcePoolId, **filter_kwargs)
+ ret = uow.resources.list_with_count(
+ resourcePoolId, *args, **filter_kwargs)
return pagination.get_result(ret)
def deployment_managers(uow: unit_of_work.AbstractUnitOfWork, **kwargs):
pagination = Pagination(**kwargs)
filter_kwargs = pagination.get_filter()
+ args = gen_filter(ocloud.DeploymentManager,
+ kwargs['filter']) if 'filter' in kwargs else []
with uow:
- li = uow.deployment_managers.list_with_count(**filter_kwargs)
+ li = uow.deployment_managers.list_with_count(*args, **filter_kwargs)
return pagination.get_result(li)
def subscriptions(uow: unit_of_work.AbstractUnitOfWork, **kwargs):
pagination = Pagination(**kwargs)
filter_kwargs = pagination.get_filter()
+ args = gen_filter(ocloud.DeploymentManager,
+ kwargs['filter']) if 'filter' in kwargs else []
with uow:
- li = uow.subscriptions.list_with_count(**filter_kwargs)
+ li = uow.subscriptions.list_with_count(*args, **filter_kwargs)
return pagination.get_result(li)
order_by = MagicMock()
order_by.count.return_value = 1
order_by.limit.return_value.offset.return_value = [alarm_event_record1]
- session.return_value.query.return_value.filter_by.return_value.\
+ session.return_value.query.return_value.filter.return_value.\
order_by.return_value = order_by
result = alarm_view.alarm_event_records(uow)
order_by = MagicMock()
order_by.count.return_value = 1
order_by.limit.return_value.offset.return_value = [sub1]
- session.return_value.query.return_value.filter_by.return_value.\
+ session.return_value.query.return_value.filter.return_value.\
order_by.return_value = order_by
result = alarm_view.subscriptions(uow)
order_by = MagicMock()
order_by.count.return_value = 0
order_by.limit.return_value.offset.return_value = []
- session.return_value.query.return_value.filter_by.return_value.\
+ session.return_value.query.return_value.filter.return_value.\
order_by.return_value = order_by
apibase = config.get_o2ims_monitoring_api_base()
order_by = MagicMock()
order_by.count.return_value = 1
order_by.limit.return_value.offset.return_value = [restype1]
- session.return_value.query.return_value.filter_by.return_value.\
+ session.return_value.query.return_value.filter.return_value.\
order_by.return_value = order_by
result = ocloud_view.resource_types(uow)
order_by = MagicMock()
order_by.count.return_value = 1
order_by.limit.return_value.offset.return_value = [respool1]
- session.return_value.query.return_value.filter_by.return_value.\
+ session.return_value.query.return_value.filter.return_value.\
order_by.return_value = order_by
result = ocloud_view.resource_pools(uow)
order_by = MagicMock()
order_by.count.return_value = 1
order_by.limit.return_value.offset.return_value = [res1]
- session.return_value.query.return_value.filter_by.return_value.\
+ session.return_value.query.return_value.filter.return_value.\
order_by.return_value = order_by
+ # TODO: workaround for sqlalchemy not mapping with resource object
+ setattr(ocloud.Resource, 'resourcePoolId', '')
result = ocloud_view.resources(resource_pool_id1, uow)
assert result['count'] == 1
order_by = MagicMock()
order_by.count.return_value = 1
order_by.limit.return_value.offset.return_value = [dm1]
- session.return_value.query.return_value.filter_by.return_value.\
+ session.return_value.query.return_value.filter.return_value.\
order_by.return_value = order_by
result = ocloud_view.deployment_managers(uow)
order_by = MagicMock()
order_by.count.return_value = 1
order_by.limit.return_value.offset.return_value = [sub1]
- session.return_value.query.return_value.filter_by.return_value.\
+ session.return_value.query.return_value.filter.return_value.\
order_by.return_value = order_by
result = ocloud_view.subscriptions(uow)
order_by = MagicMock()
order_by.count.return_value = 0
order_by.limit.return_value.offset.return_value = []
- session.return_value.query.return_value.filter_by.return_value.\
+ session.return_value.query.return_value.filter.return_value.\
order_by.return_value = order_by
apibase = config.get_o2ims_api_base()
+ # TODO: workaround for sqlalchemy not mapping with resource object
+ setattr(ocloud.Resource, 'resourcePoolId', '')
with app.test_client() as client:
# Get list and return empty list
Code Review / pti / o2.git / commitdiff