4.3. Python を使用した API 要求
本セクションでは、Satellite API で Python を使用して、さまざまなタスクを実行する方法を説明します。
以下は、スクリプトおよびコマンドの例です。以下のスクリプトを慎重にレビューしてから使用するようにしてください。変数、ユーザー名、パスワード、その他の情報は、お使いのデプロイメントに適した値に置き換えてください。
本セクションのスクリプト例では、REST API との対話に SSL 検証を使用しません。
4.3.1. Python を使用したオブジェクトの作成
以下のスクリプトは Red Hat Satellite 6 API と接続して、組織を作成し、その新規組織内に 3 つの環境を作成します。組織がすでに存在する場合には、このスクリプトはその組織を使用します。組織内に環境が 1 つでも存在する場合は、スクリプトによりエラーが送出されて、スクリプトは終了します。
Python 2 の例
#!/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. Python を使用した API からの情報要求
以下は、さまざまな API 要求に Python を使用するスクリプト例です。
Python 2 の例
#!/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 の例
#!/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()