Chapter 4. API Requests in Different Languages
Chapter 4. API Requests in Different Languages This chapter outlines sending API requests to Red Hat Satellite with curl, Ruby, and Python and provides examples. 4.1. API Requests with curl This section outlines how to use curl with the Satellite API to perform various tasks. Red Hat Satellite requires the use of HTTPS, and by default a certificate for host identification. If you have not added the Satellite Server certificate as described in Section 3.1, "SSL Authentication Overview" , then you can use the --insecure option to bypass certificate checks. For user authentication, you can use the --user option to provide Satellite user credentials in the form --user username:password or, if you do not include the password, the command prompts you to enter it. To reduce security risks, do not include the password as part of the command, because it then becomes part of your shell history. Examples in this section include the password only for the sake of simplicity. Be aware that if you use the --silent option, curl does not display a progress meter or any error messages. Examples in this chapter use the Python json.tool module to format the output. 4.1.1. Passing JSON Data to the API Request You can pass data to Satellite Server with the API request. The data must be in JSON format. When specifying JSON data with the --data option, you must set the following HTTP headers with the --header option: Use one of the following options to include data with the --data option: The quoted JSON formatted data enclosed in curly braces {} . When passing a value for a JSON type parameter, you must escape quotation marks " with backslashes \ . For example, within curly braces, you must format "Example JSON Variable" as \"Example JSON Variable\" : The unquoted JSON formatted data enclosed in a file and specified by the @ sign and the filename. For example: Using external files for JSON formatted data has the following advantages: You can use your favorite text editor. You can use syntax checker to find and avoid mistakes. You can use tools to check the validity of JSON data or to reformat it. Validating a JSON file Use the json_verify tool to check the validity of a JSON file: 4.1.2. Retrieving a List of Resources This section outlines how to use curl with the Satellite 6 API to request information from your Satellite deployment. These examples include both requests and responses. Expect different results for each deployment. Note The example requests below use python3 to format the respone from the Satellite Server. On RHEL 7 and some older systems, you must use python instead of python3 . Listing Users This example is a basic request that returns a list of Satellite resources, Satellite users in this case. Such requests return a list of data wrapped in metadata, while other request types only return the actual object. Example request: Example response: 4.1.3. Creating and Modifying Resources This section outlines how to use curl with the Satellite 6 API to manipulate resources on the Satellite Server. These API calls require that you pass data in json format with the API call. For more information, see Section 4.1.1, "Passing JSON Data to the API Request" . Creating a User This example creates a user using --data option to provide required information. Example request: Modifying a User This example modifies first name and login of the test_user that was created in Creating a User . Example request: 4.2. API Requests with Ruby This section outlines how to use Ruby with the Satellite API to perform various tasks. Important These are example scripts and commands. Ensure you review these scripts carefully before use, and replace any variables, user names, passwords, and other information to suit your own deployment. 4.2.1. Creating Objects Using Ruby This script connects to the Red Hat Satellite 6 API and creates an organization, and then creates three environments in the organization. If the organization already exists, the script uses that organization. If any of the environments already exist in the organization, the script raises an error and quits. #!/usr/bin/ruby require 'rest-client' require 'json' url = 'https://satellite.example.com/api/v2/' katello_url = "#{url}/katello/api/v2/" USDusername = 'admin' USDpassword = 'changeme' org_name = "MyOrg" environments = [ "Development", "Testing", "Production" ] # Performs a GET using the passed URL location def get_json(location) response = RestClient::Request.new( :method => :get, :url => location, :user => USDusername, :password => USDpassword, :headers => { :accept => :json, :content_type => :json } ).execute JSON.parse(response.to_str) end # Performs a POST and passes the data to the URL location def post_json(location, json_data) response = RestClient::Request.new( :method => :post, :url => location, :user => USDusername, :password => USDpassword, :headers => { :accept => :json, :content_type => :json}, :payload => json_data ).execute JSON.parse(response.to_str) end # Creates a hash with ids mapping to names for an array of records def id_name_map(records) records.inject({}) do |map, record| map.update(record['id'] => record['name']) end end # Get list of existing organizations orgs = get_json("#{katello_url}/organizations") org_list = id_name_map(orgs['results']) if !org_list.has_value?(org_name) # If our organization is not found, create it puts "Creating organization: \t#{org_name}" org_id = post_json("#{katello_url}/organizations", JSON.generate({"name"=> org_name}))["id"] else # Our organization exists, so let's grab it org_id = org_list.key(org_name) puts "Organization \"#{org_name}\" exists" end # Get list of organization's lifecycle environments envs = get_json("#{katello_url}/organizations/#{org_id}/environments") env_list = id_name_map(envs['results']) prior_env_id = env_list.key("Library") # Exit the script if at least one life cycle environment already exists environments.each do |e| if env_list.has_value?(e) puts "ERROR: One of the Environments is not unique to organization" exit end end # Create life cycle environments environments.each do |environment| puts "Creating environment: \t#{environment}" prior_env_id = post_json("#{katello_url}/organizations/#{org_id}/environments", JSON.generate({"name" => environment, "organization_id" => org_id, "prior_id" => prior_env_id}))["id"] end 4.2.2. Using Apipie Bindings with Ruby Apipie bindings are the Ruby bindings for apipie documented API calls. They fetch and cache the API definition from Satellite and then generate API calls on demand. This example creates an organization, and then creates three environments in the organization. If the organization already exists, the script uses that organization. If any of the environments already exist in the organization, the script raises an error and quits. #!/usr/bin/tfm-ruby require 'apipie-bindings' org_name = "MyOrg" environments = [ "Development", "Testing", "Production" ] # Create an instance of apipie bindings @api = ApipieBindings::API.new({ :uri => 'https://satellite.example.com/', :username => 'admin', :password => 'changeme', :api_version => 2 }) # Performs an API call with default options def call_api(resource_name, action_name, params = {}) http_headers = {} apipie_options = { :skip_validation => true } @api.resource(resource_name).call(action_name, params, http_headers, apipie_options) end # Creates a hash with IDs mapping to names for an array of records def id_name_map(records) records.inject({}) do |map, record| map.update(record['id'] => record['name']) end end # Get list of existing organizations orgs = call_api(:organizations, :index) org_list = id_name_map(orgs['results']) if !org_list.has_value?(org_name) # If our organization is not found, create it puts "Creating organization: \t#{org_name}" org_id = call_api(:organizations, :create, {'organization' => { :name => org_name }})['id'] else # Our organization exists, so let's grab it org_id = org_list.key(org_name) puts "Organization \"#{org_name}\" exists" end # Get list of organization's life cycle environments envs = call_api(:lifecycle_environments, :index, {'organization_id' => org_id}) env_list = id_name_map(envs['results']) prior_env_id = env_list.key("Library") # Exit the script if at least one life cycle environment already exists environments.each do |e| if env_list.has_value?(e) puts "ERROR: One of the Environments is not unique to organization" exit end end # Create life cycle environments environments.each do |environment| puts "Creating environment: \t#{environment}" prior_env_id = call_api(:lifecycle_environments, :create, {"name" => environment, "organization_id" => org_id, "prior_id" => prior_env_id })['id'] end 4.3. API Requests with Python This section outlines how to use Python with the Satellite API to perform various tasks. Important These are example scripts and commands. Ensure you review these scripts carefully before use, and replace any variables, user names, passwords, and other information to suit your own deployment. Example scripts in this section do not use SSL verification for interacting with the REST API. 4.3.1. Creating Objects Using Python This script connects to the Red Hat Satellite 6 API and creates an organization, and then creates three environments in the organization. If the organization already exists, the script uses that organization. If any of the environments already exist in the organization, the script raises an error and quits. Python 2 Example #!/usr/bin/python import json import sys try: import requests except ImportError: print "Please install the python-requests module." sys.exit(-1) # URL to your Satellite 6 server URL = "https://satellite.example.com" # URL for the API to your deployed Satellite 6 server SAT_API = "%s/katello/api/v2/" % URL # Katello-specific API KATELLO_API = "%s/katello/api/" % URL POST_HEADERS = {'content-type': 'application/json'} # Default credentials to login to Satellite 6 USERNAME = "admin" PASSWORD = "changeme" # Ignore SSL for now SSL_VERIFY = False # Name of the organization to be either created or used ORG_NAME = "MyOrg" # Name for life cycle environments to be either created or used ENVIRONMENTS = ["Development", "Testing", "Production"] def get_json(location): """ Performs a GET using the passed URL location """ r = requests.get(location, auth=(USERNAME, PASSWORD), verify=SSL_VERIFY) return r.json() def post_json(location, json_data): """ Performs a POST and passes the data to the URL location """ result = requests.post( location, data=json_data, auth=(USERNAME, PASSWORD), verify=SSL_VERIFY, headers=POST_HEADERS) return result.json() def main(): """ Main routine that creates or re-uses an organization and life cycle environments. If life cycle environments already exist, exit out. """ # Check if our organization already exists org = get_json(SAT_API + "organizations/" + ORG_NAME) # If our organization is not found, create it if org.get('error', None): org_id = post_json( SAT_API + "organizations/", json.dumps({"name": ORG_NAME}))["id"] print "Creating organization: \t" + ORG_NAME else: # Our organization exists, so let's grab it org_id = org['id'] print "Organization '%s' exists." % ORG_NAME # Now, let's fetch all available life cycle environments for this org... envs = get_json( SAT_API + "organizations/" + str(org_id) + "/environments/") # ... and add them to a dictionary, with respective 'Prior' environment prior_env_id = 0 env_list = {} for env in envs['results']: env_list[env['id']] = env['name'] prior_env_id = env['id'] if env['name'] == "Library" else prior_env_id # Exit the script if at least one life cycle environment already exists if all(environment in env_list.values() for environment in ENVIRONMENTS): print "ERROR: One of the Environments is not unique to organization" sys.exit(-1) # Create life cycle environments for environment in ENVIRONMENTS: new_env_id = post_json( SAT_API + "organizations/" + str(org_id) + "/environments/", json.dumps( { "name": environment, "organization_id": org_id, "prior": prior_env_id} ))["id"] print "Creating environment: \t" + environment prior_env_id = new_env_id if __name__ == "__main__": main() 4.3.2. Requesting information from the API using Python This is an example script that uses Python for various API requests. Python 2 Example #!/usr/bin/python import json import sys try: import requests except ImportError: print "Please install the python-requests module." sys.exit(-1) SAT_API = 'https://satellite.example.com/api/v2/' USERNAME = "admin" PASSWORD = "password" SSL_VERIFY = False # Ignore SSL for now def get_json(url): # Performs a GET using the passed URL location r = requests.get(url, auth=(USERNAME, PASSWORD), verify=SSL_VERIFY) return r.json() def get_results(url): jsn = get_json(url) if jsn.get('error'): print "Error: " + jsn['error']['message'] else: if jsn.get('results'): return jsn['results'] elif 'results' not in jsn: return jsn else: print "No results found" return None def display_all_results(url): results = get_results(url) if results: print json.dumps(results, indent=4, sort_keys=True) def display_info_for_hosts(url): hosts = get_results(url) if hosts: for host in hosts: print "ID: %-10d Name: %-30s IP: %-20s OS: %-30s" % (host['id'], host['name'], host['ip'], host['operatingsystem_name']) def main(): host = 'satellite.example.com' print "Displaying all info for host %s ..." % host display_all_results(SAT_API + 'hosts/' + host) print "Displaying all facts for host %s ..." % host display_all_results(SAT_API + 'hosts/%s/facts' % host) host_pattern = 'example' print "Displaying basic info for hosts matching pattern '%s'..." % host_pattern display_info_for_hosts(SAT_API + 'hosts?search=' + host_pattern) environment = 'production' print "Displaying basic info for hosts in environment %s..." % environment display_info_for_hosts(SAT_API + 'hosts?search=environment=' + environment) model = 'RHEV Hypervisor' print "Displaying basic info for hosts with model name %s..." % model display_info_for_hosts(SAT_API + 'hosts?search=model="' + model + '"') if __name__ == "__main__": main() Python 3 Example #!/usr/bin/env python3 import json import sys try: import requests except ImportError: print("Please install the python-requests module.") sys.exit(-1) SAT = "satellite.example.com" # URL for the API to your deployed Satellite 6 server SAT_API = f"https://{SAT}/api/" KATELLO_API = f"https://{SAT}/katello/api/v2/" POST_HEADERS = {'content-type': 'application/json'} # Default credentials to login to Satellite 6 USERNAME = "admin" PASSWORD = "password" # Ignore SSL for now SSL_VERIFY = False #SSL_VERIFY = "./path/to/CA-certificate.crt" # Put the path to your CA certificate here to allow SSL_VERIFY def get_json(url): # Performs a GET using the passed URL location r = requests.get(url, auth=(USERNAME, PASSWORD), verify=SSL_VERIFY) return r.json() def get_results(url): jsn = get_json(url) if jsn.get('error'): print("Error: " + jsn['error']['message']) else: if jsn.get('results'): return jsn['results'] elif 'results' not in jsn: return jsn else: print("No results found") return None def display_all_results(url): results = get_results(url) if results: print(json.dumps(results, indent=4, sort_keys=True)) def display_info_for_hosts(url): hosts = get_results(url) if hosts: print(f"{'ID':10}{'Name':40}{'IP':30}{'Operating System':30}") for host in hosts: print(f"{str(host['id']):10}{host['name']:40}{str(host['ip']):30}{str(host['operatingsystem_name']):30}") def display_info_for_subs(url): subs = get_results(url) if subs: print(f"{'ID':10}{'Name':90}{'Start Date':30}") for sub in subs: print(f"{str(sub['id']):10}{sub['name']:90}{str(sub['start_date']):30}") def main(): host = SAT print(f"Displaying all info for host {host} ...") display_all_results(SAT_API + 'hosts/' + host) print(f"Displaying all facts for host {host} ...") display_all_results(SAT_API + f'hosts/{host}/facts') host_pattern = 'example' print(f"Displaying basic info for hosts matching pattern '{host_pattern}'...") display_info_for_hosts(SAT_API + 'hosts?per_page=1&search=name~' + host_pattern) print(f"Displaying basic info for subscriptions") display_info_for_subs(KATELLO_API + 'subscriptions') environment = 'production' print(f"Displaying basic info for hosts in environment {environment}...") display_info_for_hosts(SAT_API + 'hosts?search=environment=' + environment) if __name__ == "__main__": main()
--header "Accept:application/json" --header "Content-Type:application/json" --data {"id":44, "smart_class_parameter":{"override":"true", "parameter_type":"json", "default_value":"{\"GRUB_CMDLINE_LINUX\": {\"audit\":\"1\",\"crashkernel\":\"true\"}}"}} --data @ file .json json_verify < test_file .json curl --request GET --insecure --user sat_username:sat_password https:// satellite.example.com /api/users | python3 -m json.tool { "page": 1, "per_page": 20, "results": [ { "admin": false, "auth_source_id": 1, "auth_source_name": "Internal", "created_at": "2018-09-21 08:59:22 UTC", "default_location": null, "default_organization": null, "description": "", "effective_admin": false, "firstname": "", "id": 5, "last_login_on": "2018-09-21 09:03:25 UTC", "lastname": "", "locale": null, "locations": [], "login": "test", "mail": "example@domain.com", "organizations": [ { "id": 1, "name": "Default Organization" } ], "ssh_keys": [], "timezone": null, "updated_at": "2018-09-21 09:04:45 UTC" }, { "admin": true, "auth_source_id": 1, "auth_source_name": "Internal", "created_at": "2018-09-20 07:09:41 UTC", "default_location": null, "default_organization": { "description": null, "id": 1, "name": "Default Organization", "title": "Default Organization" }, "description": "", "effective_admin": true, "firstname": "Admin", "id": 4, "last_login_on": "2018-12-07 07:31:09 UTC", "lastname": "User", "locale": null, "locations": [ { "id": 2, "name": "Default Location" } ], "login": "admin", "mail": "root@example.com", "organizations": [ { "id": 1, "name": "Default Organization" } ], "ssh_keys": [], "timezone": null, "updated_at": "2018-11-14 08:19:46 UTC" } ], "search": null, "sort": { "by": null, "order": null }, "subtotal": 2, "total": 2 } curl --header "Accept:application/json" --header "Content-Type:application/json" --request POST --user sat_username:sat_password --insecure --data "{\"firstname\":\" Test Name \",\"mail\":\" test@example.com \",\"login\":\" test_user \",\"password\":\" password123 \",\"auth_source_id\": 1 }" https:// satellite.example.com /api/users | python3 -m json.tool curl --header "Accept:application/json" --header "Content-Type:application/json" --request PUT --user sat_username:sat_password --insecure --data "{\"firstname\":\" New Test Name \",\"mail\":\" test@example.com \",\"login\":\" new_test_user \",\"password\":\" password123 \",\"auth_source_id\": 1 }" https:// satellite.example.com /api/users/ 8 | python3 -m json.tool #!/usr/bin/ruby require 'rest-client' require 'json' url = 'https://satellite.example.com/api/v2/' katello_url = "#{url}/katello/api/v2/" USDusername = 'admin' USDpassword = 'changeme' org_name = "MyOrg" environments = [ "Development", "Testing", "Production" ] Performs a GET using the passed URL location def get_json(location) response = RestClient::Request.new( :method => :get, :url => location, :user => USDusername, :password => USDpassword, :headers => { :accept => :json, :content_type => :json } ).execute JSON.parse(response.to_str) end Performs a POST and passes the data to the URL location def post_json(location, json_data) response = RestClient::Request.new( :method => :post, :url => location, :user => USDusername, :password => USDpassword, :headers => { :accept => :json, :content_type => :json}, :payload => json_data ).execute JSON.parse(response.to_str) end Creates a hash with ids mapping to names for an array of records def id_name_map(records) records.inject({}) do |map, record| map.update(record['id'] => record['name']) end end Get list of existing organizations orgs = get_json("#{katello_url}/organizations") org_list = id_name_map(orgs['results']) if !org_list.has_value?(org_name) # If our organization is not found, create it puts "Creating organization: \t#{org_name}" org_id = post_json("#{katello_url}/organizations", JSON.generate({"name"=> org_name}))["id"] else # Our organization exists, so let's grab it org_id = org_list.key(org_name) puts "Organization \"#{org_name}\" exists" end Get list of organization's lifecycle environments envs = get_json("#{katello_url}/organizations/#{org_id}/environments") env_list = id_name_map(envs['results']) prior_env_id = env_list.key("Library") Exit the script if at least one life cycle environment already exists environments.each do |e| if env_list.has_value?(e) puts "ERROR: One of the Environments is not unique to organization" exit end end # Create life cycle environments environments.each do |environment| puts "Creating environment: \t#{environment}" prior_env_id = post_json("#{katello_url}/organizations/#{org_id}/environments", JSON.generate({"name" => environment, "organization_id" => org_id, "prior_id" => prior_env_id}))["id"] end #!/usr/bin/tfm-ruby require 'apipie-bindings' org_name = "MyOrg" environments = [ "Development", "Testing", "Production" ] Create an instance of apipie bindings @api = ApipieBindings::API.new({ :uri => 'https://satellite.example.com/', :username => 'admin', :password => 'changeme', :api_version => 2 }) Performs an API call with default options def call_api(resource_name, action_name, params = {}) http_headers = {} apipie_options = { :skip_validation => true } @api.resource(resource_name).call(action_name, params, http_headers, apipie_options) end Creates a hash with IDs mapping to names for an array of records def id_name_map(records) records.inject({}) do |map, record| map.update(record['id'] => record['name']) end end Get list of existing organizations orgs = call_api(:organizations, :index) org_list = id_name_map(orgs['results']) if !org_list.has_value?(org_name) # If our organization is not found, create it puts "Creating organization: \t#{org_name}" org_id = call_api(:organizations, :create, {'organization' => { :name => org_name }})['id'] else # Our organization exists, so let's grab it org_id = org_list.key(org_name) puts "Organization \"#{org_name}\" exists" end Get list of organization's life cycle environments envs = call_api(:lifecycle_environments, :index, {'organization_id' => org_id}) env_list = id_name_map(envs['results']) prior_env_id = env_list.key("Library") Exit the script if at least one life cycle environment already exists environments.each do |e| if env_list.has_value?(e) puts "ERROR: One of the Environments is not unique to organization" exit end end # Create life cycle environments environments.each do |environment| puts "Creating environment: \t#{environment}" prior_env_id = call_api(:lifecycle_environments, :create, {"name" => environment, "organization_id" => org_id, "prior_id" => prior_env_id })['id'] end #!/usr/bin/python import json import sys try: import requests except ImportError: print "Please install the python-requests module." sys.exit(-1) URL to your Satellite 6 server URL = "https://satellite.example.com" URL for the API to your deployed Satellite 6 server SAT_API = "%s/katello/api/v2/" % URL Katello-specific API KATELLO_API = "%s/katello/api/" % URL POST_HEADERS = {'content-type': 'application/json'} Default credentials to login to Satellite 6 USERNAME = "admin" PASSWORD = "changeme" Ignore SSL for now SSL_VERIFY = False Name of the organization to be either created or used ORG_NAME = "MyOrg" Name for life cycle environments to be either created or used ENVIRONMENTS = ["Development", "Testing", "Production"] def get_json(location): """ Performs a GET using the passed URL location """ r = requests.get(location, auth=(USERNAME, PASSWORD), verify=SSL_VERIFY) return r.json() def post_json(location, json_data): """ Performs a POST and passes the data to the URL location """ result = requests.post( location, data=json_data, auth=(USERNAME, PASSWORD), verify=SSL_VERIFY, headers=POST_HEADERS) return result.json() def main(): """ Main routine that creates or re-uses an organization and life cycle environments. If life cycle environments already exist, exit out. """ # Check if our organization already exists org = get_json(SAT_API + "organizations/" + ORG_NAME) # If our organization is not found, create it if org.get('error', None): org_id = post_json( SAT_API + "organizations/", json.dumps({"name": ORG_NAME}))["id"] print "Creating organization: \t" + ORG_NAME else: # Our organization exists, so let's grab it org_id = org['id'] print "Organization '%s' exists." % ORG_NAME # Now, let's fetch all available life cycle environments for this org envs = get_json( SAT_API + "organizations/" + str(org_id) + "/environments/") # ... and add them to a dictionary, with respective 'Prior' environment prior_env_id = 0 env_list = {} for env in envs['results']: env_list[env['id']] = env['name'] prior_env_id = env['id'] if env['name'] == "Library" else prior_env_id # Exit the script if at least one life cycle environment already exists if all(environment in env_list.values() for environment in ENVIRONMENTS): print "ERROR: One of the Environments is not unique to organization" sys.exit(-1) # Create life cycle environments for environment in ENVIRONMENTS: new_env_id = post_json( SAT_API + "organizations/" + str(org_id) + "/environments/", json.dumps( { "name": environment, "organization_id": org_id, "prior": prior_env_id} ))["id"] print "Creating environment: \t" + environment prior_env_id = new_env_id if __name__ == "__main__": main() #!/usr/bin/python import json import sys try: import requests except ImportError: print "Please install the python-requests module." sys.exit(-1) SAT_API = 'https://satellite.example.com/api/v2/' USERNAME = "admin" PASSWORD = "password" SSL_VERIFY = False # Ignore SSL for now def get_json(url): # Performs a GET using the passed URL location r = requests.get(url, auth=(USERNAME, PASSWORD), verify=SSL_VERIFY) return r.json() def get_results(url): jsn = get_json(url) if jsn.get('error'): print "Error: " + jsn['error']['message'] else: if jsn.get('results'): return jsn['results'] elif 'results' not in jsn: return jsn else: print "No results found" return None def display_all_results(url): results = get_results(url) if results: print json.dumps(results, indent=4, sort_keys=True) def display_info_for_hosts(url): hosts = get_results(url) if hosts: for host in hosts: print "ID: %-10d Name: %-30s IP: %-20s OS: %-30s" % (host['id'], host['name'], host['ip'], host['operatingsystem_name']) def main(): host = 'satellite.example.com' print "Displaying all info for host %s ..." % host display_all_results(SAT_API + 'hosts/' + host) print "Displaying all facts for host %s ..." % host display_all_results(SAT_API + 'hosts/%s/facts' % host) host_pattern = 'example' print "Displaying basic info for hosts matching pattern '%s'..." % host_pattern display_info_for_hosts(SAT_API + 'hosts?search=' + host_pattern) environment = 'production' print "Displaying basic info for hosts in environment %s..." % environment display_info_for_hosts(SAT_API + 'hosts?search=environment=' + environment) model = 'RHEV Hypervisor' print "Displaying basic info for hosts with model name %s..." % model display_info_for_hosts(SAT_API + 'hosts?search=model="' + model + '"') if __name__ == "__main__": main() #!/usr/bin/env python3 import json import sys try: import requests except ImportError: print("Please install the python-requests module.") sys.exit(-1) SAT = "satellite.example.com" URL for the API to your deployed Satellite 6 server SAT_API = f"https://{SAT}/api/" KATELLO_API = f"https://{SAT}/katello/api/v2/" POST_HEADERS = {'content-type': 'application/json'} Default credentials to login to Satellite 6 USERNAME = "admin" PASSWORD = "password" Ignore SSL for now SSL_VERIFY = False #SSL_VERIFY = "./path/to/CA-certificate.crt" # Put the path to your CA certificate here to allow SSL_VERIFY def get_json(url): # Performs a GET using the passed URL location r = requests.get(url, auth=(USERNAME, PASSWORD), verify=SSL_VERIFY) return r.json() def get_results(url): jsn = get_json(url) if jsn.get('error'): print("Error: " + jsn['error']['message']) else: if jsn.get('results'): return jsn['results'] elif 'results' not in jsn: return jsn else: print("No results found") return None def display_all_results(url): results = get_results(url) if results: print(json.dumps(results, indent=4, sort_keys=True)) def display_info_for_hosts(url): hosts = get_results(url) if hosts: print(f"{'ID':10}{'Name':40}{'IP':30}{'Operating System':30}") for host in hosts: print(f"{str(host['id']):10}{host['name']:40}{str(host['ip']):30}{str(host['operatingsystem_name']):30}") def display_info_for_subs(url): subs = get_results(url) if subs: print(f"{'ID':10}{'Name':90}{'Start Date':30}") for sub in subs: print(f"{str(sub['id']):10}{sub['name']:90}{str(sub['start_date']):30}") def main(): host = SAT print(f"Displaying all info for host {host} ...") display_all_results(SAT_API + 'hosts/' + host) print(f"Displaying all facts for host {host} ...") display_all_results(SAT_API + f'hosts/{host}/facts') host_pattern = 'example' print(f"Displaying basic info for hosts matching pattern '{host_pattern}'...") display_info_for_hosts(SAT_API + 'hosts?per_page=1&search=name~' + host_pattern) print(f"Displaying basic info for subscriptions") display_info_for_subs(KATELLO_API + 'subscriptions') environment = 'production' print(f"Displaying basic info for hosts in environment {environment}...") display_info_for_hosts(SAT_API + 'hosts?search=environment=' + environment) if __name__ == "__main__": main()