IBM Z Software Asset Management 8.3.1 API Showcase
Introduction
In this blog post, we will explore the new IBM Z Software Asset Management API available in the 8.3.1 release. We will demonstrate how to use the API by making requests and loading them into a Pandas DataFrame for downstream use cases. This guide assumes the API is installed, configured, and running on your z/OS machine.
Customization
For convenience and security, we store the username and password in a zsecrets.py file, as shown below:
UNAME = "<RACF uname>"
PWD = "<RACF passwords>"
API_PORT = "<listening port>"
API_HOST = "<hostname/ip>"
The username and password will authenticate each request using the RACF facility. The API_PORT and API_HOST specify the port and hostname/IP where your API instance is running.
Boilerplate Code
The following code provides a skeleton for interacting with the API. While errors are loosely handled, the code is explicit, enabling you to understand how requests are constructed.
In addition to the client performing the requests, we provide Pydantic classes to document the parameters needed for customizing each report request.
Parameters
Here, we define the Pydantic models. If you are unfamiliar with Pydantic, it is a library for validating input data based on Python type hints. Validation is skipped in this walkthrough, but you can reuse these models as needed.
from pydantic import BaseModel
from datetime import date
from typing import Optional, Literal
class ProductInventoryParameters(BaseModel):
vendor: Optional[str]
product: Optional[str]
system: Optional[str]
show_product_discovery_name: bool = False
show_subcap_values: bool = False
show_sysplex_names: bool = False
show_system_names: bool = False
show_product_suite_only: bool = False
class ProductEndOfServiceParameters(BaseModel):
vendor: Optional[str] = None
product_release: Optional[str] = None
start_date: date | Literal["earliest"] = "earliest"
end_date: date | Literal["latest"] = "latest"
show_sysplex_names: bool = False
show_system_names: bool = False
knowledge_base: Literal['GKB', 'GKU'] = 'GKB'
class ProductUseByMonthParameters(BaseModel):
vendor: Optional[str] = None
product: Optional[str] = None
system: Optional[str] = None
start_date: date | Literal["earliest"] = "earliest"
end_date: date | Literal["latest"] = "latest"
show_product_discovery_name: bool = False
show_s_s_pid: bool = False
metrics: Literal[
"MODULE_EVENTS",
"PRODUCT_USE",
"USER_ID_COUNT",
"JOB_NAME_COUNT",
"JOB_ACCOUNT_COUNT",
"SCRT_MSU"
] = "JOB_NAME_COUNT"
class ProductUseByMachineParameters(BaseModel):
vendor: Optional[str] = None
product_name: Optional[str] = None
machine: Optional[str] = None
start_date: date | Literal["earliest"] = "earliest"
end_date: date | Literal["latest"] = "latest"
show_product_discovery_name: bool = False
show_service_and_support_pid: bool = False
show_hardware_partition: bool = False
metrics: Literal[
"LAST_USE_DATE",
"FIRST_USE_DATE",
"MODULE_EVENTS",
"PRODUCT_USE",
"USER_ID_COUNT",
"JOB_NAME_COUNT",
"JOB_ACCOUNT_COUNT",
"SCRT_MSU"] = "LAST_USE_DATE"
Client
A Python client, ZsamClient, provides methods to interact with the API, including actions on repositories and reports.
from requests.auth import HTTPBasicAuth
import requests
import pandas as pd
from typing import List, Tuple, Dict, Optional
class ZsamClient:
__reports: Dict[str, Tuple[callable, str, BaseModel]] = dict(
product_inventory=(requests.get, '/api/products/inventory', ProductInventoryParameters),
end_of_service=(requests.get, '/api/products/end-of-service', ProductEndOfServiceParameters),
product_use_by_machine=(requests.get, '/api/products/usage/machine', ProductUseByMachineParameters),
product_use_by_month=(requests.get, '/api/products/usage/month', ProductUseByMonthParameters),
)
__repository_ep: Dict[str, Tuple[callable, str]] = dict(
refresh=(requests.patch, '/api/repositories/'),
get_list=(requests.get, '/api/repositories'),
add_batch=(requests.post, '/api/repositories/'),
add=(requests.put, '/api/repositories/:reponame'),
delete=(requests.delete, '/api/repositories/:reponame'),
get_tparam=(requests.get, '/api/repositories/:reponame')
)
def __init__(self, uname: str, pwd: str, hostname: str, port: str, secure: bool = True, verify: bool = False):
self.__prot: str = 'https' if secure else 'http'
self.__url: str = f'{self.__prot}://{hostname}:{port}'
self.common_req_params = dict(
auth=HTTPBasicAuth(uname, pwd),
verify=verify
)
def report(self, reportname: str, reponame: str, **params) -> Optional[pd.DataFrame]:
try:
method_fn, path, param_model = self.__reports[reportname]
except KeyError:
print(f"Report {reportname} is not available")
return None
validated_params: BaseModel = param_model(**params).dict()
validated_params["repository"] = reponame
req_path = self.__url + path
validated_params = {k: v for k, v in validated_params.items() if v is not None}
res = method_fn(req_path, params=validated_params, **self.common_req_params)
if res.status_code != 200:
return None
try:
body = res.json()
except requests.exceptions.JSONDecodeError:
print("Failed parsing the body")
return None
df = pd.DataFrame(body["result"], columns=body["columns"])
return df
def refresh(self) -> bool:
method_fn, path = self.__repository_ep['refresh']
req_path = self.__url + path
res = method_fn(req_path, **self.common_req_params)
return res.status_code == 200
def list(self) -> Tuple[List[str], int]:
method_fn, path = self.__repository_ep['get_list']
req_path = self.__url + path
res = method_fn(req_path, **self.common_req_params)
if res.status_code != 200:
raise Exception("Error fetching repository list")
body = res.json()
return body['repositories'], body['row_count']
def add(self, reponame: str) -> bool:
method_fn, path = self.__repository_ep['add']
req_path = self.__url + path.replace(":reponame", reponame)
res = method_fn(req_path, **self.common_req_params)
if res.status_code != 200:
raise Exception("Error adding repository")
return True
def delete(self, reponame: str) -> bool:
method_fn, path = self.__repository_ep['delete']
req_path = self.__url + path.replace(":reponame", reponame)
res = method_fn(req_path, **self.common_req_params)
return res.status_code == 200
def get_tparam(self, reponame: str) -> Dict[str, str]:
method_fn, path = self.__repository_ep['get_tparam']
req_path = self.__url + path.replace(":reponame", reponame)
res = method_fn(req_path, **self.common_req_params)
if res.status_code != 200:
raise Exception("Error fetching repository parameters")
return res.json()
Example Usage:
client = ZsamClient(UNAME, PWD, API_HOST, API_PORT, secure=True, verify=False)
Repository Management
Refreshing the repository list:
ok = client.refresh()
if not ok:
print("Something went wrong refreshing the repositories list")
Fetching repositories:
repositories, count = client.list()
for repo in repositories:
print(repo, end=", ")
Adding and deleting repositories:
ok = client.add(reponame)
ok = client.delete(reponame)
Querying Reports
Querying the "product_inventory" report:
df = client.report("product_inventory", "SC83REPZ")
df.head()
Filtering by Vendor:
df = client.report("product_inventory", "SC83REPZ", vendor="IBM")
df.head()
Data Insights
Counting rows:
len(df)
Counting vendors:
df["VENDOR"].value_counts()
Conclusion
This showcase highlights the flexibility of IBM Z Software Asset Management's API. For further details, refer to the official documentation:
#AIOpsonZ #APIstrategy #IBMz/OS