Skip to content

Commit

Permalink
Add service_catalog_info module
Browse files Browse the repository at this point in the history
This module retreive information about service catalog along with
categories and items
  • Loading branch information
tupyy committed Apr 16, 2024
1 parent ee60349 commit a774ccc
Show file tree
Hide file tree
Showing 2 changed files with 442 additions and 0 deletions.
162 changes: 162 additions & 0 deletions plugins/module_utils/service_catalog.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,162 @@
# -*- coding: utf-8 -*-
# Copyright: (c) 2024, Red Hat
#
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)

from __future__ import absolute_import, division, print_function

__metaclass__ = type


SN_BASE_PATH = "api/sn_sc/servicecatalog"


class ItemContent(object):
FULL = 1
BRIEF = 2
NONE = 3

@classmethod
def from_str(self, s):
if s == "full":
return ItemContent.FULL
if s == "brief":
return ItemContent.BRIEF
return ItemContent.NONE


class ServiceCatalogObject(object):
def to_ansible(self):
"""Filters out the fields which we don't want to return like `header_icon`"""
ansible_data = dict()
for key in self.DISPLAY_FIELDS:
if key in self.data:
if isinstance(self.data[key], ServiceCatalogObject):
ansible_data[key] = self.data[key].to_ansible()
continue
if type(self.data[key]) is list:
ansible_data[key] = []
for item in self.data[key]:
if isinstance(item, ServiceCatalogObject):
ansible_data[key].append(item.to_ansible())
else:
ansible_data[key].append(item)
continue
ansible_data[key] = self.data[key]
return ansible_data

def valid(self):
for key in self.MANDATORY_FIELS:
if key not in self.data or not self.data[key]:
return False
return True

@property
def sys_id(self):
return self.data["sys_id"] if "sys_id" in self.data else ""


class Catalog(ServiceCatalogObject):
DISPLAY_FIELDS = ["sys_id", "description", "title",
"has_categories", "has_items", "categories", "items"]
MANDATORY_FIELS = ["sys_id"]

def __init__(self, data=dict()):
self.data = data
self._categories = []
self._items = []

@property
def categories(self):
return self._categories

@categories.setter
def categories(self, categories):
self._categories = categories

@property
def items(self):
return self._items

@items.setter
def items(self, items):
self._items = items

def to_ansible(self):
self.data["categories"] = self.categories
self.data["items"] = self.items
return super().to_ansible()


class Category(ServiceCatalogObject):
DISPLAY_FIELDS = ["sys_id", "description", "title",
"full_description", "subcategories"]

def __init__(self, data=dict()):
self.data = data


class Item(ServiceCatalogObject):
DISPLAY_FIELDS = ["sys_id", "short_description", "description",
"availabiltiy", "mandatory_attachment", "request_method",
"type", "sys_class_name", "catalogs", "name", "category", "order",
"categories", "variables"]

def __init__(self, data=dict()):
self.data = data


class ServiceCatalogClient(object):
"""Wraps the generic client with Service Catalog specific methods"""

def __init__(self, generic_client):
if not generic_client:
raise ValueError("generic client cannot be none")
self.generic_client = generic_client

def get_catalogs(self):
"""Returns the list of all catalogs"""
records = self.generic_client.list_records("/".join([SN_BASE_PATH, "catalogs"]))
if records:
return [Catalog(record) for record in records]
return []

def get_catalog(self, id):
"""Returns the catalog identified by id"""
if not id:
raise ValueError("catalog sys_id is missing")
record = self.generic_client.get_record_by_sys_id("/".join([SN_BASE_PATH, "catalogs"]), id)
if record:
return Catalog(record)
return None

def get_categories(self, catalog_id):
"""Returns the list of all categories of the catalog `catalog_id`"""
if not id:
raise ValueError("catalog sys_id is missing")
records = self.generic_client.list_records(
"/".join([SN_BASE_PATH, "catalogs", catalog_id, "categories"]))
if records:
return [Category(record) for record in records]
return dict()

def get_items(self, catalog_id, query=None, batch_size=1000):
"""Returns the list of all items of the catalog `catalog_id`"""
if not id:
raise ValueError("catalog sys_id is missing")
_query = dict(sysparm_catalog=catalog_id)
if query:
_query.update(query)
self.generic_client.batch_size = batch_size
records = self.generic_client.list_records(
"/".join([SN_BASE_PATH, "items"]), _query)
if records:
return [Item(record) for record in records]
return dict()

def get_item(self, id):
if not id:
raise ValueError("item sys_id is missing")
return Item(self.generic_client.get_record_by_sys_id("/".join([SN_BASE_PATH, "items"]), id))


0 comments on commit a774ccc

Please sign in to comment.