-
Notifications
You must be signed in to change notification settings - Fork 81
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
This module retreive information about service catalog along with categories and items
- Loading branch information
Showing
2 changed files
with
442 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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)) | ||
|
||
|
Oops, something went wrong.